Annotate Network and check usage of it runtime

canopen/emcy.py

@@ -2,7 +2,10 @@
 import logging
 import threading
 import time
-from typing import Callable, List, Optional
+from typing import Callable, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from canopen.network import Network
 
 # Error code, error register, vendor specific data
 EMCY_STRUCT = struct.Struct("<HB5s")
@@ -82,14 +85,18 @@ def wait(
 class EmcyProducer:
 
     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""):
+        if self.network is None:
+            raise RuntimeError("A Network is required")
         payload = EMCY_STRUCT.pack(code, register, data)
         self.network.send_message(self.cob_id, payload)
 
     def reset(self, register: int = 0, data: bytes = b""):
+        if self.network is None:
+            raise RuntimeError("A Network is required to reset")
         payload = EMCY_STRUCT.pack(0, register, data)
         self.network.send_message(self.cob_id, payload)
 

canopen/lss.py

@@ -1,8 +1,14 @@
+from __future__ import annotations
+from typing import Optional, TYPE_CHECKING
 import logging
 import time
 import struct
 import queue
 
+if TYPE_CHECKING:
+    from canopen.network import Network
+
+
 logger = logging.getLogger(__name__)
 
 # Command Specifier (CS)
@@ -79,7 +85,7 @@ class LssMaster:
     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()
@@ -375,6 +381,8 @@ def __send_command(self, message):
             logger.info("There were unexpected messages in the queue")
             self.responses = queue.Queue()
 
+        if self.network is None:
+            raise RuntimeError("A Network is required to do send messages")
         self.network.send_message(self.LSS_TX_COBID, message)
 
         if not bool(message[0] in ListMessageNeedResponse):

canopen/network.py

@@ -350,7 +350,7 @@ class MessageListener(Listener):
     """
 
     def __init__(self, network: Network):
-        self.network = network
+        self.network: Network = network
 
     def on_message_received(self, msg):
         if msg.is_error_frame or msg.is_remote_frame:
@@ -385,7 +385,7 @@ class NodeScanner:
     SERVICES = (0x700, 0x580, 0x180, 0x280, 0x380, 0x480, 0x80)
 
     def __init__(self, network: Optional[Network] = None):
-        self.network = network
+        self.network: Optional[Network] = network
         #: A :class:`list` of nodes discovered
         self.nodes: List[int] = []
 

canopen/nmt.py

@@ -2,7 +2,10 @@
 import logging
 import struct
 import time
-from typing import Callable, Optional
+from typing import Callable, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from canopen.network import Network
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +48,7 @@ class NmtBase:
 
     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):
@@ -139,6 +142,8 @@ def send_command(self, code: int):
         super(NmtMaster, self).send_command(code)
         logger.info(
             "Sending NMT command 0x%X to node %d", code, self.id)
+        if self.network is None:
+            raise RuntimeError("A Network is required to send")
         self.network.send_message(0, [code, self.id])
 
     def wait_for_heartbeat(self, timeout: float = 10):
@@ -181,6 +186,8 @@ def start_node_guarding(self, period: float):
             Period (in seconds) at which the node guarding should be advertised to the slave node.
         """
         if self._node_guarding_producer : self.stop_node_guarding()
+        if self.network is None:
+            raise RuntimeError("A Network is required to send")
         self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, None, period, True)
 
     def stop_node_guarding(self):
@@ -216,6 +223,8 @@ def send_command(self, code: int) -> None:
 
         if self._state == 0:
             logger.info("Sending boot-up message")
+            if self.network is None:
+                raise RuntimeError("Cannot send without network")
             self.network.send_message(0x700 + self.id, [0])
 
         # The heartbeat service should start on the transition
@@ -246,6 +255,8 @@ def start_heartbeat(self, heartbeat_time_ms: int):
         self.stop_heartbeat()
         if heartbeat_time_ms > 0:
             logger.info("Start the heartbeat timer, interval is %d ms", self._heartbeat_time_ms)
+            if self.network is None:
+                raise RuntimeError("Cannot send without network")
             self._send_task = self.network.send_periodic(
                 0x700 + self.id, [self._state], heartbeat_time_ms / 1000.0)
 

canopen/node/base.py

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

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 canopen.node.base import BaseNode
 from canopen.sdo import SdoServer, SdoAbortedError
@@ -9,6 +10,9 @@
 from canopen.objectdictionary import ObjectDictionary
 from canopen import objectdictionary
 
+if TYPE_CHECKING:
+    from canopen.network import Network
+
 logger = logging.getLogger(__name__)
 
 
