Merge pull request #271 from andlaus/strict_mypy

Enable `mypy --strict`

examples/isotp_send.py

@@ -37,7 +37,7 @@ def isotp_error_handler(error: str) -> None:
 # bs: maximum block size. (Must be smaller than 4096?)
 isotp_socket.set_fc_opts(stmin=5, bs=100)
 
-can_bus = can.Bus(channel=channel, interface="socketcan")  # type: ignore[abstract]
+can_bus = can.Bus(channel=channel, interface="socketcan")
 
 # note that we specify the CAN IDs from the ECU's point of view, i.e.,
 # from the tester's (our) perspective, they are reversed.

examples/somersaultecu.py

@@ -81,7 +81,7 @@ class SomersaultSID(IntEnum):
 
 # extend the Somersault SIDs by the UDS ones
 tmp = [(i.name, i.value) for i in chain(uds.SID, SomersaultSID)]
-SID: Any = IntEnum("SID", tmp)  # type: ignore
+SID: Any = IntEnum("SID", tmp)  # type: ignore[misc]
 
 dlc_short_name = "somersault"
 
@@ -625,7 +625,7 @@ class SomersaultSID(IntEnum):
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
                     coded_value=uds.positive_response_id(
-                        SID.DiagnosticSessionControl.value),  # type: ignore
+                        SID.DiagnosticSessionControl.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -661,7 +661,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint16"],
                     byte_position=0,
-                    coded_value=uds.positive_response_id(SID.TesterPresent.value),  # type: ignore
+                    coded_value=uds.positive_response_id(
+                        SID.TesterPresent.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -745,7 +746,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=uds.positive_response_id(SID.ForwardFlip.value),  # type: ignore
+                    coded_value=uds.positive_response_id(
+                        SID.ForwardFlip.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -793,7 +795,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=uds.positive_response_id(SID.BackwardFlip.value),  # type: ignore
+                    coded_value=uds.positive_response_id(
+                        SID.BackwardFlip.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -843,7 +846,7 @@ class SomersaultSID(IntEnum):
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
                     coded_value=uds.positive_response_id(
-                        SID.ReadDataByIdentifier.value),  # type: ignore
+                        SID.ReadDataByIdentifier.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -919,7 +922,7 @@ class SomersaultSID(IntEnum):
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
                     coded_value=uds.positive_response_id(
-                        SID.ForwardFlipCondition.value),  # type: ignore
+                        SID.ForwardFlipCondition.value),  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1008,7 +1011,7 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=1,
-                    coded_value=uds.SID.TesterPresent.value,  # type: ignore
+                    coded_value=uds.SID.TesterPresent.value,  # type: ignore[attr-defined]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1345,7 +1348,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.DiagnosticSessionControl.value,  # type: ignore
+                    coded_value=SID.DiagnosticSessionControl  # type: ignore[attr-defined,arg-type]
+                    .value,
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1379,7 +1383,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.DiagnosticSessionControl.value,  # type: ignore
+                    coded_value=SID.DiagnosticSessionControl  # type: ignore[attr-defined,arg-type]
+                    .value,
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1413,7 +1418,7 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.TesterPresent.value,  # type: ignore
+                    coded_value=SID.TesterPresent.value,  # type: ignore[attr-defined,arg-type]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1448,7 +1453,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.ForwardFlipCondition.value,  # type: ignore
+                    coded_value=SID.ForwardFlipCondition  # type: ignore[attr-defined,arg-type]
+                    .value,
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1483,7 +1489,7 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.ForwardFlip.value,  # type: ignore
+                    coded_value=SID.ForwardFlip.value,  # type: ignore[attr-defined,arg-type]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1531,7 +1537,7 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.BackwardFlip.value,  # type: ignore
+                    coded_value=SID.BackwardFlip.value,  # type: ignore[attr-defined,arg-type]
                     bit_position=None,
                     sdgs=[],
                 ),
@@ -1579,7 +1585,8 @@ class SomersaultSID(IntEnum):
                     description=None,
                     diag_coded_type=somersault_diagcodedtypes["uint8"],
                     byte_position=0,
-                    coded_value=SID.ReadDataByIdentifier.value,  # type: ignore
+                    coded_value=SID.ReadDataByIdentifier  # type: ignore[attr-defined,arg-type]
+                    .value,
                     bit_position=None,
                     sdgs=[],
                 ),

