Add a dummy object to designate "uninitialized" networks (fixes #511)

canopen/emcy.py

@@ -4,6 +4,9 @@
 import time
 from typing import Callable, List, Optional
 
+import canopen.network
+
+
 # Error code, error register, vendor specific data
 EMCY_STRUCT = struct.Struct("<HB5s")
 
@@ -82,7 +85,7 @@ def wait(
 class EmcyProducer:
 
     def __init__(self, cob_id: int):
-        self.network = None
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
         self.cob_id = cob_id
 
     def send(self, code: int, register: int = 0, data: bytes = b""):

canopen/lss.py

@@ -3,6 +3,9 @@
 import struct
 import queue
 
+import canopen.network
+
+
 logger = logging.getLogger(__name__)
 
 # Command Specifier (CS)
@@ -78,8 +81,8 @@ class LssMaster:
     #: Max time in seconds to wait for response from server
     RESPONSE_TIMEOUT = 0.5
 
-    def __init__(self):
-        self.network = None
+    def __init__(self) -> None:
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
         self._node_id = 0
         self._data = None
         self.responses = queue.Queue()

canopen/network.py

@@ -279,6 +279,17 @@ def __len__(self) -> int:
         return len(self.nodes)
 
 
+class _DummyNetwork(Network):
+    """Empty network implementation as a placeholder before actual initialization."""
+
+    def __init__(self, bus: Optional[can.BusABC] = None):
+        """Do not initialize attributes, by skipping the parent constructor."""
+
+    def __getattribute__(self, name):
+        raise RuntimeError("No actual Network object was assigned, "
+                           "try associating to a real network first.")
+
+
 class PeriodicMessageTask:
     """
     Task object to transmit a message periodically using python-can's

canopen/nmt.py

@@ -2,7 +2,13 @@
 import logging
 import struct
 import time
-from typing import Callable, Optional
+from typing import Callable, Optional, TYPE_CHECKING
+
+import canopen.network
+
+if TYPE_CHECKING:
+    from canopen.network import PeriodicMessageTask
+
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +51,7 @@ class NmtBase:
 
     def __init__(self, node_id: int):
         self.id = node_id
-        self.network = None
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
         self._state = 0
 
     def on_command(self, can_id, data, timestamp):

canopen/node/base.py

@@ -1,4 +1,6 @@
 from typing import TextIO, Union
+
+import canopen.network
 from canopen.objectdictionary import ObjectDictionary, import_od
 
 
@@ -17,10 +19,14 @@ def __init__(
         node_id: int,
         object_dictionary: Union[ObjectDictionary, str, TextIO],
     ):
-        self.network = None
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
 
         if not isinstance(object_dictionary, ObjectDictionary):
             object_dictionary = import_od(object_dictionary, node_id)
         self.object_dictionary = object_dictionary
 
         self.id = node_id or self.object_dictionary.node_id
+
+    def has_network(self) -> bool:
+        """Check whether the node has been associated to a network."""
+        return not isinstance(self.network, canopen.network._DummyNetwork)

canopen/node/local.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import logging
 from typing import Dict, Union
 
+import canopen.network
 from canopen.node.base import BaseNode
 from canopen.sdo import SdoServer, SdoAbortedError
 from canopen.pdo import PDO, TPDO, RPDO
@@ -34,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: canopen.network.Network):
         self.network = network
         self.sdo.network = network
         self.tpdo.network = network
@@ -44,15 +47,15 @@ def associate_network(self, network):
         network.subscribe(self.sdo.rx_cobid, self.sdo.on_request)
         network.subscribe(0, self.nmt.on_command)
 
-    def remove_network(self):
+    def remove_network(self) -> 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
-        self.rpdo.network = None
-        self.nmt.network = None
-        self.emcy.network = None
+        self.network = canopen.network._DummyNetwork()
+        self.sdo.network = self.network
+        self.tpdo.network = self.network
+        self.rpdo.network = self.network
+        self.nmt.network = self.network
+        self.emcy.network = self.network
 
     def add_read_callback(self, callback):
         self._read_callbacks.append(callback)

canopen/node/remote.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import logging
 from typing import Union, TextIO
 
+import canopen.network
 from canopen.sdo import SdoClient, SdoCommunicationError, SdoAbortedError
 from canopen.nmt import NmtMaster
 from canopen.emcy import EmcyConsumer
@@ -46,7 +49,7 @@ def __init__(
         if load_od:
             self.load_configuration()
 
-    def associate_network(self, network):
+    def associate_network(self, network: canopen.network.Network):
         self.network = network
         self.sdo.network = network
         self.pdo.network = network
@@ -59,18 +62,18 @@ def associate_network(self, network):
         network.subscribe(0x80 + self.id, self.emcy.on_emcy)
         network.subscribe(0, self.nmt.on_command)
 
-    def remove_network(self):
+    def remove_network(self) -> 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
-        self.tpdo.network = None
-        self.rpdo.network = None
-        self.nmt.network = None
+        self.network = canopen.network._DummyNetwork()
+        self.sdo.network = self.network
+        self.pdo.network = self.network
+        self.tpdo.network = self.network
+        self.rpdo.network = self.network
+        self.nmt.network = self.network
 
     def add_sdo(self, rx_cobid, tx_cobid):
         """Add an additional SDO channel.
@@ -87,7 +90,7 @@ def add_sdo(self, rx_cobid, tx_cobid):
         """
         client = SdoClient(rx_cobid, tx_cobid, self.object_dictionary)
         self.sdo_channels.append(client)
-        if self.network is not None:
+        if self.has_network():
             self.network.subscribe(client.tx_cobid, client.on_response)
         return client
 

canopen/pdo/base.py

@@ -6,12 +6,12 @@
 import logging
 import binascii
 
+import canopen.network
 from canopen.sdo import SdoAbortedError
 from canopen import objectdictionary
 from canopen import variable
 
 if TYPE_CHECKING:
-    from canopen.network import Network
     from canopen import LocalNode, RemoteNode
     from canopen.pdo import RPDO, TPDO
     from canopen.sdo import SdoRecord
@@ -30,7 +30,7 @@ class PdoBase(Mapping):
     """
 
     def __init__(self, node: Union[LocalNode, RemoteNode]):
-        self.network: Optional[Network] = None
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
         self.map: Optional[PdoMaps] = None
         self.node: Union[LocalNode, RemoteNode] = node
 

canopen/sdo/base.py

@@ -4,6 +4,7 @@
 from typing import Iterator, Optional, Union
 from collections.abc import Mapping
 
+import canopen.network
 from canopen import objectdictionary
 from canopen import variable
 from canopen.utils import pretty_index
@@ -43,7 +44,7 @@ def __init__(
         """
         self.rx_cobid = rx_cobid
         self.tx_cobid = tx_cobid
-        self.network = None
+        self.network: canopen.network.Network = canopen.network._DummyNetwork()
         self.od = od
 
     def __getitem__(