Add mypy support and fixup project to give no errors

canopen/emcy.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import struct
 import logging
 import threading
@@ -52,7 +53,7 @@ def reset(self):
 
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
-    ) -> "EmcyError":
+    ) -> Optional[EmcyError]:
         """Wait for a new EMCY to arrive.
 
         :param emcy_code: EMCY code to wait for
@@ -86,10 +87,14 @@ 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,18 +3,21 @@
 from collections.abc import MutableMapping
 import logging
 import threading
-from typing import Callable, Dict, Iterator, List, Optional, Union
+from typing import Callable, Dict, Iterator, List, Optional, Union, TYPE_CHECKING, TextIO
 
 try:
     import can
     from can import Listener
     from can import CanError
 except ImportError:
-    # Do not fail if python-can is not installed
-    can = None
-    CanError = Exception
-    class Listener:
-        """ Dummy listener """
+    # Type checkers don't like this conditional logic, so it is only run when
+    # not type checking
+    if not TYPE_CHECKING:
+        # Do not fail if python-can is not installed
+        can = None
+        CanError = Exception
+        class Listener:
+            """ Dummy listener """
 
 from canopen.node import RemoteNode, LocalNode
 from canopen.sync import SyncProducer
@@ -24,6 +27,9 @@ class Listener:
 from canopen.objectdictionary.eds import import_from_node
 from canopen.objectdictionary import ObjectDictionary
 
+if TYPE_CHECKING:
+    from can.typechecking import CanData
+
 logger = logging.getLogger(__name__)
 
 Callback = Callable[[int, bytearray, float], None]
@@ -45,7 +51,7 @@ def __init__(self, bus: Optional[can.BusABC] = None):
         #: List of :class:`can.Listener` objects.
         #: Includes at least MessageListener.
         self.listeners = [MessageListener(self)]
-        self.notifier = None
+        self.notifier: Optional[can.Notifier] = None
         self.nodes: Dict[int, Union[RemoteNode, LocalNode]] = {}
         self.subscribers: Dict[int, List[Callback]] = {}
         self.send_lock = threading.Lock()
@@ -138,15 +144,15 @@ def __exit__(self, type, value, traceback):
 
     def add_node(
         self,
-        node: Union[int, RemoteNode, LocalNode],
-        object_dictionary: Union[str, ObjectDictionary, None] = None,
+        node: Union[int, RemoteNode],
+        object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None,
         upload_eds: bool = False,
     ) -> RemoteNode:
         """Add a remote node to the network.
 
         :param node:
             Can be either an integer representing the node ID, a
-            :class:`canopen.RemoteNode` or :class:`canopen.LocalNode` object.
+            :class:`canopen.RemoteNode` object.
         :param object_dictionary:
             Can be either a string for specifying the path to an
             Object Dictionary file or a
@@ -161,14 +167,16 @@ def add_node(
             if upload_eds:
                 logger.info("Trying to read EDS from node %d", node)
                 object_dictionary = import_from_node(node, self)
-            node = RemoteNode(node, object_dictionary)
-        self[node.id] = node
-        return node
+            nodeobj = RemoteNode(node, object_dictionary)
+        else:
+            nodeobj = node
+        self[nodeobj.id] = nodeobj
+        return nodeobj
 
     def create_node(
         self,
         node: int,
-        object_dictionary: Union[str, ObjectDictionary, None] = None,
+        object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None,
     ) -> LocalNode:
         """Create a local node in the network.
 
@@ -183,11 +191,13 @@ def create_node(
             The Node object that was added.
         """
         if isinstance(node, int):
-            node = LocalNode(node, object_dictionary)
-        self[node.id] = node
-        return node
+            nodeobj = LocalNode(node, object_dictionary)
+        else:
+            nodeobj = node
+        self[nodeobj.id] = nodeobj
+        return nodeobj
 
-    def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None:
+    def send_message(self, can_id: int, data: CanData, remote: bool = False) -> None:
         """Send a raw CAN message to the network.
 
         This method may be overridden in a subclass if you need to integrate
@@ -215,7 +225,7 @@ def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None:
         self.check()
 
     def send_periodic(
-        self, can_id: int, data: bytes, period: float, remote: bool = False
+        self, can_id: int, data: CanData, period: float, remote: bool = False
     ) -> PeriodicMessageTask:
         """Start sending a message periodically.
 
@@ -295,7 +305,7 @@ class PeriodicMessageTask:
     def __init__(
         self,
         can_id: int,
-        data: bytes,
+        data: CanData,
         period: float,
         bus,
         remote: bool = False,
@@ -335,10 +345,12 @@ 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)
-            self._task.stop()
+            if self._task is not None:
+                self._task.stop()
             self._start()
 
 

canopen/nmt.py

@@ -2,7 +2,10 @@
 import logging
 import struct
 import time
-from typing import Callable, Optional
+from typing import Callable, Optional, List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from canopen.network import Network, PeriodicMessageTask
 
 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):
