Merge branch 'master' into fix/mypy-type-fixes

Author: acolomb

canopen/lss.py

@@ -85,7 +85,7 @@ def __init__(self) -> None:
         self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK
         self._node_id = 0
         self._data = None
-        self.responses = queue.Queue()
+        self.responses: queue.Queue[bytes] = queue.Queue()
 
     def send_switch_state_global(self, mode):
         """switch mode to CONFIGURATION_STATE or WAITING_STATE

canopen/node/local.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Callable
 from typing import Union
 
 import canopen.network
@@ -26,8 +27,8 @@ def __init__(
         super(LocalNode, self).__init__(node_id, object_dictionary)
 
         self.data_store: dict[int, dict[int, bytes]] = {}
-        self._read_callbacks = []
-        self._write_callbacks = []
+        self._read_callbacks: list[Callable] = []
+        self._write_callbacks: list[Callable] = []
 
         self.sdo = SdoServer(0x600 + self.id, 0x580 + self.id, self)
         self.tpdo = TPDO(self)
@@ -62,10 +63,10 @@ def remove_network(self) -> None:
         self.nmt.network = canopen.network._UNINITIALIZED_NETWORK
         self.emcy.network = canopen.network._UNINITIALIZED_NETWORK
 
-    def add_read_callback(self, callback):
+    def add_read_callback(self, callback: Callable):
         self._read_callbacks.append(callback)
 
-    def add_write_callback(self, callback):
+    def add_write_callback(self, callback: Callable):
         self._write_callbacks.append(callback)
 
     def get_data(

canopen/node/remote.py

@@ -39,7 +39,7 @@ def __init__(
         #: Enable WORKAROUND for reversed PDO mapping entries
         self.curtis_hack = False
 
-        self.sdo_channels = []
+        self.sdo_channels: list[SdoClient] = []
         self.sdo = self.add_sdo(0x600 + self.id, 0x580 + self.id)
         self.tpdo = TPDO(self)
         self.rpdo = RPDO(self)

canopen/objectdictionary/__init__.py

@@ -163,7 +163,7 @@ def __iter__(self) -> Iterator[int]:
     def __len__(self) -> int:
         return len(self.indices)
 
-    def __contains__(self, index: Union[int, str]):
+    def __contains__(self, index: object) -> bool:
         return index in self.names or index in self.indices
 
     def add_object(self, obj: Union[ODArray, ODRecord, ODVariable]) -> None:
@@ -211,8 +211,8 @@ def __init__(self, name: str, index: int):
         self.name = name
         #: Storage location of index
         self.storage_location = None
-        self.subindices = {}
-        self.names = {}
+        self.subindices: dict[int, ODVariable] = {}
+        self.names: dict[str, ODVariable] = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -238,10 +238,12 @@ def __len__(self) -> int:
     def __iter__(self) -> Iterator[int]:
         return iter(sorted(self.subindices))
 
-    def __contains__(self, subindex: Union[int, str]) -> bool:
+    def __contains__(self, subindex: object) -> bool:
         return subindex in self.names or subindex in self.subindices
 
-    def __eq__(self, other: ODRecord) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODRecord):
+            return NotImplemented
         return self.index == other.index
 
     def add_member(self, variable: ODVariable) -> None:
@@ -270,8 +272,8 @@ def __init__(self, name: str, index: int):
         self.name = name
         #: Storage location of index
         self.storage_location = None
-        self.subindices = {}
-        self.names = {}
+        self.subindices: dict[int, ODVariable] = {}
+        self.names: dict[str, ODVariable] = {}
 
     def __repr__(self) -> str:
         return f"<{type(self).__qualname__} {self.name!r} at {pretty_index(self.index)}>"
@@ -302,7 +304,9 @@ def __len__(self) -> int:
     def __iter__(self) -> Iterator[int]:
         return iter(sorted(self.subindices))
 
-    def __eq__(self, other: ODArray) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODArray):
+            return NotImplemented
         return self.index == other.index
 
     def add_member(self, variable: ODVariable) -> None:
@@ -391,7 +395,9 @@ def qualname(self) -> str:
             return f"{self.parent.name}.{self.name}"
         return self.name
 
-    def __eq__(self, other: ODVariable) -> bool:
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ODVariable):
+            return NotImplemented
         return (self.index == other.index and
                 self.subindex == other.subindex)
 
@@ -417,7 +423,7 @@ def add_value_description(self, value: int, descr: str) -> None:
         """
         self.value_descriptions[value] = descr
 
