fixed object_dictionary.write

Author: Günther Jena

CHANGELOG.md

@@ -1,4 +1,10 @@
+# 0.4.11
+
+* check limits for numeric datatypes when using `object_dictionary.write`
+* check if object is existing when using `object_dictionary.write`
+
 # 0.4.10
+
 * fix upload of objects with size 0 (use segmented transfer instead of expetited)
 
 # 0.4.9

requirements_dev.txt

@@ -1,6 +1,6 @@
-python-can==4.0.0
-pytest==6.2.4
-pytest-cov==2.12.1
+python-can==4.1.0
+pytest==7.2.2
+pytest-cov==4.0.0
 flake8==3.9.2
 mypy==0.960
 rstcheck==6.0.0

src/durand/datatypes.py

@@ -47,3 +47,18 @@ def is_float(datatype: DatatypeEnum):
     DatatypeEnum.OCTET_STRING: Struct("s"),
     DatatypeEnum.DOMAIN: Struct("s"),
 }
+
+
+range_dict = {
+    DatatypeEnum.BOOLEAN: (0, 1),
+    DatatypeEnum.UNSIGNED8: (0, (2**8) - 1),
+    DatatypeEnum.INTEGER8: (-(2**7), (2**7) - 1),
+    DatatypeEnum.UNSIGNED16: (0, (2**16) - 1),
+    DatatypeEnum.INTEGER16: (-(2**15), (2**15) - 1),
+    DatatypeEnum.UNSIGNED32: (0, (2**32) - 1),
+    DatatypeEnum.INTEGER32: (-(2**31), (2**31) - 1),
+    DatatypeEnum.UNSIGNED64: (0, (2**64) - 1),
+    DatatypeEnum.INTEGER64: (-(2**63), (2**63) - 1),
+    DatatypeEnum.REAL32: (float("-inf"), float("inf")),
+    DatatypeEnum.REAL64: (float("-inf"), float("inf")),
+}

src/durand/object_dictionary.py

@@ -4,7 +4,7 @@
 import itertools
 import logging
 
-from .datatypes import DatatypeEnum, struct_dict, is_numeric, is_float
+from .datatypes import DatatypeEnum, struct_dict, is_numeric, is_float, range_dict
 from .callback_handler import CallbackHandler, FailMode
 
 
@@ -27,15 +27,36 @@ class Variable:
     def __post_init__(self):
         if self.datatype not in DatatypeEnum:
             raise ValueError("Unsupported datatype")
+
         if self.access not in ("rw", "ro", "wo", "const"):
             raise ValueError("Invalid access type")
+
         if not is_numeric(self.datatype) and (
             self.maximum is not None or self.minimum is not None
         ):
             raise ValueError(
                 f"Minimum and Maximum not available with datatype {self.datatype!r}"
             )
 
+        if range_dict.get(self.datatype, None) is not None:
+            minimum, maximum = range_dict[self.datatype]
+
+            if self.minimum is not None and self.minimum < minimum:
+                raise ValueError(
+                    f"Specified minimum of {self.minimum} is lower than datatype supported minimum of {minimum}"
+                )
+
+            if self.maximum is not None and self.maximum > maximum:
+                raise ValueError(
+                    f"Specified maximum of {self.maximum} is higher than datatype supported maximum of {maximum}"
+                )
+
+            if self.minimum is None:
+                self.minimum = minimum
+
+            if self.maximum is None:
+                self.maximum = maximum
+
     @property
     def writable(self):
         return self.access in ("wo", "rw")
@@ -178,15 +199,27 @@ def __setitem__(self, index: int, obj: TObject):
             self._objects[index] = obj
 
     def lookup(self, index: int, subindex: int = None) -> TObject:
+        """Return object on index:subindex in object dictionary. When subindex is None
+        the object is returned. Otherwise a lookup is extended with subindex in the Array/Record.
+
+        :param index: index in object dictionary
+        :param subindex: optional subindex to lookup inside object
+        :returns: the Variable, Record or Array
+
+        :raises KeyError: when object not found in dictionary
+        """
         try:
-            return self._variables[index]
-        except KeyError:
+            if index in self._variables:
+                return self._variables[index]
+
             if subindex is None:
                 return self._objects[index]
 
             obj = self._objects[index]
             assert isinstance(obj, (Record, Array)), "Record or Array expected"
             return obj[subindex]
+        except KeyError:
+            raise KeyError("Object {index}:{subindex} not in object dictionary")
 
     def write(self, index: int, subindex: int, value: Any, downloaded: bool = False):
         """Write the given value to the according variable.
@@ -197,6 +230,9 @@ def write(self, index: int, subindex: int, value: Any, downloaded: bool = False)
         :param value: value to be written
         :param downloaded: flag is set, when the write is caused by an actual download
                            (instead of a internal value change)
+
+        :raises KeyError: when index:subindex not found
+        :raises Exception: when validate_callback fails
         """
         assert isinstance(
             value, (bytes, bool, int, float)
@@ -207,6 +243,19 @@ def write(self, index: int, subindex: int, value: Any, downloaded: bool = False)
         else:
             multiplexor = (index, subindex)
 
+        variable = self.lookup(*multiplexor)  # check if variable exists
+
+        if not downloaded:
+            if variable.minimum is not None and value < variable.minimum:
+                raise ValueError(
+                    f"Value {value} is too low (minimum is {variable.minimum})"
+                )
+
+            if variable.maximum is not None and value > variable.maximum:
+                raise ValueError(
+                    f"Value {value} is too high (minimum is {variable.maximum})"
+                )
+
         if multiplexor in self.validate_callbacks:
             self.validate_callbacks[multiplexor].call(value)  # may raises exception
 

tests/test_object_dictionary.py

@@ -1 +1,46 @@
+""" Testing object dictionary functionality """
+import re
+
 import pytest
+
+from durand import Node, Variable
+from durand.datatypes import DatatypeEnum as DT
+
+from .mock_network import MockNetwork
+
+
+def test_write():
+    # create the node
+    network = MockNetwork()
+    node = Node(network, node_id=2)
+
+    # add a variable with index 0x2000:0 to the object dictionary of the node
+    node.object_dictionary[0x2000] = Variable(DT.INTEGER16, "rw", value=5)
+    assert node.object_dictionary.read(0x2000, 0) == 5
+
+    # these test should pass
+    node.object_dictionary.write(0x2000, 0, 32767)
+    assert node.object_dictionary.read(0x2000, 0) == 32767
+
+    node.object_dictionary.write(0x2000, 0, -32768)
+    assert node.object_dictionary.read(0x2000, 0) == -32768
+    
+    # test too low value
+    with pytest.raises(ValueError, match=re.escape('Value -32769 is too low (minimum is -32768)')):
+        node.object_dictionary.write(0x2000, 0, value=-32769)
+
+    # test too high value
+    with pytest.raises(ValueError, match=re.escape('Value 32768 is too high (minimum is 32767)')):
+        node.object_dictionary.write(0x2000, 0, value=32768)
+
+
+def test_access_of_unknown_variable():
+    # create the node
+    network = MockNetwork()
+    node = Node(network, node_id=2)
+
+    with pytest.raises(KeyError):
+        node.object_dictionary.write(0x2000, 0, 5)
+    
+    with pytest.raises(KeyError):
+        node.object_dictionary.read(0x2000, 0)