@@ -107,11 +110,11 @@ class NmtMaster(NmtBase):
     def __init__(self, node_id: int):
         super(NmtMaster, self).__init__(node_id)
         self._state_received = None
-        self._node_guarding_producer = None
+        self._node_guarding_producer: Optional[PeriodicMessageTask] = None
         #: Timestamp of last heartbeat message
         self.timestamp: Optional[float] = None
         self.state_update = threading.Condition()
-        self._callbacks = []
+        self._callbacks: List[Callable[[int], None]] = []
 
     def on_heartbeat(self, can_id, data, timestamp):
         with self.state_update:
@@ -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")
         self.network.send_message(0, [code, self.id])
 
     def wait_for_heartbeat(self, timeout: float = 10):
@@ -181,7 +186,9 @@ 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()
-        self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, None, period, True)
+        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):
         """Stops the node guarding mechanism."""
@@ -197,7 +204,7 @@ class NmtSlave(NmtBase):
 
     def __init__(self, node_id: int, local_node):
         super(NmtSlave, self).__init__(node_id)
-        self._send_task = None
+        self._send_task: Optional[PeriodicMessageTask] = None
         self._heartbeat_time_ms = 0
         self._local_node = local_node
 
@@ -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("A Network is required")
             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("A network is required")
             self._send_task = self.network.send_periodic(
                 0x700 + self.id, [self._state], heartbeat_time_ms / 1000.0)
 

canopen/node/base.py

@@ -1,4 +1,4 @@
-from typing import TextIO, Union
+from typing import TextIO, Union, Optional
 from canopen.objectdictionary import ObjectDictionary, import_od
 
 
@@ -14,13 +14,16 @@ class BaseNode:
 
     def __init__(
         self,
-        node_id: int,
-        object_dictionary: Union[ObjectDictionary, str, TextIO],
+        node_id: Optional[int],
+        object_dictionary: Union[ObjectDictionary, str, TextIO, None],
     ):
         self.network = None
 
         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
+        node_id = node_id or self.object_dictionary.node_id
+        if node_id is None:
+            raise ValueError("Node ID must be specified")
+        self.id: int = node_id

canopen/node/local.py

@@ -1,29 +1,45 @@
 import logging
-from typing import Dict, Union
+from typing import Dict, Union, List, Protocol, TextIO, Optional
 
 from canopen.node.base import BaseNode
 from canopen.sdo import SdoServer, SdoAbortedError
 from canopen.pdo import PDO, TPDO, RPDO
 from canopen.nmt import NmtSlave
 from canopen.emcy import EmcyProducer
-from canopen.objectdictionary import ObjectDictionary
+from canopen.objectdictionary import ObjectDictionary, ODVariable
 from canopen import objectdictionary
 
 logger = logging.getLogger(__name__)
 
 
+class WriteCallback(Protocol):
+    """LocalNode Write Callback Protocol"""
+    def __call__(self, *, index: int, subindex: int,
+                 od: ODVariable,
+                 data: bytes) -> None:
+        ''' Write Callback '''
+
+
+class ReadCallback(Protocol):
+    """LocalNode Read Callback Protocol"""
+    def __call__(self, *, index: int, subindex: int,
+                 od: ODVariable
+                 ) -> Union[bool, int, float, str, bytes, None]:
+        ''' Read Callback '''
+
+
 class LocalNode(BaseNode):
 
     def __init__(
         self,
-        node_id: int,
-        object_dictionary: Union[ObjectDictionary, str],
+        node_id: Optional[int],
+        object_dictionary: Union[ObjectDictionary, str, TextIO, None],
     ):
         super(LocalNode, self).__init__(node_id, object_dictionary)
 
         self.data_store: Dict[int, Dict[int, bytes]] = {}
-        self._read_callbacks = []
-        self._write_callbacks = []
+        self._read_callbacks: List[ReadCallback] = []
+        self._write_callbacks: List[WriteCallback] = []
 
         self.sdo = SdoServer(0x600 + self.id, 0x580 + self.id, self)
         self.tpdo = TPDO(self)

canopen/node/remote.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Union, TextIO
+from typing import Union, TextIO, List, Optional
 
 from canopen.sdo import SdoClient, SdoCommunicationError, SdoAbortedError
 from canopen.nmt import NmtMaster
@@ -26,16 +26,16 @@ class RemoteNode(BaseNode):
 
     def __init__(
         self,
-        node_id: int,
-        object_dictionary: Union[ObjectDictionary, str, TextIO],
+        node_id: Optional[int],
+        object_dictionary: Union[ObjectDictionary, str, TextIO, None],
         load_od: bool = False,
     ):
         super(RemoteNode, self).__init__(node_id, object_dictionary)
 
         #: Enable WORKAROUND for reversed PDO mapping entries
         self.curtis_hack = False
 
-        self.sdo_channels = []
+        self.sdo_channels: List[SdoClient] = []
         self.sdo = self.add_sdo(0x600 + self.id, 0x580 + self.id)
         self.tpdo = TPDO(self)
         self.rpdo = RPDO(self)