Merge remote-tracking branch 'origin/master' into pr/bizfsc/654

Author: bizfsc

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

@@ -107,7 +107,8 @@ def connect(self, *args, **kwargs) -> Network:
         if self.bus is None:
             self.bus = can.Bus(*args, **kwargs)
         logger.info("Connected to '%s'", self.bus.channel_info)
-        self.notifier = can.Notifier(self.bus, self.listeners, self.NOTIFIER_CYCLE)
+        if self.notifier is None:
+            self.notifier = can.Notifier(self.bus, self.listeners, self.NOTIFIER_CYCLE)
         return self
 
     def disconnect(self) -> None:
@@ -123,7 +124,11 @@ def disconnect(self) -> None:
         if self.bus is not None:
             self.bus.shutdown()
         self.bus = None
-        self.check()
+        try:
+            self.check()
+        finally:
+            # Release notifier after check
+            self.notifier = None
 
     def __enter__(self):
         return self

canopen/node/base.py

@@ -8,7 +8,7 @@ class BaseNode:
     """A CANopen node.
 
     :param node_id:
-        Node ID (set to None or 0 if specified by object dictionary)
+        Node ID (set to 0 if specified by object dictionary)
     :param object_dictionary:
         Object dictionary as either a path to a file, an ``ObjectDictionary``
         or a file like object.
@@ -25,7 +25,9 @@ def __init__(
             object_dictionary = import_od(object_dictionary, node_id)
         self.object_dictionary = object_dictionary
 
-        self.id = node_id or self.object_dictionary.node_id
+        self.id = node_id or object_dictionary.node_id or 0
+        if not 1 <= self.id <= 127:
+            raise ValueError(f"No valid Node ID provided, {self.id} not in range 1..127")
 
     def has_network(self) -> bool:
         """Check whether the node has been associated to a network."""

canopen/node/local.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Callable
 from typing import Union
 
 import canopen.network
@@ -17,6 +18,21 @@
 
 
 class LocalNode(BaseNode):
+    """Local CANopen node implementing essential communication services.
+
+    This does not provide a full-fledged communication logic stack, but needs
+    additional application logic to wire up the various services, such as
+    triggering PDO transmissions according to their communication parameters.
+
+    Notable exceptions are a local data store for SDO server access, and using
+    the Heartbeat Producer Time parameter to control Heartbeat transmission.
+
+    :param node_id:
+        Node ID (set to 0 if specified by object dictionary)
+    :param object_dictionary:
+        Object dictionary as either a path to a file, an ``ObjectDictionary``
+        or a file like object.
+    """
 
     def __init__(
         self,
@@ -26,8 +42,8 @@ def __init__(
         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[Callable] = []
+        self._write_callbacks: list[Callable] = []
 
         self.sdo = SdoServer(0x600 + self.id, 0x580 + self.id, self)
         self.tpdo = TPDO(self)
@@ -62,10 +78,10 @@ def remove_network(self) -> None:
         self.nmt.network = canopen.network._UNINITIALIZED_NETWORK
         self.emcy.network = canopen.network._UNINITIALIZED_NETWORK
 
-    def add_read_callback(self, callback):
+    def add_read_callback(self, callback: Callable):
         self._read_callbacks.append(callback)
 
-    def add_write_callback(self, callback):
+    def add_write_callback(self, callback: Callable):
         self._write_callbacks.append(callback)
 
     def get_data(

canopen/node/remote.py

@@ -19,7 +19,7 @@ class RemoteNode(BaseNode):
     """A CANopen remote node.
 
     :param node_id:
-        Node ID (set to None or 0 if specified by object dictionary)
+        Node ID (set to 0 if specified by object dictionary)
     :param object_dictionary:
         Object dictionary as either a path to a file, an ``ObjectDictionary``
         or a file like object.