@@ -34,7 +38,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
@@ -45,8 +49,9 @@ def associate_network(self, network):
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self):
-        self.network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)
-        self.network.unsubscribe(0, self.nmt.on_command)
+        if self.network is not None:
+            self.network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)
+            self.network.unsubscribe(0, self.nmt.on_command)
         self.network = None
         self.sdo.network = None
         self.tpdo.network = None

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 canopen.sdo import SdoClient, SdoCommunicationError, SdoAbortedError
 from canopen.nmt import NmtMaster
@@ -8,6 +9,9 @@
 from canopen.objectdictionary import ODRecord, ODArray, ODVariable, ObjectDictionary
 from canopen.node.base import BaseNode
 
+if TYPE_CHECKING:
+    from canopen.network import Network
+
 logger = logging.getLogger(__name__)
 
 
@@ -46,7 +50,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
@@ -60,11 +64,12 @@ def associate_network(self, network):
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self):
-        for sdo in self.sdo_channels:
-            self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)
-        self.network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)
-        self.network.unsubscribe(0x80 + self.id, self.emcy.on_emcy)
-        self.network.unsubscribe(0, self.nmt.on_command)
+        if self.network is not None:
+            for sdo in self.sdo_channels:
+                self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)
+            self.network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)
+            self.network.unsubscribe(0x80 + self.id, self.emcy.on_emcy)
+            self.network.unsubscribe(0, self.nmt.on_command)
         self.network = None
         self.sdo.network = None
         self.pdo.network = None

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,9 @@
 from canopen.objectdictionary import ObjectDictionary, datatypes
 from canopen.sdo import SdoClient
 
+if TYPE_CHECKING:
+    from canopen.network import Network
+
 logger = logging.getLogger(__name__)
 
 # Object type. Don't confuse with Data type
@@ -172,7 +177,7 @@ def import_eds(source, node_id):
     return od
 
 
-def import_from_node(node_id, network):
+def import_from_node(node_id, network: Network):
     """ Download the configuration from the remote node
     :param int node_id: Identifier of the node
     :param network: network object

canopen/pdo/base.py

@@ -464,6 +464,8 @@ def subscribe(self) -> None:
         known to match what's stored on the node.
         """
         if self.enabled:
+            if self.pdo_node.network is None:
+               raise RuntimeError("A Network is required")
             logger.info("Subscribing to enabled PDO 0x%X on the network", self.cob_id)
             self.pdo_node.network.subscribe(self.cob_id, self.on_message)
 
@@ -511,6 +513,8 @@ def add_variable(
 
     def transmit(self) -> None:
         """Transmit the message once."""
+        if self.pdo_node.network is None:
+            raise RuntimeError("A Network is required")
         self.pdo_node.network.send_message(self.cob_id, self.data)
 
     def start(self, period: Optional[float] = None) -> None:
@@ -521,6 +525,9 @@ def start(self, period: Optional[float] = None) -> None:
             on the object before.
         :raises ValueError: When neither the argument nor the :attr:`period` is given.
         """
+        if self.pdo_node.network is None:
+            raise RuntimeError("A Network is required")
+
         # Stop an already running transmission if we have one, otherwise we
         # overwrite the reference and can lose our handle to shut it down
         self.stop()
@@ -551,6 +558,8 @@ def remote_request(self) -> None:
         Silently ignore if not allowed.
         """
         if self.enabled and self.rtr_allowed:
+            if self.pdo_node.network is None:
+                raise RuntimeError("A Network is required")
             self.pdo_node.network.send_message(self.cob_id, bytes(), remote=True)
 
     def wait_for_reception(self, timeout: float = 10) -> float:

canopen/sdo/base.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 import binascii
-from typing import Iterator, Optional, Union
+from typing import Iterator, Optional, Union, TYPE_CHECKING
 from collections.abc import Mapping
 
 from canopen import objectdictionary
 from canopen import variable
 from canopen.utils import pretty_index
 
+if TYPE_CHECKING:
+    from canopen.network import Network
+
 
 class CrcXmodem:
     """Mimics CrcXmodem from crccheck."""
@@ -43,7 +46,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/sdo/client.py

@@ -45,6 +45,8 @@ def on_response(self, can_id, data, timestamp):
         self.responses.put(bytes(data))
 
     def send_request(self, request):
+        if self.network is None:
+            raise RuntimeError("A Network is required to send a request")
         retries_left = self.MAX_RETRIES
         if self.PAUSE_BEFORE_SEND:
             time.sleep(self.PAUSE_BEFORE_SEND)

canopen/sdo/server.py

@@ -172,6 +172,8 @@ def segmented_download(self, command, request):
         self.send_response(response)
 
     def send_response(self, response):
+        if self.network is None:
+            raise RuntimeError("A Network is required to send")
         self.network.send_message(self.tx_cobid, response)
 
     def abort(self, abort_code=0x08000000):

canopen/sync.py

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

canopen/timestamp.py

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