Merge remote-tracking branch 'origin/master' into fix/mypy-type-fixes

Author: bizfsc

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

@@ -18,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,

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.

canopen/objectdictionary/__init__.py

@@ -207,7 +207,7 @@ 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
         self.subindices: dict[int, ODVariable] = {}
         self.names: dict[str, ODVariable] = {}
 
@@ -268,7 +268,7 @@ 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
         self.subindices: dict[int, ODVariable] = {}
         self.names: dict[str, ODVariable] = {}
 
@@ -369,14 +369,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
         #: Can this variable be mapped to a PDO
         self.pdo_mappable = False
 
@@ -496,7 +498,8 @@ def decode_phys(self, value: Union[int, bool, float, str, bytes]) -> Union[int,
     def encode_phys(self, value: Union[int, bool, float, str, bytes]) -> int:
         if self.data_type in INTEGER_TYPES:
             assert isinstance(value, (int, float))
-            value = int(round(value / self.factor))
+            if self.factor != 1:
+                value = round(value / self.factor)
         return value  # type: ignore[return-value]
 
     def decode_desc(self, value: int) -> str:

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
                 od.add_object(arr)
             elif object_type == objectcodes.ARRAY:
@@ -158,7 +158,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
@@ -210,7 +214,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):
@@ -252,13 +258,22 @@ def _revert_variable(var_type, value):
         return f"0x{value:02X}"
 
 
-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)
@@ -268,15 +283,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
 
@@ -305,16 +324,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"))
@@ -370,7 +390,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:

doc/lss.rst

@@ -47,6 +47,10 @@ Or, you can run fastscan procedure ::
 
     ret_bool, lss_id_list = network.lss.fast_scan()
 
+.. note::
+   Before you run fastscan, you have to make sure the devices are in non-configured
+   or unconfigured state.
+
 Once one of sensors goes to CONFIGURATION state, you can read the current node id of the LSS slave::
 
     node_id = network.lss.inquire_node_id()

pyproject.toml

@@ -56,3 +56,14 @@ exclude = [
     "^test*",
     "^setup.py*",
 ]
+
+[tool.ruff]
+line-length = 96
+[tool.ruff.lint.isort]
+case-sensitive = true
+order-by-type = false
+lines-after-imports = 2
+
+[tool.black]
+line-length = 96
+skip-string-normalization = true

test/sample.eds

@@ -1017,3 +1017,30 @@ PDOMapping=0x0
 Factor=ERROR
 Description=
 Unit=
+
+[3063]
+ParameterName=DOMAIN object
+ObjectType=0x2
+DataType=0x0007
+AccessType=rw
+PDOMapping=0
+
+[3064]
+ParameterName=Record with DOMAIN sub-object
+SubNumber=0x2
+ObjectType=0x9
+
+[3064sub0]
+ParameterName=Highest subindex
+ObjectType=0x7
+DataType=0x0005
+AccessType=ro
+DefaultValue=0x01
+PDOMapping=0
+
+[3064sub1]
+ParameterName=DOMAIN sub-object
+ObjectType=0x2
+DataType=0x0007
+AccessType=rw
+PDOMapping=0

test/test_eds.py

@@ -111,6 +111,7 @@ def test_variable(self):
         self.assertEqual(var.name, 'Producer heartbeat time')
         self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED16)
         self.assertEqual(var.access_type, 'rw')
+        self.assertFalse(var.is_domain)
         self.assertEqual(var.default, 0)
         self.assertFalse(var.relative)
 
@@ -132,6 +133,7 @@ def test_record(self):
         self.assertEqual(var.subindex, 1)
         self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32)
         self.assertEqual(var.access_type, 'ro')
+        self.assertFalse(var.is_domain)
 
     def test_record_with_limits(self):
         int8 = self.od[0x3020]
@@ -166,6 +168,7 @@ def test_array_compact_subobj(self):
         self.assertEqual(var.subindex, 5)
         self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32)
         self.assertEqual(var.access_type, 'ro')
+        self.assertFalse(var.is_domain)
 
     def test_explicit_name_subobj(self):
         name = self.od[0x3004].name
@@ -197,6 +200,7 @@ def test_dummy_variable(self):
         self.assertEqual(var.name, 'Dummy0003')
         self.assertEqual(var.data_type, canopen.objectdictionary.INTEGER16)
         self.assertEqual(var.access_type, 'const')
+        self.assertFalse(var.is_domain)
         self.assertEqual(len(var), 16)
 
     def test_dummy_variable_undefined(self):
@@ -213,6 +217,39 @@ def test_reading_factor(self):
         self.assertEqual(var2.factor, 1)
         self.assertEqual(var2.unit, '')
 
+    def test_read_domain_object(self):
+        var = self.od[0x3063]
+        self.assertIsInstance(var, canopen.objectdictionary.ODVariable)
+        self.assertEqual(var.index, 0x3063)
+        self.assertEqual(var.subindex, 0)
+        self.assertEqual(var.name, 'DOMAIN object')
+        self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32)
+        self.assertEqual(var.access_type, 'rw')
+        self.assertTrue(var.is_domain)
+
+    def test_read_domain_subobject(self):
+        record = self.od[0x3064]
+        var = record[1]
+        self.assertIsInstance(var, canopen.objectdictionary.ODVariable)
+        self.assertEqual(var.index, 0x3064)
+        self.assertEqual(var.subindex, 1)
+        self.assertEqual(var.name, 'DOMAIN sub-object')
+        self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32)
+        self.assertEqual(var.access_type, 'rw')
+        self.assertTrue(var.is_domain)
+
+    def test_roundtrip_domain_objects(self):
+        # ObjectType==DOMAIN survive an EDS export/import round-trip
+        import io
+        with io.StringIO() as dest:
+            canopen.export_od(self.od, dest, 'eds')
+            dest.name = 'mock.eds'
+            dest.seek(0)
+            od2 = canopen.import_od(dest)
+        self.assertFalse(od2['Producer heartbeat time'].is_domain)
+        self.assertFalse(od2['Identity object']['Vendor-ID'].is_domain)
+        self.assertTrue(od2[0x3063].is_domain)
+        self.assertTrue(od2[0x3064][1].is_domain)
 
 
     def test_comments(self):