examples/somersaultlazy.py

@@ -14,15 +14,19 @@
 
 import odxtools
 import odxtools.uds as uds
+from odxtools.exceptions import odxrequire
 from odxtools.message import Message
 from odxtools.odxtypes import ParameterValueDict
+from odxtools.response import Response, ResponseType
+
+from .somersaultecu import database as somersaultdatabase
 
 tester_logger = logging.getLogger("somersault_lazy_tester")
 ecu_logger = logging.getLogger("somersault_lazy_ecu")
 
 is_sterile = False
 can_channel: Optional[str] = None
-somersault_lazy_diag_layer = somersaultecu.database.ecus.somersault_lazy  # type:ignore
+somersault_lazy_diag_layer = somersaultdatabase.ecus.somersault_lazy
 
 # the raw payload data of the telegrams received by the ECU and by the
 # tester when in sterile mode (unittest without a CAN channel)
@@ -162,8 +166,8 @@ def __init__(self) -> None:
 
         self.isotp_socket = create_isotp_socket(
             can_channel,
-            rxid=somersault_lazy_diag_layer.get_can_receive_id(),
-            txid=somersault_lazy_diag_layer.get_can_send_id(),
+            rxid=odxrequire(somersault_lazy_diag_layer.get_can_receive_id()),
+            txid=odxrequire(somersault_lazy_diag_layer.get_can_send_id()),
         )
 
         ##############
@@ -172,13 +176,11 @@ def __init__(self) -> None:
         ##############
 
         # determine the timeout on inactivity [s]
-        self._idle_timeout = somersault_lazy_diag_layer.get_tester_present_time()
-        if self._idle_timeout is None:
-            self._idle_timeout = 3.0  # default specified by the standard
+        self._idle_timeout = somersault_lazy_diag_layer.get_tester_present_time() or 3.0
 
         # we send a response to tester present messages. make sure
         # that this is specified
-        cp = somersault_lazy_diag_layer.get_communication_parameter("CP_TesterPresentReqRsp")
+        cp = odxrequire(somersault_lazy_diag_layer.get_comparam("CP_TesterPresentReqRsp"))
         assert cp.value == "Response expected" or cp.value == "1"
 
     async def _handle_requests_task(self) -> None:
@@ -367,17 +369,18 @@ async def tester_await_response(isotp_socket: Optional[isotp.tpsock.socket],
     try:
         replies = somersault_lazy_diag_layer.decode_response(raw_response, raw_message)
         assert len(replies) == 1  # replies must always be uniquely decodable
+        assert isinstance(replies[0].coding_object, Response)
 
-        if replies[0].coding_object.response_type == "POS-RESPONSE":
+        if replies[0].coding_object.response_type == ResponseType.POSITIVE:
             rtype = "positive"
-        elif replies[0].coding_object.response_type == "NEG-RESPONSE":
+        elif replies[0].coding_object.response_type == ResponseType.NEGATIVE:
             rtype = "negative"
         else:
             rtype = "unknown"
 
         tester_logger.debug(f"received {rtype} response")
 
-        return replies[0]
+        return replies[0].param_dict
 
     except odxtools.exceptions.DecodeError as e:
         if len(raw_response) >= 3:
@@ -408,8 +411,8 @@ async def tester_main() -> None:
     # reversed.
     isotp_socket = create_isotp_socket(
         can_channel,
-        txid=somersault_lazy_diag_layer.get_can_receive_id(),
-        rxid=somersault_lazy_diag_layer.get_can_send_id(),
+        txid=odxrequire(somersault_lazy_diag_layer.get_can_receive_id()),
+        rxid=odxrequire(somersault_lazy_diag_layer.get_can_send_id()),
     )
 
     # try to to do a single forward flip without having an active session (ought to fail)

odxtools/__init__.py

@@ -62,11 +62,11 @@
 - _`[ISO22901]` The ISO 22901 Standard: https://www.iso.org/standard/41207.html
 
 """
-from .load_file import load_file
-from .load_odx_d_file import load_odx_d_file
-from .load_pdx_file import load_pdx_file
-from .version import __version__
-from .write_pdx_file import write_pdx_file
+from .load_file import load_file as load_file
+from .load_odx_d_file import load_odx_d_file as load_odx_d_file
+from .load_pdx_file import load_pdx_file as load_pdx_file
+from .version import __version__ as __version__
+from .write_pdx_file import write_pdx_file as write_pdx_file
 
 __author__ = "Katrin Bauer"
 

odxtools/basicstructure.py

@@ -254,9 +254,7 @@ def encode(self, coded_request: Optional[bytes] = None, **params: ParameterValue
             Parameters of the RPC as mapping from SHORT-NAME of the parameter to the value
         """
         return self.convert_physical_to_internal(
-            params,  # type: ignore[arg-type]
-            triggering_coded_request=coded_request,
-            is_end_of_pdu=True)
+            params, triggering_coded_request=coded_request, is_end_of_pdu=True)
 
     def decode(self, message: bytes) -> ParameterValueDict:
         decode_state = DecodeState(coded_message=message)