@@ -39,7 +39,7 @@ def __init__(
         #: 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)

canopen/objectdictionary/__init__.py

@@ -160,7 +160,7 @@ def __iter__(self) -> Iterator[int]:
     def __len__(self) -> int:
         return len(self.indices)
 
-    def __contains__(self, index: Union[int, str]):
+    def __contains__(self, index: object) -> bool:
         return index in self.names or index in self.indices
 
     def add_object(self, obj: Union[ODArray, ODRecord, ODVariable]) -> None:
@@ -206,11 +206,11 @@ def __init__(self, name: str, index: int):
         #: Name of record
         self.name = name
         #: Storage location of index
-        self.storage_location = None
+        self.storage_location: Optional[str] = None
         #: CiA 306 ObjFlags bitfield
         self.obj_flags: int = 0
-        self.subindices = {}
-        self.names = {}
+        self.subindices: dict[int, ODVariable] = {}
+        self.names: dict[str, ODVariable] = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -236,10 +236,12 @@ def __len__(self) -> int:
     def __iter__(self) -> Iterator[int]:
         return iter(sorted(self.subindices))
 
-    def __contains__(self, subindex: Union[int, str]) -> bool:
+    def __contains__(self, subindex: object) -> bool:
         return subindex in self.names or subindex in self.subindices
 
-    def __eq__(self, other: ODRecord) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODRecord):
+            return NotImplemented
         return self.index == other.index
 
     def add_member(self, variable: ODVariable) -> None:
@@ -267,11 +269,11 @@ def __init__(self, name: str, index: int):
         #: Name of array
         self.name = name
         #: Storage location of index
-        self.storage_location = None
+        self.storage_location: Optional[str] = None
         #: CiA 306 ObjFlags bitfield
         self.obj_flags: int = 0
-        self.subindices = {}
-        self.names = {}
+        self.subindices: dict[int, ODVariable] = {}
+        self.names: dict[str, ODVariable] = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -302,7 +304,9 @@ def __len__(self) -> int:
     def __iter__(self) -> Iterator[int]:
         return iter(sorted(self.subindices))
 
-    def __eq__(self, other: ODArray) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODArray):
+            return NotImplemented
         return self.index == other.index
 
     def add_member(self, variable: ODVariable) -> None:
@@ -368,14 +372,16 @@ def __init__(self, name: str, index: int, subindex: int = 0):
         self.data_type: Optional[int] = None
         #: Access type, should be "rw", "ro", "wo", or "const"
         self.access_type: str = "rw"
+        #: The variable represents a DOMAIN ObjectType
+        self.is_domain: bool = False
         #: Description of variable
         self.description: str = ""
         #: Dictionary of value descriptions
         self.value_descriptions: dict[int, str] = {}
         #: Dictionary of bitfield definitions
         self.bit_definitions: dict[str, list[int]] = {}
         #: Storage location of index
-        self.storage_location = None
+        self.storage_location: Optional[str] = None
         #: CiA 306 ObjFlags bitfield
         self.obj_flags: int = 0
         #: CiA 306 Denotation string (DCF only)
@@ -395,7 +401,9 @@ def qualname(self) -> str:
             return f"{self.parent.name}.{self.name}"
         return self.name
 
-    def __eq__(self, other: ODVariable) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODVariable):
+            return NotImplemented
         return (self.index == other.index and
                 self.subindex == other.subindex)
 
@@ -421,7 +429,7 @@ def add_value_description(self, value: int, descr: str) -> None:
         """
         self.value_descriptions[value] = descr
 
-    def add_bit_definition(self, name: str, bits: List[int]) -> None:
+    def add_bit_definition(self, name: str, bits: list[int]) -> None:
         """Associate bit(s) with a string description.
 
         :param name: Name of bit(s)
@@ -492,8 +500,8 @@ def decode_phys(self, value: int) -> Union[int, bool, float, str, bytes]:
 
     def encode_phys(self, value: Union[int, bool, float, str, bytes]) -> int:
         if self.data_type in INTEGER_TYPES:
