diff --git a/canopen/emcy.py b/canopen/emcy.py index ec2c489f..ed642a9a 100644 --- a/canopen/emcy.py +++ b/canopen/emcy.py @@ -1,3 +1,4 @@ +from __future__ import annotations import logging import struct import threading @@ -55,7 +56,7 @@ def reset(self): def wait( self, emcy_code: Optional[int] = None, timeout: float = 10 - ) -> "EmcyError": + ) -> Optional[EmcyError]: """Wait for a new EMCY to arrive. :param emcy_code: EMCY code to wait for diff --git a/canopen/lss.py b/canopen/lss.py index 7c0b92a6..311f77b5 100644 --- a/canopen/lss.py +++ b/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 diff --git a/canopen/network.py b/canopen/network.py index 02bec899..80a9ef5e 100644 --- a/canopen/network.py +++ b/canopen/network.py @@ -3,7 +3,7 @@ import logging import threading from collections.abc import MutableMapping -from typing import Callable, Dict, Final, Iterator, List, Optional, Union +from typing import Callable, Dict, Final, Iterator, List, Optional, Union, TYPE_CHECKING, TextIO import can from can import Listener @@ -16,6 +16,8 @@ from canopen.sync import SyncProducer from canopen.timestamp import TimeProducer +if TYPE_CHECKING: + from can.typechecking import CanData logger = logging.getLogger(__name__) @@ -134,15 +136,15 @@ def __exit__(self, type, value, traceback): def add_node( self, - node: Union[int, RemoteNode, LocalNode], - object_dictionary: Union[str, ObjectDictionary, None] = None, + node: Union[int, RemoteNode], + object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None, upload_eds: bool = False, ) -> RemoteNode: """Add a remote node to the network. :param node: Can be either an integer representing the node ID, a - :class:`canopen.RemoteNode` or :class:`canopen.LocalNode` object. + :class:`canopen.RemoteNode` object. :param object_dictionary: Can be either a string for specifying the path to an Object Dictionary file or a @@ -157,14 +159,16 @@ def add_node( if upload_eds: logger.info("Trying to read EDS from node %d", node) object_dictionary = import_from_node(node, self) - node = RemoteNode(node, object_dictionary) - self[node.id] = node - return node + nodeobj = RemoteNode(node, object_dictionary) + else: + nodeobj = node + self[nodeobj.id] = nodeobj + return nodeobj def create_node( self, - node: int, - object_dictionary: Union[str, ObjectDictionary, None] = None, + node: Union[int, LocalNode], + object_dictionary: Union[str, ObjectDictionary, TextIO, None] = None, ) -> LocalNode: """Create a local node in the network. @@ -179,11 +183,13 @@ def create_node( The Node object that was added. """ if isinstance(node, int): - node = LocalNode(node, object_dictionary) - self[node.id] = node - return node + nodeobj = LocalNode(node, object_dictionary) + else: + nodeobj = node + self[nodeobj.id] = nodeobj + return nodeobj - def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None: + def send_message(self, can_id: int, data: CanData, remote: bool = False) -> None: """Send a raw CAN message to the network. This method may be overridden in a subclass if you need to integrate @@ -211,7 +217,7 @@ def send_message(self, can_id: int, data: bytes, remote: bool = False) -> None: self.check() def send_periodic( - self, can_id: int, data: bytes, period: float, remote: bool = False + self, can_id: int, data: CanData, period: float, remote: bool = False ) -> PeriodicMessageTask: """Start sending a message periodically. @@ -227,6 +233,8 @@ def send_periodic( :return: An task object with a ``.stop()`` method to stop the transmission """ + if not self.bus: + raise RuntimeError("Not connected to CAN bus") return PeriodicMessageTask(can_id, data, period, self.bus, remote) def notify(self, can_id: int, data: bytearray, timestamp: float) -> None: @@ -306,9 +314,9 @@ class PeriodicMessageTask: def __init__( self, can_id: int, - data: bytes, + data: CanData, period: float, - bus, + bus: can.BusABC, remote: bool = False, ): """ diff --git a/canopen/nmt.py b/canopen/nmt.py index 622e0a33..3b0bc8bf 100644 --- a/canopen/nmt.py +++ b/canopen/nmt.py @@ -2,7 +2,7 @@ import struct import threading import time -from typing import Callable, Optional, TYPE_CHECKING +from typing import Callable, Optional, TYPE_CHECKING, List import canopen.network @@ -117,7 +117,7 @@ def __init__(self, node_id: int): #: Timestamp of last heartbeat message self.timestamp: Optional[float] = None self.state_update = threading.Condition() - self._callbacks = [] + self._callbacks: List[Callable[[int], None]] = [] def on_heartbeat(self, can_id, data, timestamp): with self.state_update: @@ -187,7 +187,7 @@ def start_node_guarding(self, period: float): Period (in seconds) at which the node guarding should be advertised to the slave node. """ if self._node_guarding_producer : self.stop_node_guarding() - self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, None, period, True) + self._node_guarding_producer = self.network.send_periodic(0x700 + self.id, [], period, True) def stop_node_guarding(self): """Stops the node guarding mechanism.""" diff --git a/canopen/node/base.py b/canopen/node/base.py index 45ad35b4..3d853228 100644 --- a/canopen/node/base.py +++ b/canopen/node/base.py @@ -1,4 +1,4 @@ -from typing import TextIO, Union +from typing import TextIO, Union, Optional import canopen.network from canopen.objectdictionary import ObjectDictionary, import_od @@ -16,8 +16,8 @@ class BaseNode: def __init__( self, - node_id: int, - object_dictionary: Union[ObjectDictionary, str, TextIO], + node_id: Optional[int], + object_dictionary: Union[ObjectDictionary, str, TextIO, None], ): self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK @@ -25,7 +25,10 @@ def __init__( object_dictionary = import_od(object_dictionary, node_id) self.object_dictionary = object_dictionary - self.id = node_id or self.object_dictionary.node_id + node_id = node_id or self.object_dictionary.node_id + if node_id is None: + raise ValueError("Node ID must be specified") + self.id: int = node_id def has_network(self) -> bool: """Check whether the node has been associated to a network.""" diff --git a/canopen/node/local.py b/canopen/node/local.py index 8f2493d9..917782e0 100644 --- a/canopen/node/local.py +++ b/canopen/node/local.py @@ -1,14 +1,14 @@ from __future__ import annotations import logging -from typing import Dict, Union +from typing import Dict, Union, List, Protocol, TextIO, Optional import canopen.network from canopen import objectdictionary from canopen.emcy import EmcyProducer from canopen.nmt import NmtSlave from canopen.node.base import BaseNode -from canopen.objectdictionary import ObjectDictionary +from canopen.objectdictionary import ObjectDictionary, ODVariable from canopen.pdo import PDO, RPDO, TPDO from canopen.sdo import SdoAbortedError, SdoServer @@ -16,18 +16,34 @@ logger = logging.getLogger(__name__) +class WriteCallback(Protocol): + """LocalNode Write Callback Protocol""" + def __call__(self, *, index: int, subindex: int, + od: ODVariable, + data: bytes) -> None: + ''' Write Callback ''' + + +class ReadCallback(Protocol): + """LocalNode Read Callback Protocol""" + def __call__(self, *, index: int, subindex: int, + od: ODVariable + ) -> Union[bool, int, float, str, bytes, None]: + ''' Read Callback ''' + + class LocalNode(BaseNode): def __init__( self, - node_id: int, - object_dictionary: Union[ObjectDictionary, str], + node_id: Optional[int], + object_dictionary: Union[ObjectDictionary, str, TextIO, None], ): 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[ReadCallback] = [] + self._write_callbacks: List[WriteCallback] = [] self.sdo = SdoServer(0x600 + self.id, 0x580 + self.id, self) self.tpdo = TPDO(self) diff --git a/canopen/node/remote.py b/canopen/node/remote.py index 226c0c0f..7f45abe5 100644 --- a/canopen/node/remote.py +++ b/canopen/node/remote.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from typing import TextIO, Union +from typing import TextIO, Union, Optional, List import canopen.network from canopen.emcy import EmcyConsumer @@ -30,8 +30,8 @@ class RemoteNode(BaseNode): def __init__( self, - node_id: int, - object_dictionary: Union[ObjectDictionary, str, TextIO], + node_id: Optional[int], + object_dictionary: Union[ObjectDictionary, str, TextIO, None], load_od: bool = False, ): super(RemoteNode, self).__init__(node_id, object_dictionary) @@ -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) diff --git a/canopen/objectdictionary/__init__.py b/canopen/objectdictionary/__init__.py index 09fe6e03..cea37890 100644 --- a/canopen/objectdictionary/__init__.py +++ b/canopen/objectdictionary/__init__.py @@ -7,7 +7,7 @@ import logging import struct from collections.abc import Mapping, MutableMapping -from typing import Dict, Iterator, List, Optional, TextIO, Union +from typing import Dict, Iterator, List, Optional, TextIO, Union, cast from canopen.objectdictionary.datatypes import * from canopen.objectdictionary.datatypes import IntegerN, UnsignedN @@ -69,7 +69,7 @@ def export_od( finally: # If dest is opened in this fn, it should be closed if opened_here: - dest.close() + cast(TextIO, dest).close() # The cast is needed to help the type checker def import_od( @@ -92,7 +92,7 @@ def import_od( return ObjectDictionary() if hasattr(source, "read"): # File like object - filename = source.name + filename = cast(TextIO, source).name elif hasattr(source, "tag"): # XML tree, probably from an EPF file filename = "od.epf" @@ -137,7 +137,9 @@ def __getitem__( if item is None: if isinstance(index, str) and '.' in index: idx, sub = index.split('.', maxsplit=1) - return self[idx][sub] + var = self[idx] + if not isinstance(var, ODVariable): + return var[sub] raise KeyError(f"{pretty_index(index)} was not found in Object Dictionary") return item @@ -158,7 +160,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): return index in self.names or index in self.indices def add_object(self, obj: Union[ODArray, ODRecord, ODVariable]) -> None: @@ -186,6 +188,7 @@ def get_variable( return obj elif isinstance(obj, (ODRecord, ODArray)): return obj.get(subindex) + return None class ODRecord(MutableMapping): @@ -205,14 +208,17 @@ 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)}>" def __getitem__(self, subindex: Union[int, str]) -> ODVariable: - item = self.names.get(subindex) or self.subindices.get(subindex) + if isinstance(subindex, str): + item = self.names.get(subindex) + else: + item = self.subindices.get(subindex) if item is None: raise KeyError(f"Subindex {pretty_index(None, subindex)} was not found") return item @@ -232,11 +238,11 @@ 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: - return self.index == other.index + def __eq__(self, other: object) -> bool: + return isinstance(other, type(self)) and self.index == other.index def add_member(self, variable: ODVariable) -> None: """Adds a :class:`~canopen.objectdictionary.ODVariable` to the record.""" @@ -264,14 +270,17 @@ 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)}>" def __getitem__(self, subindex: Union[int, str]) -> ODVariable: - var = self.names.get(subindex) or self.subindices.get(subindex) + if isinstance(subindex, str): + var = self.names.get(subindex) + else: + var = self.subindices.get(subindex) if var is not None: # This subindex is defined pass @@ -296,8 +305,8 @@ def __len__(self) -> int: def __iter__(self) -> Iterator[int]: return iter(sorted(self.subindices)) - def __eq__(self, other: ODArray) -> bool: - return self.index == other.index + def __eq__(self, other: object) -> bool: + return isinstance(other, type(self)) and self.index == other.index def add_member(self, variable: ODVariable) -> None: """Adds a :class:`~canopen.objectdictionary.ODVariable` to the record.""" @@ -337,7 +346,7 @@ def __init__(self, name: str, index: int, subindex: int = 0): #: The :class:`~canopen.ObjectDictionary`, #: :class:`~canopen.objectdictionary.ODRecord` or #: :class:`~canopen.objectdictionary.ODArray` owning the variable - self.parent = None + self.parent: Union[ObjectDictionary, ODArray, ODRecord, None] = None #: 16-bit address of the object in the dictionary self.index = index #: 8-bit sub-index of the object in the dictionary @@ -385,8 +394,9 @@ def qualname(self) -> str: return f"{self.parent.name}.{self.name}" return self.name - def __eq__(self, other: ODVariable) -> bool: - return (self.index == other.index and + def __eq__(self, other: object) -> bool: + return (isinstance(other, type(self)) and + self.index == other.index and self.subindex == other.subindex) def __len__(self) -> int: @@ -443,12 +453,21 @@ def encode_raw(self, value: Union[int, float, str, bytes, bytearray]) -> bytes: if isinstance(value, (bytes, bytearray)): return value elif self.data_type == VISIBLE_STRING: + if not isinstance(value, str): + raise TypeError(f"Value of type {type(value)!r} doesn't match VISIBLE_STRING") return value.encode("ascii") elif self.data_type == UNICODE_STRING: + if not isinstance(value, str): + raise TypeError(f"Value of type {type(value)!r} doesn't match UNICODE_STRING") return value.encode("utf_16_le") elif self.data_type in (DOMAIN, OCTET_STRING): + if not isinstance(value, (bytes, bytearray)): + t = "DOMAIN" if self.data_type == DOMAIN else "OCTET_STRING" + raise TypeError(f"Value of type {type(value)!r} doesn't match {t}") return bytes(value) elif self.data_type in self.STRUCT_TYPES: + if not isinstance(value, (bool, int, float)): + raise TypeError(f"Value of type {type(value)!r} is unexpected for numeric types") if self.data_type in INTEGER_TYPES: value = int(value) if self.data_type in NUMBER_TYPES: @@ -469,13 +488,17 @@ def encode_raw(self, value: Union[int, float, str, bytes, bytearray]) -> bytes: raise TypeError( f"Do not know how to encode {value!r} to data type 0x{self.data_type:X}") - def decode_phys(self, value: int) -> Union[int, bool, float, str, bytes]: + def decode_phys(self, value: Union[int, bool, float, str, bytes]) -> Union[int, bool, float, str, bytes]: if self.data_type in INTEGER_TYPES: + if not isinstance(value, (int, float)): + raise TypeError(f"Value of type {type(value)!r} is unexpected for numeric types") value *= self.factor return value - def encode_phys(self, value: Union[int, bool, float, str, bytes]) -> int: + def encode_phys(self, value: Union[int, bool, float, str, bytes]) -> Union[int, bool, float, str, bytes]: if self.data_type in INTEGER_TYPES: + if not isinstance(value, (int, float)): + raise TypeError(f"Value of type {type(value)!r} is unexpected for numeric types") value /= self.factor value = int(round(value)) return value @@ -500,27 +523,29 @@ 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: Union[range, str, List[int]]) -> int: try: - bits = self.bit_definitions[bits] + bits = self.bit_definitions[cast(str, bits)] except (TypeError, KeyError): pass mask = 0 - for bit in bits: + lbits = cast(List[int], bits) + for bit in lbits: mask |= 1 << bit - return (value & mask) >> min(bits) + return (value & mask) >> min(lbits) - def encode_bits(self, original_value: int, bits: List[int], bit_value: int): + def encode_bits(self, original_value: int, bits: Union[range, str, List[int]], bit_value: int): try: - bits = self.bit_definitions[bits] + bits = self.bit_definitions[cast(str, bits)] except (TypeError, KeyError): pass temp = original_value mask = 0 - for bit in bits: + lbits = cast(List[int], bits) + for bit in lbits: mask |= 1 << bit temp &= ~mask - temp |= bit_value << min(bits) + temp |= bit_value << min(lbits) return temp diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index 0ba65199..e99f3c90 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -340,7 +340,7 @@ def read(self, from_od=False) -> None: DCF value will be used, otherwise the EDS default will be used instead. """ - def _raw_from(param): + def _raw_from(param) -> int: if from_od: if param.od.value is not None: return param.od.value @@ -466,6 +466,8 @@ def subscribe(self) -> None: known to match what's stored on the node. """ if self.enabled: + if self.cob_id is None: + raise RuntimeError("A valid COB-ID is required") logger.info("Subscribing to enabled PDO 0x%X on the network", self.cob_id) self.pdo_node.network.subscribe(self.cob_id, self.on_message) @@ -513,6 +515,8 @@ def add_variable( def transmit(self) -> None: """Transmit the message once.""" + if self.cob_id is None: + raise RuntimeError("A valid COB-ID is required") self.pdo_node.network.send_message(self.cob_id, self.data) def start(self, period: Optional[float] = None) -> None: @@ -523,6 +527,9 @@ def start(self, period: Optional[float] = None) -> None: on the object before. :raises ValueError: When neither the argument nor the :attr:`period` is given. """ + if self.cob_id is None: + raise RuntimeError("A valid COB-ID is required") + # Stop an already running transmission if we have one, otherwise we # overwrite the reference and can lose our handle to shut it down self.stop() @@ -553,9 +560,11 @@ def remote_request(self) -> None: Silently ignore if not allowed. """ if self.enabled and self.rtr_allowed: + if self.cob_id is None: + raise RuntimeError("A valid COB-ID is required") self.pdo_node.network.send_message(self.cob_id, bytes(), remote=True) - def wait_for_reception(self, timeout: float = 10) -> float: + def wait_for_reception(self, timeout: float = 10) -> Optional[float]: """Wait for the next transmit PDO. :param float timeout: Max time to wait in seconds. @@ -583,6 +592,12 @@ def get_data(self) -> bytes: :return: PdoVariable value as :class:`bytes`. """ + # FIXME TYPING: These asserts are for type checking. More robust errors + # should be raised if these are not set. + assert self.offset is not None + assert self.pdo_parent is not None + assert self.od.data_type is not None + byte_offset, bit_offset = divmod(self.offset, 8) if bit_offset or self.length % 8: @@ -610,6 +625,12 @@ def set_data(self, data: bytes): :param data: Value for the PDO variable in the PDO message. """ + # FIXME TYPING: These asserts are for type checking. More robust errors + # should be raised if these are not set. + assert self.offset is not None + assert self.pdo_parent is not None + assert self.od.data_type is not None + byte_offset, bit_offset = divmod(self.offset, 8) logger.debug("Updating %s to %s in %s", self.name, binascii.hexlify(data), self.pdo_parent.name) diff --git a/canopen/sdo/base.py b/canopen/sdo/base.py index ddc75ed9..75bec8c3 100644 --- a/canopen/sdo/base.py +++ b/canopen/sdo/base.py @@ -2,7 +2,7 @@ import binascii from collections.abc import Mapping -from typing import Iterator, Optional, Union +from typing import Iterator, Optional, Union, cast import canopen.network from canopen import objectdictionary @@ -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( @@ -79,6 +79,7 @@ def get_variable( return obj elif isinstance(obj, (SdoRecord, SdoArray)): return obj.get(subindex) + return None def upload(self, index: int, subindex: int) -> bytes: raise NotImplementedError() @@ -113,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 @@ -134,10 +135,10 @@ def __iter__(self) -> Iterator[int]: return iter(range(1, len(self) + 1)) def __len__(self) -> int: - return self[0].raw + return cast(int, self[0].raw) - def __contains__(self, subindex: int) -> bool: - return 0 <= subindex <= len(self) + def __contains__(self, subindex: object) -> bool: + return isinstance(subindex, int) and 0 <= subindex <= len(self) class SdoVariable(variable.Variable): diff --git a/canopen/variable.py b/canopen/variable.py index d2538c3f..0360ef23 100644 --- a/canopen/variable.py +++ b/canopen/variable.py @@ -1,5 +1,6 @@ import logging from collections.abc import Mapping +from typing import List, Union, cast from typing import Union from canopen import objectdictionary @@ -78,7 +79,7 @@ def raw(self) -> Union[int, bool, float, str, bytes]: value = self.od.decode_raw(self.data) text = f"Value of {self.name!r} ({pretty_index(self.index, self.subindex)}) is {value!r}" if value in self.od.value_descriptions: - text += f" ({self.od.value_descriptions[value]})" + text += f" ({self.od.value_descriptions[cast(int, value)]})" logger.debug(text) return value @@ -109,7 +110,10 @@ def phys(self, value: Union[int, bool, float, str, bytes]): @property def desc(self) -> str: """Converts to and from a description of the value as a string.""" - value = self.od.decode_desc(self.raw) + raw = self.raw + if not isinstance(raw, int): + raise TypeError("Description can only be used with integer values") + value = self.od.decode_desc(raw) logger.debug("Description is '%s'", value) return value @@ -142,6 +146,7 @@ def read(self, fmt: str = "raw") -> Union[int, bool, float, str, bytes]: return self.phys elif fmt == "desc": return self.desc + raise ValueError(f"Uknown format {fmt!r}") def write( self, value: Union[int, bool, float, str, bytes], fmt: str = "raw" @@ -161,17 +166,23 @@ def write( elif fmt == "phys": self.phys = value elif fmt == "desc": + if not isinstance(value, str): + raise TypeError("Description must be a string") self.desc = value class Bits(Mapping): + # Attribute type (since not defined in __init__) + raw: int + def __init__(self, variable: Variable): self.variable = variable self.read() @staticmethod - def _get_bits(key): + def _get_bits(key: Union[int, str, slice]) -> Union[range, List[int], str]: + bits: Union[range, List[int], str] if isinstance(key, slice): bits = range(key.start, key.stop, key.step) elif isinstance(key, int): diff --git a/pyproject.toml b/pyproject.toml index a494188e..5b065e65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,3 +48,11 @@ testpaths = [ filterwarnings = [ "ignore::DeprecationWarning", ] + +[tool.mypy] +files = "canopen" +strict = "False" +ignore_missing_imports = "True" +disable_error_code = [ + "annotation-unchecked", +]