odxtools/cli/_parser_utils.py

@@ -2,11 +2,29 @@
 #
 # SPDX-License-Identifier: MIT
 import argparse
+from typing import Any, Protocol, TypeVar, runtime_checkable
 
 from ..database import Database
 from ..load_file import load_file as _load_file
 
 
+@runtime_checkable
+class SubparsersList(Protocol):
+    """This protocol reproduces the parts of `argparse._SubparsersAction` which are needed by odxtools
+
+    Unfortunately this is necessary because
+    `argparse._SubparsersAction` is a private class of the `argparse`
+    module that is used by some of its public API and thus cannot be
+    used outside of the `argparse` module.
+    """
+
+    def add_parser(self, name: str, **kwargs: Any) -> "argparse.ArgumentParser":
+        ...
+
+
+TSubparsersAction = TypeVar("TSubparsersAction", bound=SubparsersList)
+
+
 def add_pdx_argument(parser: argparse.ArgumentParser, is_optional: bool = False) -> None:
     parser.add_argument(
         "pdx_file",

odxtools/cli/_print_utils.py

@@ -36,7 +36,7 @@ def print_diagnostic_service(service: DiagService,
                              print_state_transitions: bool = False,
                              print_audiences: bool = False,
                              allow_unknown_bit_lengths: bool = False,
-                             print_fn: Callable = print) -> None:
+                             print_fn: Callable[..., Any] = print) -> None:
 
     print_fn(f" Service '{service.short_name}':")
 
@@ -70,7 +70,7 @@ def print_diagnostic_service(service: DiagService,
 
 def print_service_parameters(service: DiagService,
                              allow_unknown_bit_lengths: bool = False,
-                             print_fn: Callable = print) -> None:
+                             print_fn: Callable[..., Any] = print) -> None:
     # prints parameter details of request, positive response and negative response of diagnostic service
 
     # Request
@@ -227,7 +227,7 @@ def extract_parameter_tabulation_data(parameters: List[Parameter]) -> Dict[str,
     }
 
 
-def print_dl_metrics(variants: List[DiagLayer], print_fn: Callable = print) -> None:
+def print_dl_metrics(variants: List[DiagLayer], print_fn: Callable[..., Any] = print) -> None:
 
     name = []
     type = []

odxtools/cli/browse.py

@@ -4,7 +4,7 @@
 import sys
 from typing import List, Optional, Union, cast
 
-from InquirerPy import prompt as PI_prompt
+import InquirerPy.prompt as PI_prompt
 from tabulate import tabulate  # TODO: switch to rich tables
 
 from ..database import Database
@@ -20,6 +20,7 @@
 from ..request import Request
 from ..response import Response
 from . import _parser_utils
+from ._parser_utils import TSubparsersAction
 from ._print_utils import extract_parameter_tabulation_data
 
 # name of the tool
@@ -250,7 +251,8 @@ def encode_message_from_string_values(
                 parameter_values[parameter_sn] = _convert_string_to_odx_type(
                     parameter_value, DataType.A_BYTEFIELD)
 
-    payload = sub_service.encode(**parameter_values)  # type: ignore
+    payload = sub_service.encode(coded_request=b'\xff' * 100, **parameter_values)
+
     print(f"Message payload: 0x{bytes(payload).hex()}")
 
 
@@ -353,7 +355,7 @@ def browse(odxdb: Database) -> None:
                 encode_message_interactively(codec, ask_user_confirmation=True)
 
 
-def add_subparser(subparsers: "argparse._SubParsersAction") -> None:
+def add_subparser(subparsers: TSubparsersAction) -> None:
     # Browse interactively to avoid spamming the console.
     parser = subparsers.add_parser(
         "browse",