-    def add_bit_definition(self, name: str, bits: List[int]) -> None:
+    def add_bit_definition(self, name: str, bits: list[int]) -> None:
         """Associate bit(s) with a string description.
 
         :param name: Name of bit(s)
@@ -511,7 +517,7 @@ def encode_desc(self, desc: str) -> int:
         raise ValueError(
             f"No value corresponds to '{desc}'. Valid values are: {valid_values}")
 
-    def decode_bits(self, value: int, bits: List[int]) -> int:
+    def decode_bits(self, value: int, bits: list[int]) -> int:
         try:
             bits = self.bit_definitions[bits]
         except (TypeError, KeyError):
@@ -521,7 +527,7 @@ def decode_bits(self, value: int, bits: List[int]) -> int:
             mask |= 1 << bit
         return (value & mask) >> min(bits)
 
-    def encode_bits(self, original_value: int, bits: List[int], bit_value: int):
+    def encode_bits(self, original_value: int, bits: list[int], bit_value: int):
         try:
             bits = self.bit_definitions[bits]
         except (TypeError, KeyError):

canopen/pdo/base.py

@@ -98,7 +98,7 @@ def export(self, filename):
         :rtype: canmatrix.canmatrix.CanMatrix
         """
         try:
-            from canmatrix import canmatrix
+            from canmatrix import canmatrix  # type:ignore # (typing still in progress)
             from canmatrix import formats
         except ImportError:
             raise NotImplementedError("This feature requires the 'canopen[db_export]' feature")

canopen/sdo/base.py

@@ -64,7 +64,7 @@ def __iter__(self) -> Iterator[int]:
     def __len__(self) -> int:
         return len(self.od)
 
