Type annotate everywhere Network is used

canopen/network.py

@@ -391,8 +391,11 @@ class NodeScanner:
 
     SERVICES = (0x700, 0x580, 0x180, 0x280, 0x380, 0x480, 0x80)
 
-    def __init__(self, network: Optional[Network] = None):
-        self.network = network
+    def __init__(self, network: Optional[Network] = _UNINITIALIZED_NETWORK):
+        # To preseve the old API where None is uninitialized network
+        if network is None:
+            network = _UNINITIALIZED_NETWORK
+        self.network: Network = network
         #: A :class:`list` of nodes discovered
         self.nodes: List[int] = []
 
@@ -408,8 +411,6 @@ def reset(self):
 
     def search(self, limit: int = 127) -> None:
         """Search for nodes by sending SDO requests to all node IDs."""
-        if self.network is None:
-            raise RuntimeError("A Network is required to do active scanning")
         sdo_req = b"\x40\x00\x10\x00\x00\x00\x00\x00"
         for node_id in range(1, limit + 1):
             self.network.send_message(0x600 + node_id, sdo_req)

canopen/objectdictionary/eds.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
 import copy
 import logging
 import re
@@ -7,6 +9,8 @@
 from canopen.objectdictionary import ObjectDictionary, datatypes
 from canopen.sdo import SdoClient
 
+if TYPE_CHECKING:
+    import canopen.network
 
 logger = logging.getLogger(__name__)
 
@@ -173,7 +177,7 @@ def import_eds(source, node_id):
     return od
 
 
-def import_from_node(node_id, network):
+def import_from_node(node_id: int, network: canopen.network.Network):
     """ Download the configuration from the remote node
     :param int node_id: Identifier of the node
     :param network: network object

canopen/sync.py

@@ -1,13 +1,17 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
 from typing import Optional
 
+if TYPE_CHECKING:
+    import canopen.network
 
 class SyncProducer:
     """Transmits a SYNC message periodically."""
 
     #: COB-ID of the SYNC message
     cob_id = 0x80
 
-    def __init__(self, network):
+    def __init__(self, network: canopen.network.Network):
         self.network = network
         self.period: Optional[float] = None
         self._task = None

canopen/timestamp.py

@@ -1,7 +1,11 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
 import struct
 import time
 from typing import Optional
 
+if TYPE_CHECKING:
+    import canopen.network
 
 # 1 Jan 1984
 OFFSET = 441763200
@@ -17,7 +21,7 @@ class TimeProducer:
     #: COB-ID of the SYNC message
     cob_id = 0x100
 
-    def __init__(self, network):
+    def __init__(self, network: canopen.network.Network):
         self.network = network
 
     def transmit(self, timestamp: Optional[float] = None):

test/test_network.py

@@ -318,7 +318,7 @@ def test_scanner_reset(self):
         self.assertListEqual(self.scanner.nodes, [])
 
     def test_scanner_search_no_network(self):
-        with self.assertRaisesRegex(RuntimeError, "Network is required"):
+        with self.assertRaisesRegex(RuntimeError, "No actual Network object was assigned"):
             self.scanner.search()
 
     def test_scanner_search(self):