explicitly set supported file formats (#1406)

Author: zariiii9003

can/io/logger.py

@@ -7,14 +7,13 @@
 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, Set
 
 from types import TracebackType
 
 from typing_extensions import Literal
 from pkg_resources import iter_entry_points
 
-import can.io
 from ..message import Message
 from ..listener import Listener
 from .generic import BaseIOHandler, FileIOMessageWriter, MessageWriter
@@ -131,9 +130,10 @@ class BaseRotatingLogger(Listener, BaseIOHandler, ABC):
     :class:`~logging.handlers.BaseRotatingHandler`.
 
     Subclasses must set the `_writer` attribute upon initialization.
-
     """
 
+    _supported_formats: Set[str] = set()
+
     #: If this attribute is set to a callable, the :meth:`~BaseRotatingLogger.rotation_filename`
     #: method delegates to this callable. The parameters passed to the callable are
     #: those passed to :meth:`~BaseRotatingLogger.rotation_filename`.
@@ -224,17 +224,21 @@ def _get_new_writer(self, filename: StringPathLike) -> FileIOMessageWriter:
         :return:
             An instance of a writer class.
         """
-
-        logger = Logger(filename, *self.writer_args, **self.writer_kwargs)
-        if isinstance(logger, FileIOMessageWriter):
-            return logger
-        elif isinstance(logger, Printer) and logger.file is not None:
-            return cast(FileIOMessageWriter, logger)
-        else:
-            raise Exception(
-                f"The log format \"{''.join(pathlib.Path(filename).suffixes[-2:])}"
-                f'" is not supported by {self.__class__.__name__}'
-            )
+        suffix = "".join(pathlib.Path(filename).suffixes[-2:]).lower()
+
+        if suffix in self._supported_formats:
+            logger = Logger(filename, *self.writer_args, **self.writer_kwargs)
+            if isinstance(logger, FileIOMessageWriter):
+                return logger
+            elif isinstance(logger, Printer) and logger.file is not None:
+                return cast(FileIOMessageWriter, logger)
+
+        raise Exception(
+            f'The log format "{suffix}" '
+            f"is not supported by {self.__class__.__name__}. "
+            f"{self.__class__.__name__} supports the following formats: "
+            f"{', '.join(self._supported_formats)}"
+        )
 
     def stop(self) -> None:
         """Stop handling new messages.
@@ -306,6 +310,8 @@ class SizedRotatingLogger(BaseRotatingLogger):
     :meth:`~can.Listener.stop` is called.
     """
 
+    _supported_formats = {".asc", ".blf", ".csv", ".log", ".txt"}
+
     def __init__(
         self,
         base_filename: StringPathLike,

can/io/printer.py

@@ -4,7 +4,7 @@
 
 import logging
 
-from typing import Optional, TextIO, Union, Any
+from typing import Optional, TextIO, Union, Any, cast
 
 from ..message import Message
 from .generic import MessageWriter
@@ -40,11 +40,18 @@ def __init__(
         :param append: If set to `True` messages, are appended to the file,
                        else the file is truncated
         """
+        self.write_to_file = file is not None
         mode = "a" if append else "w"
         super().__init__(file, mode=mode)
 
     def on_message_received(self, msg: Message) -> None:
-        if self.file is not None:
-            self.file.write(str(msg) + "\n")
+        if self.write_to_file:
+            cast(TextIO, self.file).write(str(msg) + "\n")
         else:
             print(msg)
+
+    def file_size(self) -> int:
+        """Return an estimate of the current file size in bytes."""
+        if self.file is not None:
+            return self.file.tell()
+        return 0

test/test_rotating_loggers.py

@@ -18,6 +18,8 @@ def _get_instance(path, *args, **kwargs) -> can.io.BaseRotatingLogger:
         class SubClass(can.io.BaseRotatingLogger):
             """Subclass that implements abstract methods for testing."""
 
+            _supported_formats = {".asc", ".blf", ".csv", ".log", ".txt"}
+
             def __init__(self, *args, **kwargs) -> None:
                 super().__init__(*args, **kwargs)
                 self._writer = can.Printer(file=path / "__unused.txt")