From 81c6b00368d827ebddbcd2855af0b0f198eb8bde Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 08:17:02 +0200 Subject: [PATCH 1/7] Fix type hints on SDO classes. * Clarify that each SDO object has its corresponding OD entry linked in the .od attribute, not the whole ObjectDictionary. * __iter__() should return Iterator like in Mapping, not Iterable. * Use postponed evaluation of type annotations (PEP-563 / PEP-649 style) instead of stringized type annotations. --- canopen/sdo/base.py | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/canopen/sdo/base.py b/canopen/sdo/base.py index f9daebf5..6b2af2cb 100644 --- a/canopen/sdo/base.py +++ b/canopen/sdo/base.py @@ -1,9 +1,10 @@ +from __future__ import annotations + import binascii -from typing import Iterable, Optional, Union +from typing import Iterator, Optional, Union from collections.abc import Mapping from canopen import objectdictionary -from canopen.objectdictionary import ObjectDictionary from canopen import variable @@ -29,7 +30,7 @@ def __init__( self, rx_cobid: int, tx_cobid: int, - od: ObjectDictionary, + od: objectdictionary.ObjectDictionary, ): """ :param rx_cobid: @@ -46,7 +47,7 @@ def __init__( def __getitem__( self, index: Union[str, int] - ) -> Union["SdoVariable", "SdoArray", "SdoRecord"]: + ) -> Union[SdoVariable, SdoArray, SdoRecord]: entry = self.od[index] if isinstance(entry, objectdictionary.ODVariable): return SdoVariable(self, entry) @@ -55,7 +56,7 @@ def __getitem__( elif isinstance(entry, objectdictionary.ODRecord): return SdoRecord(self, entry) - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(self.od) def __len__(self) -> int: @@ -66,7 +67,7 @@ def __contains__(self, key: Union[int, str]) -> bool: def get_variable( self, index: Union[int, str], subindex: int = 0 - ) -> Optional["SdoVariable"]: + ) -> Optional[SdoVariable]: """Get the variable object at specified index (and subindex if applicable). :return: SdoVariable if found, else `None` @@ -92,17 +93,17 @@ def download( class SdoRecord(Mapping): - def __init__(self, sdo_node: SdoBase, od: ObjectDictionary): + def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODRecord): self.sdo_node = sdo_node self.od = od def __repr__(self) -> str: return f"<{type(self).__qualname__} {self.od.name!r} at 0x{self.od.index:04X}>" - def __getitem__(self, subindex: Union[int, str]) -> "SdoVariable": + def __getitem__(self, subindex: Union[int, str]) -> SdoVariable: return SdoVariable(self.sdo_node, self.od[subindex]) - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(self.od) def __len__(self) -> int: @@ -114,17 +115,17 @@ def __contains__(self, subindex: Union[int, str]) -> bool: class SdoArray(Mapping): - def __init__(self, sdo_node: SdoBase, od: ObjectDictionary): + def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODArray): self.sdo_node = sdo_node self.od = od def __repr__(self) -> str: return f"<{type(self).__qualname__} {self.od.name!r} at 0x{self.od.index:04X}>" - def __getitem__(self, subindex: Union[int, str]) -> "SdoVariable": + def __getitem__(self, subindex: Union[int, str]) -> SdoVariable: return SdoVariable(self.sdo_node, self.od[subindex]) - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(range(1, len(self) + 1)) def __len__(self) -> int: @@ -137,7 +138,7 @@ def __contains__(self, subindex: int) -> bool: class SdoVariable(variable.Variable): """Access object dictionary variable values using SDO protocol.""" - def __init__(self, sdo_node: SdoBase, od: ObjectDictionary): + def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable): self.sdo_node = sdo_node variable.Variable.__init__(self, od) From 9c033d0b2b3f54d95abda8cb4067b069811af466 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 09:15:36 +0200 Subject: [PATCH 2/7] Fix __iter__() return type annotation in PDO classes. --- canopen/pdo/base.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index 7312f660..5675bba0 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -1,6 +1,6 @@ import threading import math -from typing import Callable, Dict, Iterable, List, Optional, Union +from typing import Callable, Dict, Iterator, List, Optional, Union from collections.abc import Mapping import logging import binascii @@ -146,7 +146,7 @@ def __init__(self, com_offset, map_offset, pdo_node: PdoBase, cob_base=None): def __getitem__(self, key: int) -> "PdoMap": return self.maps[key] - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(self.maps) def __len__(self) -> int: @@ -229,7 +229,7 @@ def __getitem__(self, key: Union[int, str]) -> "PdoVariable": var = self.__getitem_by_name(key) return var - def __iter__(self) -> Iterable["PdoVariable"]: + def __iter__(self) -> Iterator["PdoVariable"]: return iter(self.map) def __len__(self) -> int: From 9880489701b6963c5895909e76c4d39c4dffc5d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 09:17:32 +0200 Subject: [PATCH 3/7] Fix __iter__() return type annotation in Network and OD classes. --- canopen/network.py | 4 ++-- canopen/objectdictionary/__init__.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/canopen/network.py b/canopen/network.py index 9e9a252e..1d152df7 100644 --- a/canopen/network.py +++ b/canopen/network.py @@ -1,7 +1,7 @@ from collections.abc import MutableMapping import logging import threading -from typing import Callable, Dict, Iterable, List, Optional, Union +from typing import Callable, Dict, Iterator, List, Optional, Union try: import can @@ -277,7 +277,7 @@ def __delitem__(self, node_id: int): self.nodes[node_id].remove_network() del self.nodes[node_id] - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(self.nodes) def __len__(self) -> int: diff --git a/canopen/objectdictionary/__init__.py b/canopen/objectdictionary/__init__.py index 19196951..1d0b921f 100644 --- a/canopen/objectdictionary/__init__.py +++ b/canopen/objectdictionary/__init__.py @@ -2,7 +2,7 @@ Object Dictionary module """ import struct -from typing import Dict, Iterable, List, Optional, TextIO, Union +from typing import Dict, Iterator, List, Optional, TextIO, Union from collections.abc import MutableMapping, Mapping import logging @@ -123,7 +123,7 @@ def __delitem__(self, index: Union[int, str]): del self.indices[obj.index] del self.names[obj.name] - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(sorted(self.indices)) def __len__(self) -> int: @@ -200,7 +200,7 @@ def __delitem__(self, subindex: Union[int, str]): def __len__(self) -> int: return len(self.subindices) - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(sorted(self.subindices)) def __contains__(self, subindex: Union[int, str]) -> bool: @@ -264,7 +264,7 @@ def __getitem__(self, subindex: Union[int, str]) -> "ODVariable": def __len__(self) -> int: return len(self.subindices) - def __iter__(self) -> Iterable[int]: + def __iter__(self) -> Iterator[int]: return iter(sorted(self.subindices)) def __eq__(self, other: "ODArray") -> bool: From 04c02e1bcd2a862de3aa9362af89db2662f1ab6a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 09:32:50 +0200 Subject: [PATCH 4/7] More __future__.annotations fixes. Use postponed evaluation of type annotations (PEP-563 / PEP-649 style) instead of stringized type annotations. --- canopen/network.py | 6 ++++-- canopen/objectdictionary/__init__.py | 30 +++++++++++++++------------- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/canopen/network.py b/canopen/network.py index 1d152df7..54280909 100644 --- a/canopen/network.py +++ b/canopen/network.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from collections.abc import MutableMapping import logging import threading @@ -82,7 +84,7 @@ def unsubscribe(self, can_id, callback=None) -> None: else: self.subscribers[can_id].remove(callback) - def connect(self, *args, **kwargs) -> "Network": + def connect(self, *args, **kwargs) -> Network: """Connect to CAN bus using python-can. Arguments are passed directly to :class:`can.BusABC`. Typically these @@ -214,7 +216,7 @@ def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None: def send_periodic( self, can_id: int, data: bytes, period: float, remote: bool = False - ) -> "PeriodicMessageTask": + ) -> PeriodicMessageTask: """Start sending a message periodically. :param can_id: diff --git a/canopen/objectdictionary/__init__.py b/canopen/objectdictionary/__init__.py index 1d0b921f..e962b367 100644 --- a/canopen/objectdictionary/__init__.py +++ b/canopen/objectdictionary/__init__.py @@ -1,6 +1,8 @@ """ Object Dictionary module """ +from __future__ import annotations + import struct from typing import Dict, Iterator, List, Optional, TextIO, Union from collections.abc import MutableMapping, Mapping @@ -12,7 +14,7 @@ logger = logging.getLogger(__name__) -def export_od(od, dest:Union[str,TextIO,None]=None, doc_type:Optional[str]=None): +def export_od(od, dest: Union[str, TextIO, None] = None, doc_type: Optional[str] = None): """ Export :class: ObjectDictionary to a file. :param od: @@ -54,7 +56,7 @@ def export_od(od, dest:Union[str,TextIO,None]=None, doc_type:Optional[str]=None) def import_od( source: Union[str, TextIO, None], node_id: Optional[int] = None, -) -> "ObjectDictionary": +) -> ObjectDictionary: """Parse an EDS, DCF, or EPF file. :param source: @@ -101,7 +103,7 @@ def __init__(self): def __getitem__( self, index: Union[int, str] - ) -> Union["ODArray", "ODRecord", "ODVariable"]: + ) -> Union[ODArray, ODRecord, ODVariable]: """Get object from object dictionary by name or index.""" item = self.names.get(index) or self.indices.get(index) if item is None: @@ -113,7 +115,7 @@ def __getitem__( return item def __setitem__( - self, index: Union[int, str], obj: Union["ODArray", "ODRecord", "ODVariable"] + self, index: Union[int, str], obj: Union[ODArray, ODRecord, ODVariable] ): assert index == obj.index or index == obj.name self.add_object(obj) @@ -132,7 +134,7 @@ def __len__(self) -> int: def __contains__(self, index: Union[int, str]): return index in self.names or index in self.indices - def add_object(self, obj: Union["ODArray", "ODRecord", "ODVariable"]) -> None: + def add_object(self, obj: Union[ODArray, ODRecord, ODVariable]) -> None: """Add object to the object dictionary. :param obj: @@ -147,7 +149,7 @@ def add_object(self, obj: Union["ODArray", "ODRecord", "ODVariable"]) -> None: def get_variable( self, index: Union[int, str], subindex: int = 0 - ) -> Optional["ODVariable"]: + ) -> Optional[ODVariable]: """Get the variable object at specified index (and subindex if applicable). :return: ODVariable if found, else `None` @@ -182,13 +184,13 @@ def __init__(self, name: str, index: int): def __repr__(self) -> str: return f"<{type(self).__qualname__} {self.name!r} at 0x{self.index:04X}>" - def __getitem__(self, subindex: Union[int, str]) -> "ODVariable": + def __getitem__(self, subindex: Union[int, str]) -> ODVariable: item = self.names.get(subindex) or self.subindices.get(subindex) if item is None: raise KeyError("Subindex %s was not found" % subindex) return item - def __setitem__(self, subindex: Union[int, str], var: "ODVariable"): + def __setitem__(self, subindex: Union[int, str], var: ODVariable): assert subindex == var.subindex self.add_member(var) @@ -206,10 +208,10 @@ def __iter__(self) -> Iterator[int]: def __contains__(self, subindex: Union[int, str]) -> bool: return subindex in self.names or subindex in self.subindices - def __eq__(self, other: "ODRecord") -> bool: + def __eq__(self, other: ODRecord) -> bool: return self.index == other.index - def add_member(self, variable: "ODVariable") -> None: + def add_member(self, variable: ODVariable) -> None: """Adds a :class:`~canopen.objectdictionary.ODVariable` to the record.""" variable.parent = self self.subindices[variable.subindex] = variable @@ -241,7 +243,7 @@ def __init__(self, name: str, index: int): def __repr__(self) -> str: return f"<{type(self).__qualname__} {self.name!r} at 0x{self.index:04X}>" - def __getitem__(self, subindex: Union[int, str]) -> "ODVariable": + def __getitem__(self, subindex: Union[int, str]) -> ODVariable: var = self.names.get(subindex) or self.subindices.get(subindex) if var is not None: # This subindex is defined @@ -267,10 +269,10 @@ def __len__(self) -> int: def __iter__(self) -> Iterator[int]: return iter(sorted(self.subindices)) - def __eq__(self, other: "ODArray") -> bool: + def __eq__(self, other: ODArray) -> bool: return self.index == other.index - def add_member(self, variable: "ODVariable") -> None: + def add_member(self, variable: ODVariable) -> None: """Adds a :class:`~canopen.objectdictionary.ODVariable` to the record.""" variable.parent = self self.subindices[variable.subindex] = variable @@ -348,7 +350,7 @@ def qualname(self) -> str: return f"{self.parent.name}.{self.name}" return self.name - def __eq__(self, other: "ODVariable") -> bool: + def __eq__(self, other: ODVariable) -> bool: return (self.index == other.index and self.subindex == other.subindex) From 9220fd7b3b41f07b90f0e5de49a6f4c54872e1b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 14:32:55 +0200 Subject: [PATCH 5/7] Annotate PdoVariable.pdo_parent type. --- canopen/pdo/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index 5675bba0..9fc4c9f9 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -540,7 +540,7 @@ class PdoVariable(variable.Variable): def __init__(self, od: objectdictionary.ODVariable): #: PDO object that is associated with this ODVariable Object - self.pdo_parent = None + self.pdo_parent: Optional[PdoMap] = None #: Location of variable in the message in bits self.offset = None self.length = len(od) From 2a6049ddabc18583986a3df820ac54e8c02bbe17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Colomb?= Date: Tue, 4 Jun 2024 15:40:51 +0200 Subject: [PATCH 6/7] Remove unnecessary variable initialization. PdoVariable.__getitem__() is annotated to not return None. In fact, every branch sets a valid PdoVariable return value, or raises an exception. Therefore the None default value is unnecessary and trips the type checker (mypy). --- canopen/pdo/base.py | 1 - 1 file changed, 1 deletion(-) diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index 9fc4c9f9..390fb941 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -215,7 +215,6 @@ def __getitem_by_name(self, value): value, ', '.join(valid_values))) def __getitem__(self, key: Union[int, str]) -> "PdoVariable": - var = None if isinstance(key, int): # there is a maximum available of 8 slots per PDO map if key in range(0, 8): From bda53bba53446ecb51e0958137f5151ca59658c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Frieder=20Sch=C3=BCler?= Date: Tue, 28 May 2024 10:27:44 +0200 Subject: [PATCH 7/7] More __future__.annotations fixes and type hints for PDO code. Use postponed evaluation of type annotations (PEP-563 / PEP-649 style) instead of stringized type annotations. Conditionally import types from other sub-packages to avoid dependency cycles. --- canopen/pdo/base.py | 37 ++++++++++++++++++++++--------------- 1 file changed, 22 insertions(+), 15 deletions(-) diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index 390fb941..5847bda0 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -1,6 +1,7 @@ +from __future__ import annotations import threading import math -from typing import Callable, Dict, Iterator, List, Optional, Union +from typing import Callable, Dict, Iterator, List, Optional, Union, TYPE_CHECKING from collections.abc import Mapping import logging import binascii @@ -9,6 +10,12 @@ from canopen import objectdictionary from canopen import variable +if TYPE_CHECKING: + from canopen.network import Network + from canopen import LocalNode, RemoteNode + from canopen.pdo import RPDO, TPDO + from canopen.sdo import SdoRecord + PDO_NOT_VALID = 1 << 31 RTR_NOT_ALLOWED = 1 << 30 @@ -22,10 +29,10 @@ class PdoBase(Mapping): Parent object associated with this PDO instance """ - def __init__(self, node): - self.network = None - self.map = None # instance of PdoMaps - self.node = node + def __init__(self, node: Union[LocalNode, RemoteNode]): + self.network: Optional[Network] = None + self.map: Optional[PdoMaps] = None + self.node: Union[LocalNode, RemoteNode] = node def __iter__(self): return iter(self.map) @@ -131,7 +138,7 @@ def __init__(self, com_offset, map_offset, pdo_node: PdoBase, cob_base=None): :param pdo_node: :param cob_base: """ - self.maps: Dict[int, "PdoMap"] = {} + self.maps: Dict[int, PdoMap] = {} for map_no in range(512): if com_offset + map_no in pdo_node.node.object_dictionary: new_map = PdoMap( @@ -143,7 +150,7 @@ def __init__(self, com_offset, map_offset, pdo_node: PdoBase, cob_base=None): new_map.predefined_cob_id = cob_base + map_no * 0x100 + pdo_node.node.id self.maps[map_no + 1] = new_map - def __getitem__(self, key: int) -> "PdoMap": + def __getitem__(self, key: int) -> PdoMap: return self.maps[key] def __iter__(self) -> Iterator[int]: @@ -157,9 +164,9 @@ class PdoMap: """One message which can have up to 8 bytes of variables mapped.""" def __init__(self, pdo_node, com_record, map_array): - self.pdo_node = pdo_node - self.com_record = com_record - self.map_array = map_array + self.pdo_node: Union[TPDO, RPDO] = pdo_node + self.com_record: SdoRecord = com_record + self.map_array: SdoRecord = map_array #: If this map is valid self.enabled: bool = False #: COB-ID for this PDO @@ -177,7 +184,7 @@ def __init__(self, pdo_node, com_record, map_array): #: Ignores SYNC objects up to this SYNC counter value (optional) self.sync_start_value: Optional[int] = None #: List of variables mapped to this PDO - self.map: List["PdoVariable"] = [] + self.map: List[PdoVariable] = [] self.length: int = 0 #: Current message data self.data = bytearray() @@ -214,7 +221,7 @@ def __getitem_by_name(self, value): raise KeyError('{0} not found in map. Valid entries are {1}'.format( value, ', '.join(valid_values))) - def __getitem__(self, key: Union[int, str]) -> "PdoVariable": + def __getitem__(self, key: Union[int, str]) -> PdoVariable: if isinstance(key, int): # there is a maximum available of 8 slots per PDO map if key in range(0, 8): @@ -228,7 +235,7 @@ def __getitem__(self, key: Union[int, str]) -> "PdoVariable": var = self.__getitem_by_name(key) return var - def __iter__(self) -> Iterator["PdoVariable"]: + def __iter__(self) -> Iterator[PdoVariable]: return iter(self.map) def __len__(self) -> int: @@ -303,7 +310,7 @@ def on_message(self, can_id, data, timestamp): for callback in self.callbacks: callback(self) - def add_callback(self, callback: Callable[["PdoMap"], None]) -> None: + def add_callback(self, callback: Callable[[PdoMap], None]) -> None: """Add a callback which will be called on receive. :param callback: @@ -446,7 +453,7 @@ def add_variable( index: Union[str, int], subindex: Union[str, int] = 0, length: Optional[int] = None, - ) -> "PdoVariable": + ) -> PdoVariable: """Add a variable from object dictionary as the next entry. :param index: Index of variable as name or number