-    def __contains__(self, key: Union[int, str]) -> bool:
+    def __contains__(self, key: object) -> bool:
         return key in self.od
 
     def get_variable(
@@ -114,7 +114,7 @@ def __len__(self) -> int:
         # Skip the "highest subindex" entry, which is not part of the data
         return len(self.od) - int(0 in self.od)
 
-    def __contains__(self, subindex: Union[int, str]) -> bool:
+    def __contains__(self, subindex: object) -> bool:
         return subindex in self.od
 
 
@@ -137,7 +137,9 @@ def __iter__(self) -> Iterator[int]:
     def __len__(self) -> int:
         return self[0].raw  # type: ignore[return-value]
 
-    def __contains__(self, subindex: int) -> bool:
+    def __contains__(self, subindex: object) -> bool:
+        if not isinstance(subindex, int):
+            return False
         return 0 <= subindex <= len(self)
 
 

canopen/variable.py

@@ -135,6 +135,7 @@ def read(self, fmt: str = "raw") -> Union[int, bool, float, str, bytes]:
 
         :returns:
             The value of the variable.
+        :raises ValueError: For unsupported fmt values.
         """
         if fmt == "raw":
             return self.raw

test/test_od.py

@@ -288,61 +288,22 @@ def test_subindexes(self):
         self.assertEqual(array[3].name, "Test Variable_3")
 
 
-class _StubVariable(Variable):
-    """Minimal concrete Variable for testing read/write/bits."""
+class TestEquality(unittest.TestCase):
 
-    def __init__(self, od_var):
-        super().__init__(od_var)
-        self._data = od_var.encode_raw(od_var.default)
-
-    def get_data(self):
-        return self._data
-
-    def set_data(self, data):
-        self._data = data
-
-
-class TestVariable(unittest.TestCase):
-
-    def test_read_invalid_format(self):
-        var = od.ODVariable("Test UNSIGNED8", 0x1000)
-        var.data_type = od.UNSIGNED8
-        var.default = 0
-        v = _StubVariable(var)
-        with self.assertRaises(ValueError):
-            v.read(fmt="invalid")
+    def test_record_eq_wrong_type(self):
+        record = od.ODRecord("Test Record", 0x1001)
+        self.assertNotEqual(record, "not a record")
+        self.assertNotEqual(record, 42)
 
-    def test_write_desc(self):
-        var = od.ODVariable("Test UNSIGNED8", 0x1000)
-        var.data_type = od.UNSIGNED8
-        var.default = 0
-        var.add_value_description(0, "Off")
-        var.add_value_description(1, "On")
-        v = _StubVariable(var)
-        v.write("On", fmt="desc")
-        self.assertEqual(v.raw, 1)
-
-    def test_raw_with_string_value(self):
-        var = od.ODVariable("Test VISIBLE_STRING", 0x1000)
-        var.data_type = od.VISIBLE_STRING
-        var.default = "hello"
-        var.add_value_description(0, "Off")
-        v = _StubVariable(var)
-        # String value must not be looked up in value_descriptions
-        self.assertEqual(v.raw, "hello")
+    def test_array_eq_wrong_type(self):
+        array = od.ODArray("Test Array", 0x1002)
+        self.assertNotEqual(array, "not an array")
+        self.assertNotEqual(array, 42)
 
-    def test_bits(self):
-        var = od.ODVariable("Test UNSIGNED8", 0x1000)
-        var.data_type = od.UNSIGNED8
-        var.default = 0
-        var.add_bit_definition("BIT 0", [0])
-        var.add_bit_definition("BIT 2 and 3", [2, 3])
-        v = _StubVariable(var)
-        v.raw = 5
-        bits = v.bits
-        self.assertEqual(bits[0], 1)
-        bits[0] = 0
-        self.assertEqual(v.raw, 4)
+    def test_variable_eq_wrong_type(self):
+        var = od.ODVariable("Test Variable", 0x1000, 0)
+        self.assertNotEqual(var, "not a variable")
+        self.assertNotEqual(var, 42)
 
 
 if __name__ == "__main__":

test/test_sdo.py

@@ -48,6 +48,12 @@ def test_array_members_dynamic(self):
         for var in array.values():
             self.assertIsInstance(var, canopen.sdo.SdoVariable)
 
+    def test_array_contains_non_int(self):
+        """SdoArray.__contains__ should handle non-int types gracefully."""
+        array = self.sdo_node[0x1003]
+        self.assertNotIn("not an int", array)
+        self.assertNotIn(None, array)
+
     def test_get_variable_not_found(self):
         self.assertIsNone(self.sdo_node.get_variable(0x9999))
 

test/test_variable.py

@@ -0,0 +1,65 @@
+import unittest
+
+from canopen import objectdictionary as od
+from canopen.variable import Variable
+
+
+class _StubVariable(Variable):
+    """Minimal concrete Variable for testing read/write/bits."""
+
+    def __init__(self, od_var):
+        super().__init__(od_var)
+        self._data = od_var.encode_raw(od_var.default)
+
+    def get_data(self):
+        return self._data
+
+    def set_data(self, data):
+        self._data = data
+
+
+class TestVariable(unittest.TestCase):
+
+    def test_read_invalid_format(self):
+        var = od.ODVariable("Test UNSIGNED8", 0x1000)
+        var.data_type = od.UNSIGNED8
+        var.default = 0
+        v = _StubVariable(var)
+        with self.assertRaises(ValueError):
+            v.read(fmt="invalid")
+
+    def test_write_desc(self):
+        var = od.ODVariable("Test UNSIGNED8", 0x1000)
+        var.data_type = od.UNSIGNED8
+        var.default = 0
+        var.add_value_description(0, "Off")
+        var.add_value_description(1, "On")
+        v = _StubVariable(var)
+        v.write("On", fmt="desc")
+        self.assertEqual(v.raw, 1)
+
+    def test_raw_with_string_value(self):
+        var = od.ODVariable("Test VISIBLE_STRING", 0x1000)
+        var.data_type = od.VISIBLE_STRING
+        var.default = "hello"
+        var.add_value_description(0, "Off")
+        v = _StubVariable(var)
+        # String value must not be looked up in value_descriptions
+        self.assertEqual(v.raw, "hello")
+
+    def test_bits(self):
+        var = od.ODVariable("Test UNSIGNED8", 0x1000)
+        var.data_type = od.UNSIGNED8
+        var.default = 0
+        var.add_bit_definition("BIT 0", [0])
+        var.add_bit_definition("BIT 2 and 3", [2, 3])
+        v = _StubVariable(var)
+        v.raw = 5
+        bits = v.bits
+        self.assertEqual(bits[0], 1)
+        bits[0] = 0
+        self.assertEqual(v.raw, 4)
+
+
+if __name__ == "__main__":
+    unittest.main()