can.io: Distinguish Text/Binary-IO for Reader/Writer classes.

Author: ro-i

can/io/asc.py

@@ -14,7 +14,7 @@
 from ..message import Message
 from ..typechecking import StringPathLike
 from ..util import channel2int, dlc2len, len2dlc
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import TextIOMessageReader, TextIOMessageWriter
 
 CAN_MSG_EXT = 0x80000000
 CAN_ID_MASK = 0x1FFFFFFF
@@ -24,7 +24,7 @@
 logger = logging.getLogger("can.io.asc")
 
 
-class ASCReader(MessageReader):
+class ASCReader(TextIOMessageReader):
     """
     Iterator of CAN messages from a ASC logging file. Meta data (comments,
     bus statistics, J1939 Transport Protocol messages) is ignored.
@@ -308,7 +308,7 @@ def __iter__(self) -> Generator[Message, None, None]:
         self.stop()
 
 
-class ASCWriter(FileIOMessageWriter):
+class ASCWriter(TextIOMessageWriter):
     """Logs CAN data to an ASCII log file (.asc).
 
     The measurement starts with the timestamp of the first registered message.

can/io/blf.py

@@ -22,7 +22,7 @@
 from ..message import Message
 from ..typechecking import StringPathLike
 from ..util import channel2int, dlc2len, len2dlc
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import BinaryIOMessageReader, FileIOMessageWriter
 
 TSystemTime = Tuple[int, int, int, int, int, int, int, int]
 
@@ -132,7 +132,7 @@ def systemtime_to_timestamp(systemtime: TSystemTime) -> float:
         return 0
 
 
-class BLFReader(MessageReader):
+class BLFReader(BinaryIOMessageReader):
     """
     Iterator of CAN messages from a Binary Logging File.
 

can/io/canutils.py

@@ -10,7 +10,7 @@
 from can.message import Message
 
 from ..typechecking import StringPathLike
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import TextIOMessageReader, TextIOMessageWriter
 
 log = logging.getLogger("can.io.canutils")
 
@@ -23,7 +23,7 @@
 CANFD_ESI = 0x02
 
 
-class CanutilsLogReader(MessageReader):
+class CanutilsLogReader(TextIOMessageReader):
     """
     Iterator over CAN messages from a .log Logging File (candump -L).
 
@@ -122,7 +122,7 @@ def __iter__(self) -> Generator[Message, None, None]:
         self.stop()
 
 
-class CanutilsLogWriter(FileIOMessageWriter):
+class CanutilsLogWriter(TextIOMessageWriter):
     """Logs CAN data to an ASCII log file (.log).
     This class is is compatible with "candump -L".
 

can/io/csv.py

@@ -15,10 +15,10 @@
 from can.message import Message
 
 from ..typechecking import StringPathLike
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import TextIOMessageReader, TextIOMessageWriter
 
 
-class CSVReader(MessageReader):
+class CSVReader(TextIOMessageReader):
     """Iterator over CAN messages from a .csv file that was
     generated by :class:`~can.CSVWriter` or that uses the same
     format as described there. Assumes that there is a header
@@ -67,7 +67,7 @@ def __iter__(self) -> Generator[Message, None, None]:
         self.stop()
 
 
-class CSVWriter(FileIOMessageWriter):
+class CSVWriter(TextIOMessageWriter):
     """Writes a comma separated text file with a line for
     each message. Includes a header line.
 

can/io/generic.py

@@ -4,13 +4,16 @@
 from types import TracebackType
 from typing import (
     Any,
+    BinaryIO,
     ContextManager,
     Iterable,
     Optional,
+    TextIO,
     Type,
     cast,
 )
 
+import gzip
 from typing_extensions import Literal
 
 from .. import typechecking
@@ -105,5 +108,21 @@ def file_size(self) -> int:
         return self.file.tell()
 
 
+class TextIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):
+    file: TextIO | gzip.GzipFile
+
+
+class BinaryIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):
+    file: BinaryIO | gzip.GzipFile
+
+
 class MessageReader(BaseIOHandler, Iterable[Message], metaclass=ABCMeta):
     """The base class for all readers."""
+
+
+class TextIOMessageReader(MessageReader, metaclass=ABCMeta):
+    file: TextIO | gzip.GzipFile
+
+
+class BinaryIOMessageReader(MessageReader, metaclass=ABCMeta):
+    file: BinaryIO | gzip.GzipFile

can/io/logger.py

@@ -20,7 +20,12 @@
 from .blf import BLFWriter
 from .canutils import CanutilsLogWriter
 from .csv import CSVWriter
-from .generic import BaseIOHandler, FileIOMessageWriter, MessageWriter
+from .generic import (
+    BaseIOHandler,
+    BinaryIOMessageWriter,
+    FileIOMessageWriter,
+    MessageWriter,
+)
 from .mf4 import MF4Writer
 from .printer import Printer
 from .sqlite import SqliteWriter
@@ -94,20 +99,28 @@ def __new__(  # type: ignore
 
         file_or_filename: AcceptedIOType = filename
         if suffix == ".gz":
-            suffix, file_or_filename = Logger.compress(filename, **kwargs)
+            LoggerType, file_or_filename = Logger.compress(filename, **kwargs)
+        else:
+            LoggerType = cls._get_logger_for_suffix(suffix)
+
+        return LoggerType(file=file_or_filename, **kwargs)
 
+    @classmethod
+    def _get_logger_for_suffix(cls, suffix: str) -> Type[MessageWriter]:
         try:
             LoggerType = Logger.message_writers[suffix]
             if LoggerType is None:
                 raise ValueError(f'failed to import logger for extension "{suffix}"')
-            return LoggerType(file=file_or_filename, **kwargs)
+            return LoggerType
         except KeyError:
             raise ValueError(
                 f'No write support for this unknown log format "{suffix}"'
             ) from None
 
-    @staticmethod
-    def compress(filename: StringPathLike, **kwargs: Any) -> Tuple[str, FileLike]:
+    @classmethod
+    def compress(
+        cls, filename: StringPathLike, **kwargs: Any
+    ) -> Tuple[Type[MessageWriter], FileLike]:
         """
         Return the suffix and io object of the decompressed file.
         File will automatically recompress upon close.