-            value /= self.factor
-            value = int(round(value))
+            if self.factor != 1:
+                value = round(value / self.factor)
         return value
 
     def decode_desc(self, value: int) -> str:
@@ -516,7 +524,7 @@ def encode_desc(self, desc: str) -> int:
         raise ValueError(
             f"No value corresponds to '{desc}'. Valid values are: {valid_values}")
 
-    def decode_bits(self, value: int, bits: List[int]) -> int:
+    def decode_bits(self, value: int, bits: list[int]) -> int:
         try:
             bits = self.bit_definitions[bits]
         except (TypeError, KeyError):
@@ -526,7 +534,7 @@ def decode_bits(self, value: int, bits: List[int]) -> int:
             mask |= 1 << bit
         return (value & mask) >> min(bits)
 
-    def encode_bits(self, original_value: int, bits: List[int], bit_value: int):
+    def encode_bits(self, original_value: int, bits: list[int], bit_value: int):
         try:
             bits = self.bit_definitions[bits]
         except (TypeError, KeyError):

canopen/objectdictionary/eds.py

@@ -41,7 +41,7 @@ def import_eds(source, node_id):
     od = ObjectDictionary()
 
     if eds.has_section("FileInfo"):
-        od.__edsFileInfo = {
+        od.__edsFileInfo = {  # type: ignore[attr-defined] # custom addition
             opt: eds.get("FileInfo", opt)
             for opt in eds.options("FileInfo")
         }
@@ -50,7 +50,7 @@ def import_eds(source, node_id):
         linecount = int(eds.get("Comments", "Lines"), 0)
         od.comments = '\n'.join([
             eds.get("Comments", f"Line{line}")
-            for line in range(1, linecount+1)
+            for line in range(1, linecount + 1)
         ])
 
     if not eds.has_section("DeviceInfo"):
@@ -129,15 +129,15 @@ def import_eds(source, node_id):
                 storage_location = None
 
             if object_type in (objectcodes.VAR, objectcodes.DOMAIN):
-                var = build_variable(eds, section, node_id, index)
+                var = build_variable(eds, section, node_id, object_type, index)
                 od.add_object(var)
             elif object_type == objectcodes.ARRAY and eds.has_option(section, "CompactSubObj"):
                 arr = ODArray(name, index)
                 last_subindex = ODVariable(
                     "Number of entries", index, 0)
                 last_subindex.data_type = datatypes.UNSIGNED8
                 arr.add_member(last_subindex)
-                arr.add_member(build_variable(eds, section, node_id, index, 1))
+                arr.add_member(build_variable(eds, section, node_id, object_type, index, 1))
                 arr.storage_location = storage_location
                 arr.obj_flags = _get_obj_flags(eds, section)
                 od.add_object(arr)
@@ -161,7 +161,11 @@ def import_eds(source, node_id):
             subindex = int(match.group(2), 16)
             entry = od[index]
             if isinstance(entry, (ODRecord, ODArray)):
-                var = build_variable(eds, section, node_id, index, subindex)
+                try:
+                    object_type = int(eds.get(section, "ObjectType"), 0)
+                except NoOptionError:
+                    object_type = objectcodes.VAR
+                var = build_variable(eds, section, node_id, object_type, index, subindex)
                 entry.add_member(var)
 
         # Match [index]Name
@@ -213,7 +217,9 @@ def _calc_bit_length(data_type):
     elif data_type == datatypes.INTEGER64:
         return 64
     else:
-        raise ValueError(f"Invalid data_type '{data_type}', expecting a signed integer data_type.")
+        raise ValueError(
+            f"Invalid data_type '{data_type}', expecting a signed integer data_type."
+        )
 
 
 def _signed_int_from_hex(hex_str, bit_length):
@@ -264,13 +270,23 @@ def _get_obj_flags(eds, section):
     return 0
 
 
-def build_variable(eds, section, node_id, index, subindex=0):
-    """Creates a object dictionary entry.
+def build_variable(
+    eds: RawConfigParser,
+    section: str,
+    node_id: int,
+    object_type: int,
+    index: int,
+    subindex: int = 0
+) -> ODVariable:
+    """Create a object dictionary entry.
+
+
     :param eds: String stream of the eds file
     :param section:
     :param node_id: Node ID
     :param index: Index of the CANOpen object
-    :param subindex: Subindex of the CANOpen object (if presente, else 0)
+    :param subindex: Subindex of the CANOpen object (if present, else 0)
+    :param is_domain: variable represents a DOMAIN ObjectType (if present, else False)
     """
     name = eds.get(section, "ParameterName")
     var = ODVariable(name, index, subindex)
@@ -280,15 +296,19 @@ def build_variable(eds, section, node_id, index, subindex=0):
         var.storage_location = None
     var.data_type = int(eds.get(section, "DataType"), 0)
     var.access_type = eds.get(section, "AccessType").lower()
+    var.is_domain = object_type == objectcodes.DOMAIN
     if var.data_type > 0x1B:
-        # The object dictionary editor from CANFestival creates an optional object if min max values are used
-        # This optional object is then placed in the eds under the section [A0] (start point, iterates for more)
-        # The eds.get function gives us 0x00A0 now convert to String without hex representation and upper case
-        # The sub2 part is then the section where the type parameter stands
+        # The object dictionary editor from CANFestival creates an optional object if min max
+        # values are used.  This optional object is then placed in the eds under the section
+        # [A0] (start point, iterates for more).  The eds.get function gives us 0x00A0 now
+        # convert to String without hex representation and upper case.  The sub2 part is then
+        # the section where the type parameter stands.
         try:
             var.data_type = int(eds.get(f"{var.data_type:X}sub1", "DefaultValue"), 0)
         except NoSectionError:
-            logger.warning("%s has an unknown or unsupported data type (0x%X)", name, var.data_type)
+            logger.warning(
+                "%s has an unknown or unsupported data type (0x%X)", name, var.data_type
+            )
             # Assume DOMAIN to force application to interpret the byte data
             var.data_type = datatypes.DOMAIN
 
@@ -317,16 +337,17 @@ def build_variable(eds, section, node_id, index, subindex=0):
             var.default_raw = eds.get(section, "DefaultValue")
             if '$NODEID' in var.default_raw:
                 var.relative = True
-            var.default = _convert_variable(node_id, var.data_type, eds.get(section, "DefaultValue"))
+            var.default = _convert_variable(node_id, var.data_type, var.default_raw)
         except ValueError:
             pass
     if eds.has_option(section, "ParameterValue"):
         try:
             var.value_raw = eds.get(section, "ParameterValue")
-            var.value = _convert_variable(node_id, var.data_type, eds.get(section, "ParameterValue"))
+            var.value = _convert_variable(node_id, var.data_type, var.value_raw)
         except ValueError:
             pass
-    # Factor, Description and Unit are not standard according to the CANopen specifications, but they are implemented in the python canopen package, so we can at least try to use them
+    # Factor, Description and Unit are not standard according to the CANopen specifications, but
+    # they are implemented in the python canopen package, so we can at least try to use them
     if eds.has_option(section, "Factor"):
         try:
             var.factor = float(eds.get(section, "Factor"))
@@ -385,7 +406,8 @@ def export_variable(var, eds):
             section = f"{var.index:04X}sub{var.subindex:X}"
 
         export_common(var, eds, section)
-        eds.set(section, "ObjectType", f"0x{objectcodes.VAR:X}")
+        object_type = objectcodes.DOMAIN if var.is_domain else objectcodes.VAR
+        eds.set(section, "ObjectType", f"0x{object_type:X}")
         if var.data_type:
             eds.set(section, "DataType", f"0x{var.data_type:04X}")
         if var.access_type: