can.io: Add preferred mode for opening files to Reader/Writer classes.

Author: ro-i

can/io/asc.py

@@ -32,6 +32,8 @@ class ASCReader(MessageReader):
 
     file: TextIO
 
+    PREFERRED_MODE: str = "rt"
+
     def __init__(
         self,
         file: Union[StringPathLike, TextIO],
@@ -50,7 +52,7 @@ def __init__(
                      `relative` (starting at 0.0) or `absolute` (starting at
                      the system time). Default `True = relative`.
         """
-        super().__init__(file, mode="r")
+        super().__init__(file)
 
         if not self.file:
             raise ValueError("The given file cannot be None")
@@ -319,6 +321,8 @@ class ASCWriter(FileIOMessageWriter):
 
     file: TextIO
 
+    PREFERRED_MODE: str = "wt"
+
     FORMAT_MESSAGE = "{channel}  {id:<15} {dir:<4} {dtype} {data}"
     FORMAT_MESSAGE_FD = " ".join(
         [
@@ -363,7 +367,7 @@ def __init__(
                 f"{self.__class__.__name__} is currently not equipped to "
                 f"append messages to an existing file."
             )
-        super().__init__(file, mode="w")
+        super().__init__(file)
 
         self.channel = channel
 

can/io/blf.py

@@ -142,6 +142,8 @@ class BLFReader(MessageReader):
 
     file: BinaryIO
 
+    PREFERRED_MODE: str = "rb"
+
     def __init__(
         self,
         file: Union[StringPathLike, BinaryIO],
@@ -152,7 +154,7 @@ def __init__(
                      If this is a file-like object, is has to opened in binary
                      read mode, not text read mode.
         """
-        super().__init__(file, mode="rb")
+        super().__init__(file)
         data = self.file.read(FILE_HEADER_STRUCT.size)
         header = FILE_HEADER_STRUCT.unpack(data)
         if header[0] != b"LOGG":
@@ -361,6 +363,9 @@ class BLFWriter(FileIOMessageWriter):
 
     file: BinaryIO
 
+    PREFERRED_MODE: str = "wb"
+    PREFERRED_MODE_APPEND: str = "rb+"
+
     #: Max log container size of uncompressed data
     max_container_size = 128 * 1024
 
@@ -391,13 +396,13 @@ def __init__(
             Z_DEFAULT_COMPRESSION represents a default compromise between
             speed and compression (currently equivalent to level 6).
         """
-        mode = "rb+" if append else "wb"
+        mode = self.PREFERRED_MODE_APPEND if append else self.PREFERRED_MODE
         try:
             super().__init__(file, mode=mode)
         except FileNotFoundError:
             # Trying to append to a non-existing file, create a new one
             append = False
-            mode = "wb"
+            mode = self.PREFERRED_MODE
             super().__init__(file, mode=mode)
         assert self.file is not None
         self.channel = channel

can/io/canutils.py

@@ -35,6 +35,8 @@ class CanutilsLogReader(MessageReader):
 
     file: TextIO
 
+    PREFERRED_MODE: str = "rt"
+
     def __init__(
         self,
         file: Union[StringPathLike, TextIO],
@@ -45,7 +47,7 @@ def __init__(
                      If this is a file-like object, is has to opened in text
                      read mode, not binary read mode.
         """
-        super().__init__(file, mode="r")
+        super().__init__(file)
 
     def __iter__(self) -> Generator[Message, None, None]:
         for line in self.file:
@@ -131,6 +133,9 @@ class CanutilsLogWriter(FileIOMessageWriter):
     It the first message does not have a timestamp, it is set to zero.
     """
 
+    PREFERRED_MODE: str = "wt"
+    PREFERRED_MODE_APPEND: str = "at"
+
     def __init__(
         self,
         file: Union[StringPathLike, TextIO],
@@ -147,7 +152,7 @@ def __init__(
         :param bool append: if set to `True` messages are appended to
                             the file, else the file is truncated
         """
-        mode = "a" if append else "w"
+        mode = self.PREFERRED_MODE_APPEND if append else self.PREFERRED_MODE
         super().__init__(file, mode=mode)
 
         self.channel = channel

can/io/csv.py

@@ -29,6 +29,8 @@ class CSVReader(MessageReader):
 
     file: TextIO
 
+    PREFERRED_MODE: str = "rt"
+
     def __init__(
         self,
         file: Union[StringPathLike, TextIO],
@@ -39,7 +41,7 @@ def __init__(
                      If this is a file-like object, is has to opened in text
                      read mode, not binary read mode.
         """
-        super().__init__(file, mode="r")
+        super().__init__(file)
 
     def __iter__(self) -> Generator[Message, None, None]:
         # skip the header line
@@ -90,6 +92,9 @@ class CSVWriter(FileIOMessageWriter):
 
     file: TextIO
 
+    PREFERRED_MODE: str = "wt"
+    PREFERRED_MODE_APPEND: str = "at"
+
     def __init__(
         self,
         file: Union[StringPathLike, TextIO],
@@ -105,7 +110,7 @@ def __init__(
                             the file is truncated and starts with a newly
                             written header line
         """
-        mode = "a" if append else "w"
+        mode = self.PREFERRED_MODE_APPEND if append else self.PREFERRED_MODE
         super().__init__(file, mode=mode)
 
         # Write a header row

can/io/generic.py

@@ -5,6 +5,7 @@
 from typing import (
     Any,
     ContextManager,
+    Final,
     Iterable,
     Optional,
     Type,
@@ -23,17 +24,28 @@ class BaseIOHandler(ContextManager, metaclass=ABCMeta):
 
     Can be used as a context manager.
 
+    :const PREFERRED_MODE:
+        the preferred mode to be used to open a file. Is used by default
+        when no mode is specified for `__init__`.
+    :const PREFERRED_MODE_APPEND:
+        the preferred mode to be used to open a file in append mode. Is
+        used by default when no mode is specified for `__init__` and if
+        `__init__` takes an boolean `append` argument which has been
+        specified as `True`.
     :attr file:
         the file-like object that is kept internally, or `None` if none
         was opened
     """
 
     file: Optional[typechecking.FileLike]
 
+    PREFERRED_MODE_APPEND: str = "rt+"
+    PREFERRED_MODE: str = "rt"
+
     def __init__(
         self,
         file: Optional[typechecking.AcceptedIOType],
-        mode: str = "rt",
+        mode: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
         """
@@ -46,6 +58,8 @@ def __init__(
             # file is None or some file-like object
             self.file = cast(Optional[typechecking.FileLike], file)
         else:
+            if mode is None:
+                mode = self.PREFERRED_MODE
             encoding: Optional[str] = (
                 None
                 if "b" in mode
@@ -85,20 +99,22 @@ class MessageWriter(BaseIOHandler, Listener, metaclass=ABCMeta):
 
     file: Optional[typechecking.FileLike]
 
+    PREFERRED_MODE: str = "wt"
+
 
 class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):
     """A specialized base class for all writers with file descriptors."""
 
     file: typechecking.FileLike
 
-    def __init__(
-        self, file: typechecking.AcceptedIOType, mode: str = "wt", **kwargs: Any
-    ) -> None:
+    PREFERRED_MODE: str = "wt"
+
+    def __init__(self, file: typechecking.AcceptedIOType, **kwargs: Any) -> None:
         # Not possible with the type signature, but be verbose for user-friendliness
         if file is None:
             raise ValueError("The given file cannot be None")
 
-        super().__init__(file, mode, **kwargs)
+        super().__init__(file, **kwargs)
 
     def file_size(self) -> int:
         """Return an estimate of the current file size in bytes."""

can/io/logger.py

@@ -94,20 +94,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 +125,13 @@ def compress(filename: StringPathLike, **kwargs: Any) -> Tuple[str, FileLike]:
             raise ValueError(
                 f"The file type {real_suffix} is currently incompatible with gzip."
             )
+        LoggerType = cls._get_logger_for_suffix(real_suffix)
         if kwargs.get("append", False):
-            mode = "ab" if real_suffix == ".blf" else "at"
+            mode = LoggerType.PREFERRED_MODE_APPEND
         else:
-            mode = "wb" if real_suffix == ".blf" else "wt"
+            mode = LoggerType.PREFERRED_MODE
 
-        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

@@ -85,6 +85,8 @@ class MF4Writer(FileIOMessageWriter):
     It the first message does not have a timestamp, it is set to zero.
     """
 
+    PREFERRED_MODE: str = "w+b"
+
     def __init__(
         self,
         file: Union[StringPathLike, BinaryIO],
@@ -117,7 +119,7 @@ def __init__(
                 f"append messages to an existing file."
             )
 
-        super().__init__(file, mode="w+b")
+        super().__init__(file)
         now = datetime.now()
         self._mdf = cast(MDF4, MDF(version="4.10"))
         self._mdf.header.start_time = now
@@ -272,6 +274,8 @@ class MF4Reader(MessageReader):
     The MF4Reader only supports MF4 files that were recorded with python-can.
     """
 
+    PREFERRED_MODE: str = "rb"
+
     def __init__(self, file: Union[StringPathLike, BinaryIO]) -> None:
         """
         :param file: a path-like object or as file-like object to read from
@@ -284,7 +288,7 @@ def __init__(self, file: Union[StringPathLike, BinaryIO]) -> None:
                 "the optional dependency [mf4] to use the MF4Reader."
             )
 
-        super().__init__(file, mode="rb")
+        super().__init__(file)
 
         self._mdf: MDF4
         if isinstance(file, BufferedIOBase):

can/io/player.py

@@ -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,21 @@ 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 = ReaderType.PREFERRED_MODE
 
-        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/printer.py

@@ -24,6 +24,9 @@ class Printer(MessageWriter):
 
     file: Optional[TextIO]
 
+    PREFERRED_MODE: str = "wt"
+    PREFERRED_MODE_APPEND: str = "at"
+
     def __init__(
         self,
         file: Optional[Union[StringPathLike, TextIO]] = None,
@@ -39,7 +42,7 @@ def __init__(
                        else the file is truncated
         """
         self.write_to_file = file is not None
-        mode = "a" if append else "w"
+        mode = self.PREFERRED_MODE_APPEND if append else self.PREFERRED_MODE
         super().__init__(file, mode=mode)
 
     def on_message_received(self, msg: Message) -> None: