Synchronization with main, removing supersceded code

Author: sveinse

canopen/emcy.py

@@ -90,14 +90,10 @@ def __init__(self, cob_id: int):
         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")
         payload = EMCY_STRUCT.pack(0, register, data)
         self.network.send_message(self.cob_id, payload)
 

canopen/network.py

@@ -3,8 +3,7 @@
 import logging
 import threading
 from collections.abc import MutableMapping
-from typing import Callable, Dict, Final, Iterator, List, Optional, Union
-from typing import Callable, Dict, Iterator, List, Optional, Union, TYPE_CHECKING, TextIO
+from typing import Callable, Dict, Final, Iterator, List, Optional, Union, TYPE_CHECKING, TextIO
 
 import can
 from can import Listener
@@ -168,7 +167,7 @@ def add_node(
 
     def create_node(
         self,
-        node: int,
+        node: Union[int, LocalNode],
         object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None,
     ) -> LocalNode:
         """Create a local node in the network.
@@ -234,6 +233,8 @@ def send_periodic(
         :return:
             An task object with a ``.stop()`` method to stop the transmission
         """
+        if not self.bus:
+            raise RuntimeError("Not connected to CAN bus")
         return PeriodicMessageTask(can_id, data, period, self.bus, remote)
 
     def notify(self, can_id: int, data: bytearray, timestamp: float) -> None:
@@ -315,7 +316,7 @@ def __init__(
         can_id: int,
         data: CanData,
         period: float,
-        bus,
+        bus: can.BusABC,
         remote: bool = False,
     ):
         """
@@ -352,12 +353,10 @@ def update(self, data: bytes) -> None:
         old_data = self.msg.data
         self.msg.data = new_data
         if hasattr(self._task, "modify_data"):
-            assert self._task is not None  # This will never be None, but mypy needs this
             self._task.modify_data(self.msg)
         elif new_data != old_data:
             # Stop and start (will mess up period unfortunately)
-            if self._task is not None:
-                self._task.stop()
+            self._task.stop()
             self._start()
 
 

canopen/nmt.py

@@ -145,8 +145,6 @@ 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")
         self.network.send_message(0, [code, self.id])
 
     def wait_for_heartbeat(self, timeout: float = 10):
@@ -189,8 +187,6 @@ 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")
         self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, [], period, True)
 
     def stop_node_guarding(self):
@@ -226,8 +222,6 @@ def send_command(self, code: int) -> None:
 
         if self._state == 0:
             logger.info("Sending boot-up message")
-            if self.network is None:
-                raise RuntimeError("A Network is required")
             self.network.send_message(0x700 + self.id, [0])
 
         # The heartbeat service should start on the transition
@@ -258,8 +252,6 @@ 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("A network is required")
             self._send_task = self.network.send_periodic(
                 0x700 + self.id, [self._state], heartbeat_time_ms / 1000.0)
 

canopen/pdo/base.py

@@ -466,8 +466,6 @@ 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")
             if self.cob_id is None:
                raise RuntimeError("A valid COB-ID is required")
             logger.info("Subscribing to enabled PDO 0x%X on the network", self.cob_id)
@@ -517,8 +515,6 @@ def add_variable(
 
     def transmit(self) -> None:
         """Transmit the message once."""
-        if self.pdo_node.network is None:
-            raise RuntimeError("A Network is required")
         if self.cob_id is None:
             raise RuntimeError("A valid COB-ID is required")
         self.pdo_node.network.send_message(self.cob_id, self.data)
@@ -531,8 +527,6 @@ 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")
         if self.cob_id is None:
             raise RuntimeError("A valid COB-ID is required")
 
@@ -566,8 +560,6 @@ 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")
             if self.cob_id is None:
                 raise RuntimeError("A valid COB-ID is required")
             self.pdo_node.network.send_message(self.cob_id, bytes(), remote=True)

pyproject.toml

@@ -56,4 +56,3 @@ ignore_missing_imports = "True"
 disable_error_code = [
     "annotation-unchecked",
 ]
-