Add central typings to ease project navigation

Author: sveinse

canopen/emcy.py

@@ -1,8 +1,12 @@
+from __future__ import annotations
 import struct
 import logging
 import threading
 import time
-from typing import Callable, List, Optional
+from typing import Callable, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .network import Network
 
 # Error code, error register, vendor specific data
 EMCY_STRUCT = struct.Struct("<HB5s")
@@ -82,7 +86,7 @@ def wait(
 class EmcyProducer(object):
 
     def __init__(self, cob_id: int):
-        self.network = None
+        self.network: Optional[Network] = None
         self.cob_id = cob_id
 
     def send(self, code: int, register: int = 0, data: bytes = b""):

canopen/lss.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+from typing import Optional, TYPE_CHECKING
 import logging
 import time
 import struct
@@ -6,6 +8,9 @@
 except ImportError:
     import Queue as queue
 
+if TYPE_CHECKING:
+    from .network import Network
+
 logger = logging.getLogger(__name__)
 
 # Command Specifier (CS)
@@ -82,7 +87,7 @@ class LssMaster(object):
     RESPONSE_TIMEOUT = 0.5
 
     def __init__(self):
-        self.network = None
+        self.network: Optional[Network] = None
         self._node_id = 0
         self._data = None
         self.responses = queue.Queue()

canopen/network.py

@@ -1,10 +1,14 @@
+from __future__ import annotations
 try:
     from collections.abc import MutableMapping
 except ImportError:
     from collections import MutableMapping
 import logging
 import threading
-from typing import Callable, Dict, Iterable, List, Optional, Union
+from typing import Callable, Dict, Iterable, List, Optional, Union, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from can import BusABC, Notifier
 
 try:
     import can
@@ -39,13 +43,13 @@ def __init__(self, bus=None):
         """
         #: A python-can :class:`can.BusABC` instance which is set after
         #: :meth:`canopen.Network.connect` is called
-        self.bus = bus
+        self.bus: Optional[BusABC] = bus
         #: A :class:`~canopen.network.NodeScanner` for detecting nodes
         self.scanner = NodeScanner(self)
         #: List of :class:`can.Listener` objects.
         #: Includes at least MessageListener.
         self.listeners = [MessageListener(self)]
-        self.notifier = None
+        self.notifier: Optional[Notifier] = None
         self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
         self.subscribers: Dict[int, List[Callback]] = {}
         self.send_lock = threading.Lock()

canopen/nmt.py

@@ -1,11 +1,16 @@
+from __future__ import annotations
 import threading
 import logging
 import struct
 import time
-from typing import Callable, Optional
+from typing import Callable, Optional, TYPE_CHECKING
 
 from .network import CanError
 
+if TYPE_CHECKING:
+    from .network import Network
+
+
 logger = logging.getLogger(__name__)
 
 NMT_STATES = {
@@ -47,7 +52,7 @@ class NmtBase(object):
 
     def __init__(self, node_id: int):
         self.id = node_id
-        self.network = None
+        self.network: Optional[Network] = None
         self._state = 0
 
     def on_command(self, can_id, data, timestamp):

canopen/node/base.py

@@ -1,6 +1,10 @@
-from typing import TextIO, Union
+from __future__ import annotations
+from typing import TextIO, Union, Optional, TYPE_CHECKING
 from .. import objectdictionary
 
+if TYPE_CHECKING:
+    from ..network import Network
+
 
 class BaseNode(object):
     """A CANopen node.
@@ -17,7 +21,7 @@ def __init__(
         node_id: int,
         object_dictionary: Union[objectdictionary.ObjectDictionary, str, TextIO],
     ):
-        self.network = None
+        self.network: Optional[Network] = None
 
         if not isinstance(object_dictionary,
                           objectdictionary.ObjectDictionary):

canopen/node/local.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
 import logging
-from typing import Dict, Union
+from typing import Dict, Union, TYPE_CHECKING
 
 from .base import BaseNode
 from ..sdo import SdoServer, SdoAbortedError
@@ -8,6 +9,9 @@
 from ..emcy import EmcyProducer
 from .. import objectdictionary
 
+if TYPE_CHECKING:
+    from ..network import Network
+
 logger = logging.getLogger(__name__)
 
 
@@ -33,7 +37,7 @@ def __init__(
         self.add_write_callback(self.nmt.on_write)
         self.emcy = EmcyProducer(0x80 + self.id)
 
-    def associate_network(self, network):
+    def associate_network(self, network: Network):
         self.network = network
         self.sdo.network = network
         self.tpdo.network = network

canopen/node/remote.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
 import logging
-from typing import Union, TextIO
+from typing import Union, TextIO, TYPE_CHECKING
 
 from ..sdo import SdoClient
 from ..nmt import NmtMaster
@@ -12,6 +13,9 @@
 
 from canopen import objectdictionary
 
+if TYPE_CHECKING:
+    from ..network import Network
+
 logger = logging.getLogger(__name__)
 
 
@@ -50,7 +54,7 @@ def __init__(
         if load_od:
             self.load_configuration()
 
-    def associate_network(self, network):
+    def associate_network(self, network: Network):
         self.network = network
         self.sdo.network = network
         self.pdo.network = network

canopen/pdo/base.py

@@ -1,13 +1,17 @@
+from __future__ import annotations
 import threading
 import math
-from typing import Callable, Dict, Iterable, List, Optional, Union
+from typing import Callable, Dict, Iterable, List, Optional, Union, TYPE_CHECKING
 try:
     from collections.abc import Mapping
 except ImportError:
     from collections import Mapping
 import logging
 import binascii
 
+if TYPE_CHECKING:
+    from ..network import Network
+
 from ..sdo import SdoAbortedError
 from .. import objectdictionary
 from .. import variable
@@ -26,14 +30,14 @@ class PdoBase(Mapping):
     """
 
     def __init__(self, node):
-        self.network = None
+        self.network: Optional[Network] = None
         self.map = None  # instance of Maps
         self.node = node
 
     def __iter__(self):
         return iter(self.map)
 
-    def __getitem__(self, key):
+    def __getitem__(self, key) -> PdoBase:
         if isinstance(key, int) and (0x1A00 <= key <= 0x1BFF or   # By TPDO ID (512)
                                      0x1600 <= key <= 0x17FF or   # By RPDO ID (512)
                                      0 < key <= 512):             # By PDO Index
@@ -160,7 +164,7 @@ def __len__(self) -> int:
 class Map(object):
     """One message which can have up to 8 bytes of variables mapped."""
 
-    def __init__(self, pdo_node, com_record, map_array):
+    def __init__(self, pdo_node: PdoBase, com_record, map_array):
         self.pdo_node = pdo_node
         self.com_record = com_record
         self.map_array = map_array

canopen/sdo/base.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
 import binascii
-from typing import Iterable, Union
+from typing import Iterable, Union, Optional, TYPE_CHECKING
 try:
     from collections.abc import Mapping
 except ImportError:
@@ -8,6 +9,9 @@
 from .. import objectdictionary
 from .. import variable
 
+if TYPE_CHECKING:
+    from ..network import Network
+
 
 class CrcXmodem(object):
     """Mimics CrcXmodem from crccheck."""
@@ -43,7 +47,7 @@ def __init__(
         """
         self.rx_cobid = rx_cobid
         self.tx_cobid = tx_cobid
-        self.network = None
+        self.network: Optional[Network] = None
         self.od = od
 
     def __getitem__(

canopen/sync.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
 
+from typing import Optional, TYPE_CHECKING
 
-from typing import Optional
+if TYPE_CHECKING:
+    from .network import Network
 
 
 class SyncProducer(object):
@@ -9,7 +12,7 @@ class SyncProducer(object):
     #: COB-ID of the SYNC message
     cob_id = 0x80
 
-    def __init__(self, network):
+    def __init__(self, network: Network):
         self.network = network
         self.period: Optional[float] = None
         self._task = None