eds: Fix signed integer limit parsing and export for all SIGNED_TYPES

canopen/objectdictionary/eds.py

@@ -205,14 +205,11 @@ def import_from_node(node_id: int, network: canopen.network.Network):
 
 
 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
+    if data_type in datatypes.INTEGER_TYPES:
+        st = ODVariable.STRUCT_TYPES[data_type]
+        if isinstance(st, datatypes.IntegerN):
+            return st.width
+        return st.size * 8
     else:
         raise ValueError(
             f"Invalid data_type 0x{data_type:04X}, expecting an integer data_type."
@@ -221,15 +218,29 @@ def _calc_bit_length(data_type: int) -> int:
 
 def _signed_int_from_hex(hex_str, bit_length):
     number = int(hex_str, 0)
-    max_value = (1 << (bit_length - 1)) - 1
+    min_signed = -(1 << (bit_length - 1))
+    max_signed = (1 << (bit_length - 1)) - 1
+    max_unsigned = (1 << bit_length) - 1
 
-    if number > max_value:
-        return number - (1 << bit_length)
-    else:
+    if number < min_signed:
+        raise ValueError(
+            f"Value {hex_str!r} is out of range for a {bit_length}-bit signed integer"
+        )
+    if number < 0:
+        # Negative literal (e.g. LowLimit=-32768 or -0x8000)
         return number
 
+    if number > max_unsigned:
+        raise ValueError(
+            f"Value {hex_str!r} is out of range for a {bit_length}-bit signed integer"
+        )
+    if number > max_signed:
+        # Unsigned hex literal, two's-complement (e.g. LowLimit=0xFFFF → -1 for INTEGER16)
+        return number - (1 << bit_length)
+    return number
+
 
-def _convert_variable(node_id: int, var_type: int, value: Any) -> Any:
+def _decode_from_eds(node_id: int, var_type: int, value: Any) -> Any:
     if var_type in (datatypes.OCTET_STRING, datatypes.DOMAIN):
         return bytes.fromhex(value)
     elif var_type in (datatypes.VISIBLE_STRING, datatypes.UNICODE_STRING):
@@ -245,10 +256,10 @@ def _convert_variable(node_id: int, var_type: int, value: Any) -> Any:
             return int(value, 0)
 
 
-def _revert_variable(var_type: int, value: Any) -> Any:
+def _encode_to_eds(var_type: int, value: Any) -> Any:
     if value is None:
         return None
-    if var_type in (datatypes.OCTET_STRING, datatypes.DOMAIN):
+    if var_type in (datatypes.OCTET_STRING, datatypes.DOMAIN) and isinstance(value, bytes):
         return bytes.hex(value)
     elif var_type in (datatypes.VISIBLE_STRING, datatypes.UNICODE_STRING):
         return value
@@ -309,7 +320,10 @@ def build_variable(
             else:
                 var.min = int(min_string, 0)
         except ValueError:
-            pass
+            logger.warning(
+                "Invalid LowLimit %r for %s (0x%X), ignoring",
+                min_string, var.name, var.index,
+            )
     if eds.has_option(section, "HighLimit"):
         try:
             max_string = eds.get(section, "HighLimit")
@@ -318,38 +332,44 @@ def build_variable(
             else:
                 var.max = int(max_string, 0)
         except ValueError:
-            pass
+            logger.warning(
+                "Invalid HighLimit %r for %s (0x%X), ignoring",
+                max_string, var.name, var.index,
+            )
     if eds.has_option(section, "DefaultValue"):
         try:
             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, var.default_raw)
+            var.default = _decode_from_eds(node_id, var.data_type, var.default_raw)
         except ValueError:
-            pass
+            logger.warning(
+                "Invalid DefaultValue %r for %s (0x%X), ignoring",
+                var.default_raw, var.name, var.index,
+            )
     if eds.has_option(section, "ParameterValue"):
         try:
             var.value_raw = eds.get(section, "ParameterValue")
-            var.value = _convert_variable(node_id, var.data_type, var.value_raw)
+            var.value = _decode_from_eds(node_id, var.data_type, var.value_raw)
         except ValueError:
-            pass
+            logger.warning(
+                "Invalid ParameterValue %r for %s (0x%X), ignoring",
+                var.value_raw, var.name, var.index,
+            )
     # 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"))
         except ValueError:
-            pass
+            logger.warning(
+                "Invalid Factor %r for %s (0x%X), ignoring",
+                eds.get(section, "Factor"), var.name, var.index,
+            )
     if eds.has_option(section, "Description"):
-        try:
-            var.description = eds.get(section, "Description")
-        except ValueError:
-            pass
+        var.description = eds.get(section, "Description")
     if eds.has_option(section, "Unit"):
-        try:
-            var.unit = eds.get(section, "Unit")
-        except ValueError:
-            pass
+        var.unit = eds.get(section, "Unit")
     return var
 
 
@@ -400,15 +420,15 @@ def export_variable(var, eds):
         if getattr(var, 'default_raw', None) is not None:
             eds.set(section, "DefaultValue", var.default_raw)
         elif getattr(var, 'default', None) is not None:
-            eds.set(section, "DefaultValue", _revert_variable(
+            eds.set(section, "DefaultValue", _encode_to_eds(
                 var.data_type, var.default))
 
         if device_commisioning:
             if getattr(var, 'value_raw', None) is not None:
                 eds.set(section, "ParameterValue", var.value_raw)
             elif getattr(var, 'value', None) is not None:
                 eds.set(section, "ParameterValue",
-                        _revert_variable(var.data_type, var.value))
+                        _encode_to_eds(var.data_type, var.value))
 
         eds.set(section, "DataType", f"0x{var.data_type:04X}")
         eds.set(section, "PDOMapping", hex(var.pdo_mappable))

test/sample.eds

@@ -967,6 +967,24 @@ HighLimit=0x0A
 LowLimit=0x02
 PDOMapping=0
 
+[3022]
+ParameterName=UNSIGNED16 decimal limits
+ObjectType=0x7
+DataType=0x06
+AccessType=rw
+LowLimit=100
+HighLimit=1000
+PDOMapping=0
+
+[3023]
+ParameterName=INTEGER16 decimal limits
+ObjectType=0x7
+DataType=0x03
+AccessType=rw
+LowLimit=-100
+HighLimit=100
+PDOMapping=0
+
 [3030]
 ParameterName=INTEGER32 only negative values
 ObjectType=0x7
@@ -976,6 +994,42 @@ HighLimit=0xFFFFFFFF
 LowLimit=0x80000000
 PDOMapping=0
 
+[3031]
+ParameterName=INTEGER24 value range -1 to 0
+ObjectType=0x7
+DataType=0x10
+AccessType=rw
+HighLimit=0x000000
+LowLimit=0xFFFFFF
+PDOMapping=0
+
+[3032]
+ParameterName=INTEGER40 value range -1 to 0
+ObjectType=0x7
+DataType=0x12
+AccessType=rw
+HighLimit=0x0000000000
+LowLimit=0xFFFFFFFFFF
+PDOMapping=0
+
+[3033]
+ParameterName=INTEGER48 value range -1 to 0
+ObjectType=0x7
+DataType=0x13
+AccessType=rw
+HighLimit=0x000000000000
+LowLimit=0xFFFFFFFFFFFF
+PDOMapping=0
+
+[3034]
+ParameterName=INTEGER56 value range -1 to 0
+ObjectType=0x7
+DataType=0x14
+AccessType=rw
+HighLimit=0x00000000000000
+LowLimit=0xFFFFFFFFFFFFFF
+PDOMapping=0
+
 [3040]
 ParameterName=INTEGER64 value range -10 to +10
 ObjectType=0x7

test/test_eds.py

@@ -1,8 +1,10 @@
+import io
 import os
 import unittest
+from configparser import RawConfigParser
 
 import canopen
-from canopen.objectdictionary.eds import _signed_int_from_hex
+from canopen.objectdictionary.eds import _signed_int_from_hex, build_variable
 from canopen.utils import pretty_index
 
 from .util import DATATYPES_EDS, SAMPLE_EDS, tmp_file
@@ -70,8 +72,6 @@ def test_load_implicit_nodeid(self):
         self.assertEqual(od.node_id, 16)
 
     def test_load_implicit_nodeid_fallback(self):
-        import io
-
         # First, remove the NodeID option from DeviceComissioning.
         with open(SAMPLE_EDS) as f:
             lines = [L for L in f.readlines() if not L.startswith("NodeID=")]
@@ -93,8 +93,6 @@ def test_load_baudrate(self):
         self.assertEqual(od.bitrate, 500_000)
 
     def test_load_baudrate_fallback(self):
-        import io
-
         # Remove the Baudrate option.
         with open(SAMPLE_EDS) as f:
             lines = [L for L in f.readlines() if not L.startswith("Baudrate=")]
@@ -136,18 +134,23 @@ def test_record(self):
         self.assertFalse(var.is_domain)
 
     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)
+        cases = [
+            (0x3020, 0, 127),  # _          INTEGER8   hex limits
+            (0x3021, 2, 10),  # _           UNSIGNED8  hex limits
+            (0x3022, 100, 1000),  # _       UNSIGNED16 decimal limits
+            (0x3023, -100, 100),  # _       INTEGER16  decimal limits
+            (0x3030, -2147483648, -1),  # _ INTEGER32  hex limits
+            (0x3031, -1, 0),  # _           INTEGER24  hex limits
+            (0x3032, -1, 0),  # _           INTEGER40  hex limits
+            (0x3033, -1, 0),  # _           INTEGER48  hex limits
+            (0x3034, -1, 0),  # _           INTEGER56  hex limits
+            (0x3040, -10, +10),  # _        INTEGER64  hex limits
+        ]
+        for index, expected_min, expected_max in cases:
+            with self.subTest(index=f"0x{index:04X}"):
+                var = self.od[index]
+                self.assertEqual(var.min, expected_min)
+                self.assertEqual(var.max, expected_max)
 
     def test_signed_int_from_hex(self):
         for data_type, test_cases in self.test_data.items():
@@ -156,6 +159,37 @@ def test_signed_int_from_hex(self):
                     result = _signed_int_from_hex('0x' + test_case["hex_str"], test_case["bit_length"])
                     self.assertEqual(result, test_case["expected"])
 
+    def test_signed_int_from_hex_accepts_decimal(self):
+        # Negative decimal values are valid EDS literals (CiA 306 allows both formats).
+        self.assertEqual(_signed_int_from_hex("-1", 8), -1)
+        self.assertEqual(_signed_int_from_hex("-128", 8), -128)
+        self.assertEqual(_signed_int_from_hex("-2147483648", 32), -2147483648)
+
+    def test_signed_int_from_hex_rejects_out_of_range(self):
+        with self.assertRaises(ValueError):
+            _signed_int_from_hex("0xFFFF", 8)   # 16-bit value into 8-bit field
+        with self.assertRaises(ValueError):
+            _signed_int_from_hex("-129", 8)     # below minimum for 8-bit signed
+
+    def test_build_variable_range_warnings(self):
+        eds = RawConfigParser()
+        cases = [
+            ("2003", "LowLimit", str(-0xFFFF)),  # INTEGER16 < signed min
+            ("2003", "HighLimit", "0x10000"),  # INTEGER16 > unsigned max
+            ("2001", "DefaultValue", "SOMETHING"),  # BOOLEAN non-numeric
+            ("2003", "DefaultValue", "SOMETHING"),  # INTEGER16 non-numeric
+            ("2006", "ParameterValue", ""),  # UNSIGNED16 empty
+        ]
+        for index, option, value in cases:
+            with self.subTest(index=index, option=option, value=value):
+                # Fresh version for mutating temporarily
+                eds.clear()
+                eds.read(DATATYPES_EDS)
+                eds[index][option] = value
+                with self.assertLogs(level="WARN") as cm:
+                    build_variable(eds, index, node_id=42, object_type=7, index=int(index, 16))
+                self.assertRegex(cm.output[0], option)
+
     def test_array_compact_subobj(self):
         array = self.od[0x1003]
         self.assertIsInstance(array, canopen.objectdictionary.ODArray)
@@ -240,7 +274,6 @@ def test_read_domain_subobject(self):
 
     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'
@@ -251,6 +284,17 @@ def test_roundtrip_domain_objects(self):
         self.assertTrue(od2[0x3063].is_domain)
         self.assertTrue(od2[0x3064][1].is_domain)
 
+    def test_export_without_raw_default_values(self):
+        od = canopen.import_od(DATATYPES_EDS)
+        # Make sure the values are not cached in raw form
+        for var in od.values():
+            try:
+                delattr(var, 'default_raw')
+            except AttributeError:
+                pass
+        with io.StringIO() as dest:
+            canopen.export_od(od, dest, 'eds')
+
 
     def test_comments(self):
         self.assertEqual(self.od.comments,
@@ -271,7 +315,6 @@ def test_export_eds_to_file(self):
                         self.verify_od(dest, doctype)
 
     def test_export_eds_to_file_unknown_extension(self):
-        import io
         for suffix in ".txt", "":
             with tmp_file(suffix=suffix) as tmp:
                 dest = tmp.name
@@ -294,7 +337,6 @@ def test_export_eds_to_file_unknown_extension(self):
                         self.verify_od(buf, "eds")
 
     def test_export_eds_unknown_doctype(self):
-        import io
         filelike_object = io.StringIO()
         self.addCleanup(filelike_object.close)
         for dest in "filename", None, filelike_object:
@@ -307,7 +349,6 @@ def test_export_eds_unknown_doctype(self):
                         os.stat(dest)
 
     def test_export_eds_to_filelike_object(self):
-        import io
         for doctype in "eds", "dcf":
             with io.StringIO() as dest:
                 with self.subTest(dest=dest, doctype=doctype):
@@ -321,7 +362,6 @@ def test_export_eds_to_filelike_object(self):
 
     def test_export_eds_to_stdout(self):
         import contextlib
-        import io
         with contextlib.redirect_stdout(io.StringIO()) as f:
             ret = canopen.export_od(self.od, None, "eds")
         self.assertIsNone(ret)