Type annotate everywhere Network is used (#575)

Co-authored-by: André Colomb <src@andre.colomb.de>

Author: sveinse

canopen/network.py

@@ -392,7 +392,9 @@ class NodeScanner:
     SERVICES = (0x700, 0x580, 0x180, 0x280, 0x380, 0x480, 0x80)
 
     def __init__(self, network: Optional[Network] = None):
-        self.network = network
+        if network is None:
+            network = _UNINITIALIZED_NETWORK
+        self.network: Network = network
         #: A :class:`list` of nodes discovered
         self.nodes: List[int] = []
 
@@ -408,8 +410,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,12 +1,18 @@
+from __future__ import annotations
+
 import copy
 import logging
 import re
 from configparser import NoOptionError, NoSectionError, RawConfigParser
+from typing import TYPE_CHECKING
 
 from canopen import objectdictionary
 from canopen.objectdictionary import ObjectDictionary, datatypes
 from canopen.sdo import SdoClient
 
+if TYPE_CHECKING:
+    import canopen.network
+
 
 logger = logging.getLogger(__name__)
 
@@ -173,7 +179,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,4 +1,9 @@
-from typing import Optional
+from __future__ import annotations
+
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import canopen.network
 
 
 class SyncProducer:
@@ -7,7 +12,7 @@ class SyncProducer:
     #: 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,6 +1,11 @@
+from __future__ import annotations
+
 import struct
 import time
-from typing import Optional
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import canopen.network
 
 
 # 1 Jan 1984
@@ -17,7 +22,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):