eds: Support custom_options for unknown EDS fields

canopen/objectdictionary/__init__.py

@@ -209,6 +209,8 @@ def __init__(self, name: str, index: int):
         self.storage_location = None
         self.subindices = {}
         self.names = {}
+        #: Key-Value pairs not defined by the standard
+        self.custom_options = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -268,6 +270,8 @@ def __init__(self, name: str, index: int):
         self.storage_location = None
         self.subindices = {}
         self.names = {}
+        #: Key-Value pairs not defined by the standard
+        self.custom_options = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -374,6 +378,8 @@ def __init__(self, name: str, index: int, subindex: int = 0):
         self.storage_location = None
         #: Can this variable be mapped to a PDO
         self.pdo_mappable = False
+        #: Key-Value pairs not defined by the standard
+        self.custom_options = {}
 
     def __repr__(self) -> str:
         subindex = self.subindex if isinstance(self.parent, (ODRecord, ODArray)) else None

canopen/objectdictionary/eds.py

@@ -22,6 +22,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 def import_eds(source, node_id):
     eds = RawConfigParser(inline_comment_prefixes=(';',))
     eds.optionxform = str
@@ -130,23 +131,26 @@ def import_eds(source, node_id):
 
             if object_type in (objectcodes.VAR, objectcodes.DOMAIN):
                 var = build_variable(eds, section, node_id, index)
+                var.custom_options = _get_custom_options(eds, section)
                 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 = 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.storage_location = storage_location
+                arr.custom_options = _get_custom_options(eds, section)
                 od.add_object(arr)
             elif object_type == objectcodes.ARRAY:
                 arr = ODArray(name, index)
                 arr.storage_location = storage_location
+                arr.custom_options = _get_custom_options(eds, section)
                 od.add_object(arr)
             elif object_type == objectcodes.RECORD:
                 record = ODRecord(name, index)
                 record.storage_location = storage_location
+                record.custom_options = _get_custom_options(eds, section)
                 od.add_object(record)
 
             continue
@@ -251,6 +255,22 @@ def _revert_variable(var_type, value):
     else:
         return f"0x{value:02X}"
 
+_STANDARD_OPTIONS = {
+    "ObjectType", "ParameterName", "DataType", "AccessType",
+    "PDOMapping", "LowLimit", "HighLimit", "DefaultValue",
+    "ParameterValue", "Factor", "Description", "Unit",
+    "StorageLocation", "CompactSubObj",
+    # CiA 306 fields parsed explicitly:
+    "SubNumber", "ObjFlags", "Denotation",
+}
+
+def _get_custom_options(eds, section):
+    custom_options = {}
+    for option, value in eds.items(section):
+        if option not in _STANDARD_OPTIONS:
+            custom_options[option] = value
+    return custom_options
+
 
 def build_variable(eds, section, node_id, index, subindex=0):
     """Creates a object dictionary entry.
@@ -330,6 +350,8 @@ def build_variable(eds, section, node_id, index, subindex=0):
             var.unit = eds.get(section, "Unit")
         except ValueError:
             pass
+
+    var.custom_options = _get_custom_options(eds, section)
     return var
 
 
@@ -404,12 +426,17 @@ def export_variable(var, eds):
         if getattr(var, 'unit', '') != '':
             eds.set(section, "Unit", var.unit)
 
+        for option, value in var.custom_options.items():
+            eds.set(section, option, value)
+
     def export_record(var, eds):
         section = f"{var.index:04X}"
         export_common(var, eds, section)
         eds.set(section, "SubNumber", f"0x{len(var.subindices):X}")
         ot = objectcodes.RECORD if isinstance(var, ODRecord) else objectcodes.ARRAY
         eds.set(section, "ObjectType", f"0x{ot:X}")
+        for option, value in var.custom_options.items():
+            eds.set(section, option, value)
         for i in var:
             export_variable(var[i], eds)
 

test/sample.eds

@@ -1017,3 +1017,33 @@ PDOMapping=0x0
 Factor=ERROR
 Description=
 Unit=
+
+[3061]
+ParameterName=Object with custom options
+ObjectType=0x7
+DataType=0x0007
+AccessType=rw
+PDOMapping=0
+Category=Motor
+Offset=100
+
+[3062]
+ParameterName=Record with custom options
+SubNumber=0x2
+ObjectType=0x9
+RecordTag=vendor_specific
+
+[3062sub0]
+ParameterName=Highest subindex
+ObjectType=0x7
+DataType=0x0005
+AccessType=ro
+DefaultValue=0x01
+PDOMapping=0
+
+[3062sub1]
+ParameterName=Value
+ObjectType=0x7
+DataType=0x0007
+AccessType=rw
+PDOMapping=0

test/test_eds.py

@@ -213,6 +213,53 @@ def test_reading_factor(self):
         self.assertEqual(var2.factor, 1)
         self.assertEqual(var2.unit, '')
 
+    def test_reading_custom_options(self):
+        # custom options (unknown EDS keys) are collected in custom_options dict
+        var = self.od[0x3061]
+        self.assertIsInstance(var, canopen.objectdictionary.ODVariable)
+        self.assertEqual(var.custom_options, {'Category': 'Motor', 'Offset': '100'})
+
+    def test_custom_options_standard_keys_excluded(self):
+        # Standard CiA 306 keys must NOT appear in custom_options
+        var = self.od[0x3061]
+        for key in ('ParameterName', 'ObjectType', 'DataType', 'AccessType', 'PDOMapping'):
+            self.assertNotIn(key, var.custom_options,
+                             f"Standard key {key!r} must not be in custom_options")
+
+    def test_custom_options_empty_for_standard_object(self):
+        # Objects without extra keys must have an empty custom_options dict
+        var = self.od['Producer heartbeat time']
+        self.assertEqual(var.custom_options, {})
+
+    def test_custom_options_record(self):
+        # custom_options is read for ODRecord container objects too
+        record = self.od[0x3062]
+        self.assertIsInstance(record, canopen.objectdictionary.ODRecord)
+        self.assertEqual(record.custom_options, {'RecordTag': 'vendor_specific'})
+        # sub-entries without extra keys have empty custom_options
+        self.assertEqual(record[1].custom_options, {})
+
+    def test_roundtrip_custom_options(self):
+        # custom_options 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.assertEqual(od2[0x3061].custom_options, {'Category': 'Motor', 'Offset': '100'})
+        self.assertEqual(od2[0x3062].custom_options, {'RecordTag': 'vendor_specific'})
+
+    def test_roundtrip_custom_options_not_duplicated_as_standard(self):
+        # After round-trip the re-imported object must not contain standard keys
+        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)
+        for key in ('ParameterName', 'ObjectType', 'DataType', 'AccessType', 'PDOMapping'):
+            self.assertNotIn(key, od2[0x3061].custom_options)
 
 
     def test_comments(self):