@@ -117,12 +130,15 @@ def compress(filename: StringPathLike, **kwargs: Any) -> Tuple[str, FileLike]:
             raise ValueError(
                 f"The file type {real_suffix} is currently incompatible with gzip."
             )
-        if kwargs.get("append", False):
-            mode = "ab" if real_suffix == ".blf" else "at"
+        LoggerType = cls._get_logger_for_suffix(real_suffix)
+        append = kwargs.get("append", False)
+
+        if issubclass(LoggerType, BinaryIOMessageWriter):
+            mode = "ab" if append else "wb"
         else:
-            mode = "wb" if real_suffix == ".blf" else "wt"
+            mode = "at" if append else "wt"
 
-        return real_suffix, gzip.open(filename, mode)
+        return LoggerType, gzip.open(filename, mode)
 
     def on_message_received(self, msg: Message) -> None:
         pass

can/io/mf4.py

@@ -14,7 +14,7 @@
 from ..message import Message
 from ..typechecking import StringPathLike
 from ..util import channel2int, dlc2len, len2dlc
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import BinaryIOMessageReader, BinaryIOMessageWriter
 
 logger = logging.getLogger("can.io.mf4")
 
@@ -75,7 +75,7 @@
 CAN_ID_MASK = 0x1FFFFFFF
 
 
-class MF4Writer(FileIOMessageWriter):
+class MF4Writer(BinaryIOMessageWriter):
     """Logs CAN data to an ASAM Measurement Data File v4 (.mf4).
 
     MF4Writer does not support append mode.
@@ -265,7 +265,7 @@ def on_message_received(self, msg: Message) -> None:
         self._rtr_buffer = np.zeros(1, dtype=RTR_DTYPE)
 
 
-class MF4Reader(MessageReader):
+class MF4Reader(BinaryIOMessageReader):
     """
     Iterator of CAN messages from a MF4 logging file.
 

can/io/player.py

@@ -16,7 +16,7 @@
 from .blf import BLFReader
 from .canutils import CanutilsLogReader
 from .csv import CSVReader
-from .generic import MessageReader
+from .generic import BinaryIOMessageReader, MessageReader
 from .mf4 import MF4Reader
 from .sqlite import SqliteReader
 from .trc import TRCReader
@@ -87,7 +87,13 @@ def __new__(  # type: ignore
 
         file_or_filename: AcceptedIOType = filename
         if suffix == ".gz":
-            suffix, file_or_filename = LogReader.decompress(filename)
+            ReaderType, file_or_filename = LogReader.decompress(filename)
+        else:
+            ReaderType = cls._get_logger_for_suffix(suffix)
+        return ReaderType(file=file_or_filename, **kwargs)
+
+    @classmethod
+    def _get_logger_for_suffix(cls, suffix: str) -> typing.Type[MessageReader]:
         try:
             ReaderType = LogReader.message_readers[suffix]
         except KeyError:
@@ -96,19 +102,22 @@ def __new__(  # type: ignore
             ) from None
         if ReaderType is None:
             raise ImportError(f"failed to import reader for extension {suffix}")
-        return ReaderType(file=file_or_filename, **kwargs)
+        return ReaderType
 
-    @staticmethod
+    @classmethod
     def decompress(
+        cls,
         filename: StringPathLike,
-    ) -> typing.Tuple[str, typing.Union[str, FileLike]]:
+    ) -> typing.Tuple[typing.Type[MessageReader], typing.Union[str, FileLike]]:
         """
         Return the suffix and io object of the decompressed file.
         """
         real_suffix = pathlib.Path(filename).suffixes[-2].lower()
-        mode = "rb" if real_suffix == ".blf" else "rt"
+        ReaderType = cls._get_logger_for_suffix(real_suffix)
+
+        mode = "rb" if issubclass(ReaderType, BinaryIOMessageReader) else "rt"
 
-        return real_suffix, gzip.open(filename, mode)
+        return ReaderType, gzip.open(filename, mode)
 
     def __iter__(self) -> typing.Generator[Message, None, None]:
         raise NotImplementedError()

can/io/trc.py

@@ -16,7 +16,12 @@
 from ..message import Message
 from ..typechecking import StringPathLike
 from ..util import channel2int, dlc2len, len2dlc
-from .generic import FileIOMessageWriter, MessageReader
+from .generic import (
+    FileIOMessageWriter,
+    MessageReader,
+    TextIOMessageReader,
+    TextIOMessageWriter,
+)
 
 logger = logging.getLogger("can.io.trc")
 
@@ -36,7 +41,7 @@ def __ge__(self, other):
         return NotImplemented
 
 
-class TRCReader(MessageReader):
+class TRCReader(TextIOMessageReader):
     """
     Iterator of CAN messages from a TRC logging file.
     """
@@ -241,7 +246,7 @@ def __iter__(self) -> Generator[Message, None, None]:
         self.stop()
 
 
-class TRCWriter(FileIOMessageWriter):
+class TRCWriter(TextIOMessageWriter):
     """Logs CAN data to text file (.trc).
 
     The measurement starts with the timestamp of the first registered message.