From bc0551f3780b4258fc286b0f840424c4075f7327 Mon Sep 17 00:00:00 2001 From: Taras Zaporozhets Date: Thu, 11 Sep 2025 18:46:21 +0200 Subject: [PATCH] xdd: Implement option to load object dictionary from XDD file. Signed-off-by: Taras Zaporozhets --- canopen/objectdictionary/__init__.py | 7 +- canopen/objectdictionary/xdd.py | 357 +++++++ test/sample.xdd | 1380 ++++++++++++++++++++++++++ test/test_xdd.py | 171 ++++ test/util.py | 1 + 5 files changed, 1914 insertions(+), 2 deletions(-) create mode 100644 canopen/objectdictionary/xdd.py create mode 100644 test/sample.xdd create mode 100644 test/test_xdd.py diff --git a/canopen/objectdictionary/__init__.py b/canopen/objectdictionary/__init__.py index fa694c56..bb646f4b 100644 --- a/canopen/objectdictionary/__init__.py +++ b/canopen/objectdictionary/__init__.py @@ -76,7 +76,7 @@ def import_od( source: Union[str, TextIO, None], node_id: Optional[int] = None, ) -> ObjectDictionary: - """Parse an EDS, DCF, or EPF file. + """Parse an EDS, DCF, EPF or XDD file. :param source: The path to object dictionary file, a file like object, or an EPF XML tree. @@ -106,9 +106,12 @@ def import_od( elif suffix == ".epf": from canopen.objectdictionary import epf return epf.import_epf(source) + elif suffix == ".xdd": + from canopen.objectdictionary import xdd + return xdd.import_xdd(source, node_id) else: doc_type = suffix[1:] - allowed = ", ".join(["eds", "dcf", "epf"]) + allowed = ", ".join(["eds", "dcf", "epf", "xdd"]) raise ValueError( f"Cannot import from the {doc_type!r} format; " f"supported formats: {allowed}" diff --git a/canopen/objectdictionary/xdd.py b/canopen/objectdictionary/xdd.py new file mode 100644 index 00000000..e7386c4f --- /dev/null +++ b/canopen/objectdictionary/xdd.py @@ -0,0 +1,357 @@ +import functools +import logging +import os +import re +import xml.etree.ElementTree as etree +from typing import Union, Optional +from canopen.objectdictionary import ( + ODArray, + ODRecord, + ODVariable, + ObjectDictionary, + datatypes, + objectcodes, +) + +logger = logging.getLogger(__name__) + +autoint = functools.partial(int, base=0) + +def import_xdd( + xdd: Union[etree.Element, str, bytes, os.PathLike], + node_id: Optional[int], +) -> ObjectDictionary: + od = ObjectDictionary() + if etree.iselement(xdd): + root = xdd + else: + root = etree.parse(xdd).getroot() + + if node_id is None: + device_commissioning = root.find('.//{*}DeviceCommissioning') + if device_commissioning is not None: + od.node_id = int(device_commissioning['nodeID'], 0) + else: + od.node_id = None + else: + od.node_id = node_id + + _add_device_information(od, root) + _add_object_list(od, root) + _add_dummy_objects(od, root) + + return od + + +def _add_device_information( + od: ObjectDictionary, + root: etree.Element +): + device_identity = root.find('.//{*}DeviceIdentity') + if device_identity is not None: + for src_prop, dst_prop, f in [ + ("vendorName", "vendor_name", str), + ("vendorID", "vendor_number", autoint), + ("productName", "product_name", str), + ("productID", "product_number", autoint), + ]: + val = device_identity.find(f'{{*}}{src_prop}') + if val is not None and val.text: + setattr(od.device_information, dst_prop, f(val.text)) + + general_features = root.find('.//{*}CANopenGeneralFeatures') + if general_features is not None: + for src_prop, dst_prop, f, default in [ + # properties without default value (default=None) are required + ("granularity", "granularity", autoint, None), + ("nrOfRxPDO", "nr_of_RXPDO", autoint, 0), + ("nrOfTxPDO", "nr_of_TXPDO", autoint, 0), + ("bootUpSlave", "simple_boot_up_slave", bool, False), + ]: + val = general_features.get(src_prop, default) + if val is None: + raise ValueError(f"Missing required '{src_prop}' property in XDD file") + setattr(od.device_information, dst_prop, f(val)) + + + baud_rate = root.find('.//{*}PhysicalLayer/{*}baudRate') + for baud in baud_rate: + try: + rate = int(baud.get("value").replace(' Kbps', ''), 10) * 1000 + od.device_information.allowed_baudrates.add(rate) + except (ValueError, TypeError): + pass + + if default_baud := baud_rate.get('defaultValue', None): + try: + od.bitrate = int(default_baud.replace(' Kbps', ''), 10) * 1000 + except (ValueError, TypeError): + pass + + +def _add_object_list( + od: ObjectDictionary, + root: etree.Element +): + # Process all CANopen objects in the file + for obj in root.findall('.//{*}CANopenObjectList/{*}CANopenObject'): + name = obj.get('name', '') + index = int(obj.get('index', '0'), 16) + object_type = int(obj.get('objectType', '0')) + sub_number = obj.get('subNumber') + + # Simple variable + if object_type == objectcodes.VAR: + unique_id_ref = obj.get('uniqueIDRef', None) + parameters = root.find(f'.//{{*}}parameter[@uniqueID="{unique_id_ref}"]') + + var = _build_variable(parameters, od.node_id, name, index) + _set_parameters_from_xdd_canopen_object(od.node_id, var, obj) + od.add_object(var) + + # Array + elif object_type == objectcodes.ARRAY and sub_number: + array = ODArray(name, index) + for sub_obj in obj: + sub_name = sub_obj.get('name', '') + sub_index = int(sub_obj.get('subIndex'), 16) + sub_unique_id = sub_obj.get('uniqueIDRef', None) + sub_parameters = root.find(f'.//{{*}}parameter[@uniqueID="{sub_unique_id}"]') + + sub_var = _build_variable(sub_parameters, od.node_id, sub_name, index, sub_index) + _set_parameters_from_xdd_canopen_object(od.node_id, sub_var, sub_obj) + array.add_member(sub_var) + od.add_object(array) + + # Record/Struct + elif object_type == objectcodes.RECORD and sub_number: + record = ODRecord(name, index) + for sub_obj in obj: + sub_name = sub_obj.get('name', '') + sub_index = int(sub_obj.get('subIndex')) + sub_unique_id = sub_obj.get('uniqueIDRef', None) + sub_parameters = root.find(f'.//{{*}}parameter[@uniqueID="{sub_unique_id}"]') + sub_var = _build_variable(sub_parameters, od.node_id, sub_name, index, sub_index) + _set_parameters_from_xdd_canopen_object(od.node_id, sub_var, sub_obj) + record.add_member(sub_var) + od.add_object(record) + + +def _add_dummy_objects( + od: ObjectDictionary, + root: etree.Element +): + dummy_section = root.find('.//{*}ApplicationLayers/{*}dummyUsage') + for dummy in dummy_section: + p = dummy.get('entry').split('=') + key = p[0] + value = int(p[1], 10) + index = int(key.replace('Dummy', ''), 10) + if value == 1: + var = ODVariable(key, index, 0) + var.data_type = index + var.access_type = "const" + od.add_object(var) + + +def _set_parameters_from_xdd_canopen_object( + node_id: Optional[int], + dst: ODVariable, + src: etree.Element +): + # PDO mapping of the object, optional, string + # Valid values: + # * no – not mappable + # * default – mapped by default + # * optional – optionally mapped + # * TPDO – may be mapped into TPDO only + # * RPDO – may be mapped into RPDO only + pdo_mapping = src.get('PDOmapping', 'no') + dst.pdo_mappable = pdo_mapping != 'no' + + # Name of the object, optional, string + if var_name := src.get('name', None): + dst.name = var_name + + # CANopen data type (two hex digits), optional + # data_type matches canopen library, no conversion needed + if var_data_type := src.get('dataType', None): + try: + dst.data_type = int(var_data_type, 16) + except (ValueError, TypeError): + pass + + # Access type of the object; valid values, optional, string + # * const – read access only; the value is not changing + # * ro – read access only + # * wo – write access only + # * rw – both read and write access + # strings match with access_type in canopen library, no conversion needed + if access_type := src.get('accessType', None): + dst.access_type = access_type + + # Low limit of the parameter value, optional, string + if min_value := src.get('lowLimit', None): + try: + dst.min = _convert_variable(node_id, dst.data_type, min_value) + except (ValueError, TypeError): + pass + + # High limit of the parameter value, optional, string + if max_value := src.get('highLimit', None): + try: + dst.max = _convert_variable(node_id, dst.data_type, max_value) + except (ValueError, TypeError): + pass + + # Default value of the object, optional, string + if default_value := src.get('defaultValue', None): + try: + dst.default_raw = default_value + if '$NODEID' in dst.default_raw: + dst.relative = True + dst.default = _convert_variable(node_id, dst.data_type, dst.default_raw) + except (ValueError, TypeError): + pass + + +def _build_variable( + par_tree: Optional[etree.Element], + node_id: Optional[int], + name: str, + index: int, + subindex: int = 0 +) -> ODVariable: + var = ODVariable(name, index, subindex) + # Set default parameters + var.default_raw = None + var.access_type = 'ro' + if par_tree is None: + return var + + var.description = par_tree.get('description', '') + + # Extract data type + data_types = { + 'BOOL': datatypes.BOOLEAN, + 'SINT': datatypes.INTEGER8, + 'INT': datatypes.INTEGER16, + 'DINT': datatypes.INTEGER32, + 'LINT': datatypes.INTEGER64, + 'USINT': datatypes.UNSIGNED8, + 'UINT': datatypes.UNSIGNED16, + 'UDINT': datatypes.UNSIGNED32, + 'ULINT': datatypes.UNSIGNED32, + 'REAL': datatypes.REAL32, + 'LREAL': datatypes.REAL64, + 'STRING': datatypes.VISIBLE_STRING, + 'BITSTRING': datatypes.DOMAIN, + 'WSTRING': datatypes.UNICODE_STRING + } + + for k, v in data_types.items(): + if par_tree.find(f'{{*}}{k}') is not None: + var.data_type = v + + # Extract access type + if access_type_str := par_tree.get('access', None): + # Defines which operations are valid for the parameter: + # * const – read access only; the value is not changing + # * read – read access only (default value) + # * write – write access only + # * readWrite – both read and write access + # * readWriteInput – both read and write access, but represents process input data + # * readWriteOutput – both read and write access, but represents process output data + # * noAccess – access denied + access_types = { + 'const': 'const', + 'read': 'ro', + 'write': 'wo', + 'readWrite': 'rw', + 'readWriteInput': 'rw', + 'readWriteOutput': 'rw', + 'noAccess': 'const', + } + var.access_type = access_types.get(access_type_str) + + # Extract default value + default_value = par_tree.find('{*}defaultValue') + if default_value is not None: + try: + var.default_raw = default_value.get('value') + if '$NODEID' in var.default_raw: + var.relative = True + var.default = _convert_variable(node_id, var.data_type, var.default_raw) + except (ValueError, TypeError): + pass + + # Extract allowed values range + min_value = par_tree.find('{*}allowedValues/{*}range/{*}minValue') + if min_value is not None: + try: + var.min = _convert_variable(node_id, var.data_type, min_value.get('value')) + except (ValueError, TypeError): + pass + + max_value = par_tree.find('{*}allowedValues/{*}range/{*}maxValue') + if max_value is not None: + try: + var.max = _convert_variable(node_id, var.data_type, max_value.get('value')) + except (ValueError, TypeError): + pass + return var + + +def _calc_bit_length( + data_type: int +) -> int: + if data_type == datatypes.INTEGER8: + return 8 + elif data_type == datatypes.INTEGER16: + return 16 + elif data_type == datatypes.INTEGER32: + return 32 + elif data_type == datatypes.INTEGER64: + return 64 + else: + raise ValueError(f"Invalid data_type '{data_type}', expecting a signed integer data_type.") + + +def _signed_int_from_hex( + hex_str: str, + bit_length: int +) -> int: + number = int(hex_str, 0) + max_value = (1 << (bit_length - 1)) - 1 + + if number > max_value: + return number - (1 << bit_length) + else: + return number + + +def _convert_variable( + node_id: Optional[int], + var_type: int, + value: str +) -> Optional[Union[bytes, str, float, int]]: + if var_type in (datatypes.OCTET_STRING, datatypes.DOMAIN): + return bytes.fromhex(value) + elif var_type in (datatypes.VISIBLE_STRING, datatypes.UNICODE_STRING): + return str(value) + elif var_type in datatypes.FLOAT_TYPES: + return float(value) + else: + # COB-ID can contain '$NODEID+' so replace this with node_id before converting + value = value.replace(" ", "").upper() + if '$NODEID' in value: + if node_id is None: + logger.warn("Cannot convert value with $NODEID, skipping conversion") + return None + else: + return int(re.sub(r'\+?\$NODEID\+?', '', value), 0) + node_id + else: + if var_type in datatypes.SIGNED_TYPES: + return _signed_int_from_hex(value, _calc_bit_length(var_type)) + else: + return int(value, 0) diff --git a/test/sample.xdd b/test/sample.xdd new file mode 100644 index 00000000..926f6533 --- /dev/null +++ b/test/sample.xdd @@ -0,0 +1,1380 @@ + + + + + + CANopen device profile + 1.1 + + + Device + + 1 + 1 + CANopen + + + + + Vendor Name + 1 + + + 0 + 0 + 0 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + CANopen communication network profile + 1.1 + + + CommunicationNetwork + + 1 + 1 + CANopen + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/test/test_xdd.py b/test/test_xdd.py new file mode 100644 index 00000000..8c397656 --- /dev/null +++ b/test/test_xdd.py @@ -0,0 +1,171 @@ +import unittest + +import canopen +from canopen.objectdictionary.xdd import _signed_int_from_hex +from canopen.utils import pretty_index + +from .util import SAMPLE_XDD + + +class TestXDD(unittest.TestCase): + + test_data = { + "int8": [ + {"hex_str": "7F", "bit_length": 8, "expected": 127}, + {"hex_str": "80", "bit_length": 8, "expected": -128}, + {"hex_str": "FF", "bit_length": 8, "expected": -1}, + {"hex_str": "00", "bit_length": 8, "expected": 0}, + {"hex_str": "01", "bit_length": 8, "expected": 1} + ], + "int16": [ + {"hex_str": "7FFF", "bit_length": 16, "expected": 32767}, + {"hex_str": "8000", "bit_length": 16, "expected": -32768}, + {"hex_str": "FFFF", "bit_length": 16, "expected": -1}, + {"hex_str": "0000", "bit_length": 16, "expected": 0}, + {"hex_str": "0001", "bit_length": 16, "expected": 1} + ], + "int24": [ + {"hex_str": "7FFFFF", "bit_length": 24, "expected": 8388607}, + {"hex_str": "800000", "bit_length": 24, "expected": -8388608}, + {"hex_str": "FFFFFF", "bit_length": 24, "expected": -1}, + {"hex_str": "000000", "bit_length": 24, "expected": 0}, + {"hex_str": "000001", "bit_length": 24, "expected": 1} + ], + "int32": [ + {"hex_str": "7FFFFFFF", "bit_length": 32, "expected": 2147483647}, + {"hex_str": "80000000", "bit_length": 32, "expected": -2147483648}, + {"hex_str": "FFFFFFFF", "bit_length": 32, "expected": -1}, + {"hex_str": "00000000", "bit_length": 32, "expected": 0}, + {"hex_str": "00000001", "bit_length": 32, "expected": 1} + ], + "int64": [ + {"hex_str": "7FFFFFFFFFFFFFFF", "bit_length": 64, "expected": 9223372036854775807}, + {"hex_str": "8000000000000000", "bit_length": 64, "expected": -9223372036854775808}, + {"hex_str": "FFFFFFFFFFFFFFFF", "bit_length": 64, "expected": -1}, + {"hex_str": "0000000000000000", "bit_length": 64, "expected": 0}, + {"hex_str": "0000000000000001", "bit_length": 64, "expected": 1} + ] + } + + def setUp(self): + self.od = canopen.import_od(SAMPLE_XDD, 2) + + def test_load_nonexisting_file(self): + with self.assertRaises(IOError): + canopen.import_od('/path/to/wrong_file.xdd') + + def test_load_unsupported_format(self): + with self.assertRaisesRegex(ValueError, "'py'"): + canopen.import_od(__file__) + + def test_load_file_object(self): + with open(SAMPLE_XDD) as fp: + od = canopen.import_od(fp) + self.assertTrue(len(od) > 0) + + def test_load_explicit_nodeid(self): + od = canopen.import_od(SAMPLE_XDD, node_id=3) + self.assertEqual(od.node_id, 3) + + def test_variable(self): + var = self.od['Producer heartbeat time'] + self.assertIsInstance(var, canopen.objectdictionary.ODVariable) + self.assertEqual(var.index, 0x1017) + self.assertEqual(var.subindex, 0) + self.assertEqual(var.name, 'Producer heartbeat time') + self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED16) + self.assertEqual(var.access_type, 'rw') + self.assertEqual(var.default, 0) + self.assertFalse(var.relative) + + def test_relative_variable(self): + var = self.od['Receive PDO 0 Communication Parameter']['COB-ID use by RPDO 1'] + self.assertTrue(var.relative) + self.assertEqual(var.default, 512 + self.od.node_id) + + def test_record(self): + record = self.od['Identity object'] + self.assertIsInstance(record, canopen.objectdictionary.ODRecord) + self.assertEqual(len(record), 4) + self.assertEqual(record.index, 0x1018) + self.assertEqual(record.name, 'Identity object') + var = record['Vendor-ID'] + self.assertIsInstance(var, canopen.objectdictionary.ODVariable) + self.assertEqual(var.name, 'Vendor-ID') + self.assertEqual(var.index, 0x1018) + self.assertEqual(var.subindex, 1) + self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32) + self.assertEqual(var.access_type, 'ro') + + def test_record_with_limits(self): + int8 = self.od[0x3020] + self.assertEqual(int8.min, 0) + self.assertEqual(int8.max, 127) + uint8 = self.od[0x3021] + self.assertEqual(uint8.min, 2) + self.assertEqual(uint8.max, 10) + int32 = self.od[0x3030] + self.assertEqual(int32.min, -2147483648) + self.assertEqual(int32.max, -1) + int64 = self.od[0x3040] + self.assertEqual(int64.min, -10) + self.assertEqual(int64.max, +10) + + def test_array_compact_subobj(self): + array = self.od[0x1003] + self.assertIsInstance(array, canopen.objectdictionary.ODArray) + self.assertEqual(array.index, 0x1003) + self.assertEqual(array.name, 'Pre-defined error field') + var = array[5] + self.assertIsInstance(var, canopen.objectdictionary.ODVariable) + self.assertEqual(var.name, 'Pre-defined error field_5') + self.assertEqual(var.index, 0x1003) + self.assertEqual(var.subindex, 5) + self.assertEqual(var.data_type, canopen.objectdictionary.UNSIGNED32) + self.assertEqual(var.access_type, 'ro') + + def test_explicit_name_subobj(self): + name = self.od[0x3004].name + self.assertEqual(name, 'Sensor Status') + name = self.od[0x3004][1].name + self.assertEqual(name, 'Sensor Status 1') + name = self.od[0x3004][3].name + self.assertEqual(name, 'Sensor Status 3') + value = self.od[0x3004][3].default + self.assertEqual(value, 3) + + def test_parameter_name_with_percent(self): + name = self.od[0x3003].name + self.assertEqual(name, 'Valve % open') + + def test_compact_subobj_parameter_name_with_percent(self): + name = self.od[0x3006].name + self.assertEqual(name, 'Valve 1 % Open') + + def test_sub_index_w_capital_s(self): + name = self.od[0x3010][0].name + self.assertEqual(name, 'Temperature') + + def test_dummy_variable(self): + var = self.od['Dummy0003'] + self.assertIsInstance(var, canopen.objectdictionary.ODVariable) + self.assertEqual(var.index, 0x0003) + self.assertEqual(var.subindex, 0) + self.assertEqual(var.name, 'Dummy0003') + self.assertEqual(var.data_type, canopen.objectdictionary.INTEGER16) + self.assertEqual(var.access_type, 'const') + self.assertEqual(len(var), 16) + + def test_dummy_variable_undefined(self): + with self.assertRaises(KeyError): + var_undef = self.od['Dummy0001'] + + def test_signed_int_from_hex(self): + for data_type, test_cases in self.test_data.items(): + for test_case in test_cases: + with self.subTest(data_type=data_type, test_case=test_case): + result = _signed_int_from_hex('0x' + test_case["hex_str"], test_case["bit_length"]) + self.assertEqual(result, test_case["expected"]) + +if __name__ == "__main__": + unittest.main() diff --git a/test/util.py b/test/util.py index 2c8eccca..9c87856f 100644 --- a/test/util.py +++ b/test/util.py @@ -5,6 +5,7 @@ DATATYPES_EDS = os.path.join(os.path.dirname(__file__), "datatypes.eds") SAMPLE_EDS = os.path.join(os.path.dirname(__file__), "sample.eds") +SAMPLE_XDD = os.path.join(os.path.dirname(__file__), "sample.xdd") @contextlib.contextmanager