Add more type hints

_networktables_types.py

@@ -0,0 +1,11 @@
+from typing import Sequence, Union
+
+ValueT = Union[
+    bool,
+    float,
+    str,
+    bytes,
+    Sequence[bool],
+    Sequence[float],
+    Sequence[str],
+]

_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):

_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"]
 )
 
 

_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

_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"]
 )
 
 

_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
         """

_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

_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,25 +172,24 @@ 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)):
             if len(data) == 0:
                 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 "<NetworkTableEntry: %s>" % (self._value.__repr__(),)
+    def __repr__(self) -> str:
+        return "<NetworkTableEntry: %r>" % (self._value,)

_pynetworktables/instance.py

@@ -1,15 +1,19 @@
 # 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
 from ._impl.api import NtCoreApi
 
 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,15 +658,14 @@ 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
                        mode and attempt to connect to the specified server.
                        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