From 531c0b50b91e4ddf71dfb851b3032ab91094d1f5 Mon Sep 17 00:00:00 2001 From: David Vo Date: Sun, 16 Jan 2022 15:50:27 +1100 Subject: [PATCH] Add more type hints --- _networktables_types.py | 11 +++++ _pynetworktables/_impl/callback_manager.py | 6 +-- _pynetworktables/_impl/connection_notifier.py | 4 +- _pynetworktables/_impl/entry_notifier.py | 4 +- _pynetworktables/_impl/rpc_server.py | 4 +- _pynetworktables/_impl/support/safe_thread.py | 8 ++-- _pynetworktables/_impl/value.py | 5 ++- _pynetworktables/entry.py | 22 +++++----- _pynetworktables/instance.py | 42 +++++++++---------- _pynetworktables/table.py | 28 +++++-------- _pynetworktables/types.py | 1 + networktables/types.py | 1 + networktables/util.py | 4 +- setup.py | 1 + 14 files changed, 73 insertions(+), 68 deletions(-) create mode 100644 _networktables_types.py create mode 100644 _pynetworktables/types.py create mode 100644 networktables/types.py diff --git a/_networktables_types.py b/_networktables_types.py new file mode 100644 index 0000000..1753e55 --- /dev/null +++ b/_networktables_types.py @@ -0,0 +1,11 @@ +from typing import Sequence, Union + +ValueT = Union[ + bool, + float, + str, + bytes, + Sequence[bool], + Sequence[float], + Sequence[str], +] diff --git a/_pynetworktables/_impl/callback_manager.py b/_pynetworktables/_impl/callback_manager.py index eeb81b7..41f398a 100644 --- a/_pynetworktables/_impl/callback_manager.py +++ b/_pynetworktables/_impl/callback_manager.py @@ -12,9 +12,9 @@ try: # Python 3.7 only, should be more efficient - from queue import SimpleQueue as Queue, Empty + from queue import SimpleQueue as Queue except ImportError: - from queue import Queue, Empty + from queue import Queue from .support.safe_thread import SafeThread from .support.uidvector import UidVector @@ -23,7 +23,7 @@ logger = logging.getLogger("nt") -_ListenerData = namedtuple("ListenerData", ["callback", "poller_uid"]) +_ListenerData = namedtuple("_ListenerData", ["callback", "poller_uid"]) class Poller(object): diff --git a/_pynetworktables/_impl/connection_notifier.py b/_pynetworktables/_impl/connection_notifier.py index c8a4de8..b307cad 100644 --- a/_pynetworktables/_impl/connection_notifier.py +++ b/_pynetworktables/_impl/connection_notifier.py @@ -10,10 +10,10 @@ from .callback_manager import CallbackManager, CallbackThread -_ConnectionCallback = namedtuple("ConnectionCallback", ["callback", "poller_uid"]) +_ConnectionCallback = namedtuple("_ConnectionCallback", ["callback", "poller_uid"]) _ConnectionNotification = namedtuple( - "ConnectionNotification", ["connected", "conn_info"] + "_ConnectionNotification", ["connected", "conn_info"] ) diff --git a/_pynetworktables/_impl/entry_notifier.py b/_pynetworktables/_impl/entry_notifier.py index dd20c3a..936c405 100644 --- a/_pynetworktables/_impl/entry_notifier.py +++ b/_pynetworktables/_impl/entry_notifier.py @@ -19,7 +19,7 @@ _EntryListenerData = namedtuple( - "EntryListenerData", + "_EntryListenerData", [ "prefix", "local_id", # we don't have entry handles like ntcore has @@ -31,7 +31,7 @@ # _EntryNotification = namedtuple( - "EntryNotification", ["name", "value", "flags", "local_id"] + "_EntryNotification", ["name", "value", "flags", "local_id"] ) _assign_both = NT_NOTIFY_UPDATE | NT_NOTIFY_FLAGS diff --git a/_pynetworktables/_impl/rpc_server.py b/_pynetworktables/_impl/rpc_server.py index 0b58092..038acd0 100644 --- a/_pynetworktables/_impl/rpc_server.py +++ b/_pynetworktables/_impl/rpc_server.py @@ -15,10 +15,10 @@ logger = logging.getLogger("nt") -_RpcListenerData = namedtuple("RpcListenerData", ["callback", "poller_uid"]) +_RpcListenerData = namedtuple("_RpcListenerData", ["callback", "poller_uid"]) _RpcCall = namedtuple( - "RpcCall", ["local_id", "call_uid", "name", "params", "conn_info", "send_response"] + "_RpcCall", ["local_id", "call_uid", "name", "params", "conn_info", "send_response"] ) diff --git a/_pynetworktables/_impl/support/safe_thread.py b/_pynetworktables/_impl/support/safe_thread.py index 147cbf1..c412b82 100644 --- a/_pynetworktables/_impl/support/safe_thread.py +++ b/_pynetworktables/_impl/support/safe_thread.py @@ -1,6 +1,6 @@ -import threading - import logging +import threading +from typing import Callable, Dict logger = logging.getLogger("nt.th") @@ -13,9 +13,9 @@ class SafeThread(object): # Name each thread uniquely to make debugging easier _global_indices_lock = threading.Lock() - _global_indices = {} + _global_indices = {} # type: Dict[str, int] - def __init__(self, target, name, args=()): + def __init__(self, target: Callable, name: str, args=()): """ Note: thread is automatically started and daemonized """ diff --git a/_pynetworktables/_impl/value.py b/_pynetworktables/_impl/value.py index e8e8ca7..d5e3758 100644 --- a/_pynetworktables/_impl/value.py +++ b/_pynetworktables/_impl/value.py @@ -11,7 +11,7 @@ more efficient. """ -from collections import namedtuple +from typing import NamedTuple from .constants import ( NT_BOOLEAN, NT_DOUBLE, @@ -22,9 +22,10 @@ NT_STRING_ARRAY, NT_RPC, ) +from ..types import ValueT -class Value(namedtuple("Value", ["type", "value"])): +class Value(NamedTuple("Value", [("type", bytes), ("value", ValueT)])): __slots__ = () @classmethod diff --git a/_pynetworktables/entry.py b/_pynetworktables/entry.py index 9745496..bbe60ca 100644 --- a/_pynetworktables/entry.py +++ b/_pynetworktables/entry.py @@ -10,9 +10,10 @@ NT_STRING_ARRAY, NT_PERSISTENT, ) - from ._impl.value import Value +from .types import ValueT + __all__ = ["NetworkTableEntry"] D = TypeVar("D") @@ -171,15 +172,14 @@ def getStringArray(self, defaultValue: D) -> Union[Sequence[str], D]: :param defaultValue: the value to be returned if no value is found :returns: the entry's value or the given default value - :rtype: list(float) """ value = self._value if not value or value[0] != NT_STRING_ARRAY: return defaultValue return value[1] - @classmethod - def isValidDataType(cls, data): + @staticmethod + def isValidDataType(data) -> bool: if isinstance(data, (bytes, bytearray)): return True if isinstance(data, (list, tuple)): @@ -187,9 +187,9 @@ def isValidDataType(cls, data): raise ValueError("If you use a list here, cannot be empty") data = data[0] - return isinstance(data, (int, float, str, bool)) + return isinstance(data, (int, float, str)) - def setDefaultValue(self, defaultValue) -> bool: + def setDefaultValue(self, defaultValue: ValueT) -> bool: """Sets the entry's value if it does not exist. :param defaultValue: the default value to set @@ -267,7 +267,7 @@ def setDefaultStringArray(self, defaultValue: Sequence[str]) -> bool: value = Value.makeStringArray(defaultValue) return self.__api.setDefaultEntryValueById(self._local_id, value) - def setValue(self, value) -> bool: + def setValue(self, value: ValueT) -> bool: """Sets the entry's value :param value: the value that will be assigned @@ -345,7 +345,7 @@ def setStringArray(self, value: Sequence[str]) -> bool: value = Value.makeStringArray(value) return self.__api.setEntryValueById(self._local_id, value) - def forceSetValue(self, value): + def forceSetValue(self, value: ValueT): """Sets the entry's value :param value: the value that will be assigned @@ -456,7 +456,7 @@ def delete(self) -> bool: def addListener( self, - listener: Callable[["NetworkTableEntry", str, Any, int], None], + listener: Callable[["NetworkTableEntry", str, ValueT, int], None], flags: int, paramIsNew: bool = True, ): @@ -521,5 +521,5 @@ def __bool__(self): "< not allowed on NetworkTableEntry objects. Use the .value attribute instead" ) - def __repr__(self): - return "" % (self._value.__repr__(),) + def __repr__(self) -> str: + return "" % (self._value,) diff --git a/_pynetworktables/instance.py b/_pynetworktables/instance.py index 59d39f1..6668b27 100644 --- a/_pynetworktables/instance.py +++ b/_pynetworktables/instance.py @@ -1,6 +1,8 @@ # todo: tracks NetworkTablesInstance.java -from typing import Any, Callable, List, Optional, Sequence, Tuple, Union +import logging +import typing +from typing import Callable, List, Optional, Sequence, Tuple, Union from weakref import WeakSet from ._impl import constants @@ -8,8 +10,10 @@ from .entry import NetworkTableEntry from .table import NetworkTable +from .types import ValueT -import logging +if typing.TYPE_CHECKING: + from networktables.util import _NtProperty logger = logging.getLogger("nt") @@ -48,7 +52,7 @@ class NetworkTablesInstance: instances is for unit testing, but they can also enable one program to connect to two different NetworkTables networks. - The global "default" instance (as returned by :meth:`.NetworkTablesInstance.getDefault`) is + The global "default" instance (as returned by :meth:`.getDefault`) is always available, and is intended for the common case when there is only a single NetworkTables instance being used in the program. @@ -164,17 +168,16 @@ def getDefault(cls) -> "NetworkTablesInstance": cls._defaultInstance = cls() return cls._defaultInstance - def __init__(self): + def __init__(self) -> None: self._init() - def _init(self): + def _init(self) -> None: self._api = NtCoreApi(self.__createEntry) - self._tables = {} - self._entry_listeners = {} - self._conn_listeners = {} + self._tables = {} # type: typing.Dict[str, NetworkTable] + self._conn_listeners = {} # type: typing.Dict[Callable, List[int]] if not hasattr(self, "_ntproperties"): - self._ntproperties = WeakSet() + self._ntproperties = WeakSet() # type: WeakSet[_NtProperty] else: for ntprop in self._ntproperties: ntprop.reset() @@ -202,7 +205,6 @@ def getEntries(self, prefix: str, types: int = 0) -> Sequence[NetworkTableEntry] starts with this string are returned :param types: bitmask of types; 0 is treated as a "don't care" :returns: List of matching entries. - :rtype: list of :class:`.NetworkTableEntry` .. versionadded:: 2018.0.0 """ @@ -263,7 +265,7 @@ def deleteAllEntries(self) -> None: def addEntryListener( self, - listener: Callable[[str, Any, int], None], + listener: Callable[[str, ValueT, int], None], immediateNotify: bool = True, localNotify: bool = True, paramIsNew: bool = True, @@ -306,7 +308,7 @@ def addEntryListener( def addEntryListenerEx( self, - listener: Callable[[str, Any, int], None], + listener: Callable[[str, ValueT, int], None], flags: int, paramIsNew: bool = True, ) -> None: @@ -341,7 +343,7 @@ def addEntryListenerEx( addGlobalListener = addEntryListener addGlobalListenerEx = addEntryListenerEx - def removeEntryListener(self, listener: Callable[[str, Any, int], None]) -> None: + def removeEntryListener(self, listener: Callable[[str, ValueT, int], None]) -> None: """Remove an entry listener. :param listener: Listener to remove @@ -445,7 +447,7 @@ def startServer( persistFilename: str = "networktables.ini", listenAddress: str = "", port: int = constants.NT_DEFAULT_PORT, - ): + ) -> bool: """Starts a server using the specified filename, listening address, and port. :param persistFilename: the name of the persist file to use @@ -467,7 +469,7 @@ def stopServer(self) -> None: def startClient( self, server_or_servers: Union[str, ServerPortPair, List[ServerPortPair], List[str]], - ): + ) -> bool: """Sets server addresses and port for client (without restarting client). The client will attempt to connect to each server in round robin fashion. @@ -479,7 +481,7 @@ def startClient( self.setServer(server_or_servers) return self._api.startClient() - def startClientTeam(self, team: int, port: int = constants.NT_DEFAULT_PORT): + def startClientTeam(self, team: int, port: int = constants.NT_DEFAULT_PORT) -> bool: """Starts a client using commonly known robot addresses for the specified team. @@ -580,7 +582,6 @@ def getConnections(self) -> Sequence: If operating as a client, this will return either zero or one values. :returns: list of connection information - :rtype: list .. versionadded:: 2018.0.0 """ @@ -657,7 +658,7 @@ def loadEntries(self, filename: str, prefix: str): # These methods are unique to pynetworktables # - def initialize(self, server=None): + def initialize(self, server: Optional[str] = None) -> bool: """Initializes NetworkTables and begins operations :param server: If specified, NetworkTables will be set to client @@ -665,7 +666,6 @@ def initialize(self, server=None): This is equivalent to executing:: self.startClient(server) - :type server: str :returns: True if initialized, False if already initialized @@ -691,7 +691,7 @@ def shutdown(self) -> None: self._init() - def startTestMode(self, server: bool = True): + def startTestMode(self, server: bool = True) -> bool: """Setup network tables to run in unit test mode, and enables verbose logging. @@ -736,8 +736,6 @@ def getGlobalAutoUpdateValue( :param defaultValue: The default value to return if the key doesn't exist :param writeDefault: If True, force the value to the specified default - :rtype: :class:`.NetworkTableEntry` - .. seealso:: :func:`.ntproperty` is a read-write alternative to this .. versionadded:: 2015.3.0 diff --git a/_pynetworktables/table.py b/_pynetworktables/table.py index 96c1028..d2ec522 100644 --- a/_pynetworktables/table.py +++ b/_pynetworktables/table.py @@ -1,6 +1,6 @@ __all__ = ["NetworkTable"] -from typing import Callable, List, Optional, Sequence +from typing import Callable, Dict, List, Optional, Sequence, TypeVar, Union from ._impl.constants import ( NT_BOOLEAN, @@ -22,12 +22,15 @@ from ._impl.value import Value from .entry import NetworkTableEntry +from .types import ValueT import logging logger = logging.getLogger("nt") _is_new = NT_NOTIFY_IMMEDIATE | NT_NOTIFY_NEW +D = TypeVar("D") + class NetworkTable: """ @@ -74,8 +77,6 @@ def getEntry(self, key: str) -> NetworkTableEntry: """Gets the entry for a subkey. This is the preferred API to use to access NetworkTable keys. - :rtype: :class:`.NetworkTableEntry` - .. versionadded:: 2018.0.0 """ return self._inst.getEntry(self._path + key) @@ -218,7 +219,7 @@ def addSubTableListener( .. versionchanged:: 2017.0.0 Added localNotify parameter """ - notified_tables = {} + notified_tables = {} # type: Dict[str, object] def _callback(item): key, value_, _1, _2 = item @@ -416,7 +417,6 @@ def setDefaultNumber(self, key: str, defaultValue: float) -> bool: :param key: the key to be assigned to :param defaultValue: the default value to set if key doesn't exist. - :type defaultValue: int, float :returns: False if the table key exists with a different type @@ -466,7 +466,7 @@ def setDefaultString(self, key: str, defaultValue: str) -> bool: path = self._path + key return self._api.setDefaultEntryValue(path, Value.makeString(defaultValue)) - def getString(self, key: str, defaultValue: str) -> str: + def getString(self, key: str, defaultValue: D) -> Union[str, D]: """Gets the string associated with the given name. If the key does not exist or is of different type, it will return the default value. @@ -558,7 +558,6 @@ def getBooleanArray(self, key: str, defaultValue) -> Sequence[bool]: of different type, it will return the default value. :param key: the key to look up - :type key: str :param defaultValue: the value to be returned if no value is found :returns: the value associated with the given key or the given default value @@ -711,7 +710,7 @@ def getRaw(self, key: str, defaultValue: bytes) -> bytes: return value.value - def putValue(self, key: str, value) -> bool: + def putValue(self, key: str, value: ValueT) -> bool: """Put a value in the table, trying to autodetect the NT type of the value. Refer to this table to determine the type mapping: @@ -729,7 +728,6 @@ def putValue(self, key: str, value) -> bool: :param key: the key to be assigned to :param value: the value that will be assigned - :type value: bool, int, float, str, bytes :returns: False if the table key already exists with a different type @@ -739,13 +737,12 @@ def putValue(self, key: str, value) -> bool: path = self._path + key return self._api.setEntryValue(path, value) - def setDefaultValue(self, key: str, defaultValue) -> bool: + def setDefaultValue(self, key: str, defaultValue: ValueT) -> bool: """If the key doesn't currently exist, then the specified value will be assigned to the key. :param key: the key to be assigned to :param defaultValue: the default value to set if key doesn't exist. - :type defaultValue: bool, int, float, str, bytes :returns: False if the table key exists with a different type @@ -757,16 +754,14 @@ def setDefaultValue(self, key: str, defaultValue) -> bool: path = self._path + key return self._api.setDefaultEntryValue(path, defaultValue) - def getValue(self, key: str, defaultValue): + def getValue(self, key: str, defaultValue: D) -> Union[ValueT, D]: """Gets the value associated with a key. This supports all NetworkTables types (unlike :meth:`putValue`). :param key: the key of the value to look up :param defaultValue: The default value to return if the key doesn't exist - :type defaultValue: any :returns: the value associated with the given key - :rtype: bool, int, float, str, bytes, tuple .. versionadded:: 2017.0.0 """ @@ -778,19 +773,16 @@ def getValue(self, key: str, defaultValue): return value.value def getAutoUpdateValue( - self, key: str, defaultValue, writeDefault: bool = True + self, key: str, defaultValue: ValueT, writeDefault: bool = True ) -> NetworkTableEntry: """Returns an object that will be automatically updated when the value is updated by networktables. :param key: the key name :param defaultValue: Default value to use if not in the table - :type defaultValue: any :param writeDefault: If True, put the default value to the table, overwriting existing values - :rtype: :class:`.NetworkTableEntry` - .. note:: If you modify the returned value, the value will NOT be written back to NetworkTables (though now there are functions you can use to write values). See :func:`.ntproperty` if diff --git a/_pynetworktables/types.py b/_pynetworktables/types.py new file mode 100644 index 0000000..457732f --- /dev/null +++ b/_pynetworktables/types.py @@ -0,0 +1 @@ +from _networktables_types import ValueT diff --git a/networktables/types.py b/networktables/types.py new file mode 100644 index 0000000..457732f --- /dev/null +++ b/networktables/types.py @@ -0,0 +1 @@ +from _networktables_types import ValueT diff --git a/networktables/util.py b/networktables/util.py index b649f9c..c4b121a 100644 --- a/networktables/util.py +++ b/networktables/util.py @@ -1,5 +1,6 @@ from typing import Callable, Optional, Sequence +from .types import ValueT from . import NetworkTablesInstance __all__ = ["ntproperty", "ChooserControl"] @@ -58,7 +59,7 @@ def _set_pyntcore(self, _, value): def ntproperty( key: str, - defaultValue, + defaultValue: ValueT, writeDefault: bool = True, doc: str = None, persistent: bool = False, @@ -71,7 +72,6 @@ def ntproperty( :param key: A full NetworkTables key (eg ``/SmartDashboard/foo``) :param defaultValue: Default value to use if not in the table - :type defaultValue: any :param writeDefault: If True, put the default value to the table, overwriting existing values :param doc: If given, will be the docstring of the property. diff --git a/setup.py b/setup.py index 916eba9..52a171d 100644 --- a/setup.py +++ b/setup.py @@ -48,6 +48,7 @@ url="https://github.com/robotpy/pynetworktables", keywords="frc first robotics wpilib networktables", packages=find_packages(exclude="tests"), + py_modules=["_networktables_types"], package_data={ "networktables": ["py.typed", "__init__.pyi"], "_pynetworktables": ["py.typed"],