Add mypy support and fixup project to give no errors

canopen/emcy.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import logging
 import struct
 import threading
@@ -55,7 +56,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

canopen/lss.py

@@ -85,7 +85,7 @@ def __init__(self) -> None:
         self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK
         self._node_id = 0
         self._data = None
-        self.responses = queue.Queue()
+        self.responses: queue.Queue[bytes] = queue.Queue()
 
     def send_switch_state_global(self, mode):
         """switch mode to CONFIGURATION_STATE or WAITING_STATE

canopen/network.py

@@ -3,7 +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, Final, Iterator, List, Optional, Union, TYPE_CHECKING, TextIO
 
 import can
 from can import Listener
@@ -16,6 +16,8 @@
 from canopen.sync import SyncProducer
 from canopen.timestamp import TimeProducer
 
+if TYPE_CHECKING:
+    from can.typechecking import CanData
 
 logger = logging.getLogger(__name__)
 
@@ -134,15 +136,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
@@ -157,14 +159,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,
+        node: Union[int, LocalNode],
+        object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None,
     ) -> LocalNode:
         """Create a local node in the network.
 
@@ -179,11 +183,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
@@ -211,7 +217,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.
 
@@ -227,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:
@@ -306,9 +314,9 @@ class PeriodicMessageTask:
     def __init__(
         self,
         can_id: int,
-        data: bytes,
+        data: CanData,
         period: float,
-        bus,
+        bus: can.BusABC,
         remote: bool = False,
     ):
         """

canopen/nmt.py

@@ -2,7 +2,7 @@
 import struct
 import threading
 import time
-from typing import Callable, Optional, TYPE_CHECKING
+from typing import Callable, Optional, TYPE_CHECKING, List
 
 import canopen.network
 
@@ -117,7 +117,7 @@ def __init__(self, node_id: int):
         #: 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:
@@ -187,7 +187,7 @@ 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)
+        self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, [], period, True)
 
     def stop_node_guarding(self):
         """Stops the node guarding mechanism."""

canopen/node/base.py

@@ -1,4 +1,4 @@
-from typing import TextIO, Union
+from typing import TextIO, Union, Optional
 
 import canopen.network
 from canopen.objectdictionary import ObjectDictionary, import_od
@@ -16,16 +16,19 @@ 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: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK
 
         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
 
     def has_network(self) -> bool:
         """Check whether the node has been associated to a network."""

canopen/node/local.py

@@ -1,33 +1,49 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict, Union
+from typing import Dict, Union, List, Protocol, TextIO, Optional
 
 import canopen.network
 from canopen import objectdictionary
 from canopen.emcy import EmcyProducer
 from canopen.nmt import NmtSlave
 from canopen.node.base import BaseNode
-from canopen.objectdictionary import ObjectDictionary
+from canopen.objectdictionary import ObjectDictionary, ODVariable
 from canopen.pdo import PDO, RPDO, TPDO
 from canopen.sdo import SdoAbortedError, SdoServer
 
 
 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,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TextIO, Union
+from typing import TextIO, Union, Optional, List
 
 import canopen.network
 from canopen.emcy import EmcyConsumer
@@ -30,16 +30,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)