satiate mypy

Author: driftregion

can/io/logger.py

@@ -7,7 +7,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime
 import gzip
-from typing import Any, Optional, Callable, Type, Tuple, cast, Dict
+from typing import Any, Optional, Callable, Type, Tuple, cast, Dict, Optional
 
 from types import TracebackType
 
@@ -25,10 +25,13 @@
 from .printer import Printer
 from ..typechecking import StringPathLike, FileLike, AcceptedIOType
 
+MF4Writer: Optional[Type[MessageWriter]] = None
 try:
-    from .mf4 import MF4Writer
+    from .mf4 import MF4Writer as _MF4Writer
+
+    MF4Writer = _MF4Writer
 except ImportError:
-    MF4Writer = None
+    pass
 
 
 class Logger(MessageWriter):  # pylint: disable=abstract-method
@@ -58,7 +61,7 @@ class Logger(MessageWriter):  # pylint: disable=abstract-method
     """
 
     fetched_plugins = False
-    message_writers: Dict[str, Type[MessageWriter]] = {
+    message_writers: Dict[str, Optional[Type[MessageWriter]]] = {
         ".asc": ASCWriter,
         ".blf": BLFWriter,
         ".csv": CSVWriter,

can/io/mf4.py

@@ -13,7 +13,7 @@
 from ..message import Message
 from ..listener import Listener
 from ..util import channel2int
-from .generic import BaseIOHandler
+from .generic import MessageReader, MessageWriter
 
 try:
     from asammdf import Signal
@@ -76,7 +76,7 @@
 logger = logging.getLogger("can.io.mf4")
 
 
-class MF4Writer(BaseIOHandler, Listener):
+class MF4Writer(MessageWriter, Listener):
     """Logs CAN data to an ASAM Measurement Data File v4 (.mf4).
 
     MF4Writer does not support append mode.
@@ -220,7 +220,7 @@ def on_message_received(self, msg):
             self._mdf.extend(0, sigs)
 
 
-class MF4Reader(BaseIOHandler):
+class MF4Reader(MessageReader):
     """
     Iterator of CAN messages from a MF4 logging file.
 

can/io/player.py

@@ -19,10 +19,14 @@
 from ..typechecking import StringPathLike, FileLike, AcceptedIOType
 from ..message import Message
 
+
+MF4Reader: typing.Optional[typing.Type[MessageReader]] = None
 try:
-    from .mf4 import MF4Reader
+    from .mf4 import MF4Reader as _MF4Reader
+
+    MF4Reader = _MF4Reader
 except ImportError:
-    MF4Reader = None
+    pass
 
 
 class LogReader(MessageReader):
@@ -56,7 +60,7 @@ class LogReader(MessageReader):
     """
 
     fetched_plugins = False
-    message_readers: typing.Dict[str, typing.Type[MessageReader]] = {
+    message_readers: typing.Dict[str, typing.Optional[typing.Type[MessageReader]]] = {
         ".asc": ASCReader,
         ".blf": BLFReader,
         ".csv": CSVReader,
@@ -91,11 +95,14 @@ def __new__(  # type: ignore
         if suffix == ".gz":
             suffix, file_or_filename = LogReader.decompress(filename)
         try:
-            return LogReader.message_readers[suffix](file_or_filename, *args, **kwargs)
+            ReaderType = LogReader.message_readers[suffix]
         except KeyError:
             raise ValueError(
                 f'No read support for this unknown log format "{suffix}"'
             ) from None
+        if ReaderType is None:
+            raise ImportError(f"failed to import reader for extension {suffix}")
+        return ReaderType(file_or_filename, *args, **kwargs)
 
     @staticmethod
     def decompress(