Reworking the Gzipped ASC Reader/Writer

Author: pierreluctg

can/__init__.py

@@ -25,7 +25,7 @@
 )
 
 from .io import Logger, SizedRotatingLogger, Printer, LogReader, MessageSync
-from .io import ASCWriter, ASCReader, CompressedASCWriter, CompressedASCReader
+from .io import ASCWriter, ASCReader, GzipASCWriter, GzipASCReader
 from .io import BLFReader, BLFWriter
 from .io import CanutilsLogReader, CanutilsLogWriter
 from .io import CSVWriter, CSVReader

can/io/__init__.py

@@ -8,7 +8,7 @@
 from .player import LogReader, MessageSync
 
 # Format specific
-from .asc import ASCWriter, ASCReader, CompressedASCWriter, CompressedASCReader
+from .asc import ASCWriter, ASCReader, GzipASCWriter, GzipASCReader
 from .blf import BLFReader, BLFWriter
 from .canutils import CanutilsLogReader, CanutilsLogWriter
 from .csv import CSVWriter, CSVReader

can/io/asc.py

@@ -6,12 +6,13 @@
     - under `test/data/logfile.asc`
 """
 import gzip
-from typing import cast, Any, Generator, IO, List, Optional, Dict
+from typing import cast, Any, Generator, IO, List, Optional, Dict, Union
 
 from datetime import datetime
 import time
 import logging
 
+from .. import typechecking
 from ..message import Message
 from ..listener import Listener
 from ..util import channel2int
@@ -398,7 +399,7 @@ def on_message_received(self, msg: Message) -> None:
         self.log_event(serialized, msg.timestamp)
 
 
-class CompressedASCReader(ASCReader):
+class GzipASCReader(ASCReader):
     """Gzipped version of :class:`~can.ASCReader`"""
 
     def __init__(
@@ -418,12 +419,21 @@ def __init__(
                      `relative` (starting at 0.0) or `absolute` (starting at
                      the system time). Default `True = relative`.
         """
-        super(CompressedASCReader, self).__init__(
+        self._fileobj = None
+        if file is not None and (hasattr(file, "read") and hasattr(file, "write")):
+            # file is None or some file-like object
+            self._fileobj = file
+        super(GzipASCReader, self).__init__(
             gzip.open(file, mode="rt"), base, relative_timestamp
         )
 
+    def stop(self) -> None:
+        super(GzipASCReader, self).stop()
+        if self._fileobj is not None:
+            self._fileobj.close()
+
 
-class CompressedASCWriter(ASCWriter):
+class GzipASCWriter(ASCWriter):
     """Gzipped version of :class:`~can.ASCWriter`"""
 
     def __init__(
@@ -441,6 +451,15 @@ def __init__(
         :param compresslevel: Gzip compresslevel, see
                               :class:`~gzip.GzipFile` for details. The default is 6.
         """
-        super(CompressedASCWriter, self).__init__(
+        self._fileobj = None
+        if file is not None and (hasattr(file, "read") and hasattr(file, "write")):
+            # file is None or some file-like object
+            self._fileobj = file
+        super(GzipASCWriter, self).__init__(
             gzip.open(file, mode="wt", compresslevel=compresslevel), channel
         )
+
+    def stop(self) -> None:
+        super(GzipASCWriter, self).stop()
+        if self._fileobj is not None:
+            self._fileobj.close()

can/io/logger.py

@@ -21,7 +21,7 @@
 from ..message import Message
 from ..listener import Listener
 from .generic import BaseIOHandler, FileIOMessageWriter
-from .asc import ASCWriter, CompressedASCWriter
+from .asc import ASCWriter, GzipASCWriter
 from .blf import BLFWriter
 from .canutils import CanutilsLogWriter
 from .csv import CSVWriter
@@ -55,7 +55,7 @@ class Logger(BaseIOHandler, Listener):  # pylint: disable=abstract-method
     fetched_plugins = False
     message_writers = {
         ".asc": ASCWriter,
-        ".asc.gz": CompressedASCWriter,
+        ".asc.gz": GzipASCWriter,
         ".blf": BLFWriter,
         ".csv": CSVWriter,
         ".db": SqliteWriter,
@@ -85,7 +85,7 @@ def __new__(  # type: ignore
             )
             Logger.fetched_plugins = True
 
-        suffix = pathlib.PurePath(filename).suffix.lower()
+        suffix = "".join(s.lower() for s in pathlib.PurePath(filename).suffixes)
         try:
             return cast(
                 Listener, Logger.message_writers[suffix](filename, *args, **kwargs)

can/io/player.py

@@ -14,7 +14,7 @@
     import can
 
 from .generic import BaseIOHandler, MessageReader
-from .asc import ASCReader, CompressedASCReader
+from .asc import ASCReader, GzipASCReader
 from .blf import BLFReader
 from .canutils import CanutilsLogReader
 from .csv import CSVReader
@@ -50,7 +50,7 @@ class LogReader(BaseIOHandler):
     fetched_plugins = False
     message_readers = {
         ".asc": ASCReader,
-        ".asc.gz": CompressedASCReader,
+        ".asc.gz": GzipASCReader,
         ".blf": BLFReader,
         ".csv": CSVReader,
         ".db": SqliteReader,
@@ -77,7 +77,7 @@ def __new__(  # type: ignore
             )
             LogReader.fetched_plugins = True
 
-        suffix = pathlib.PurePath(filename).suffix.lower()
+        suffix = "".join(s.lower() for s in pathlib.PurePath(filename).suffixes)
         try:
             return typing.cast(
                 MessageReader,

test/logformats_test.py

@@ -11,7 +11,7 @@
 
 TODO: correctly set preserves_channel and adds_default_channel
 """
-
+import gzip
 import logging
 import unittest
 import tempfile
@@ -555,6 +555,34 @@ def test_can_and_canfd_error_frames(self):
         self.assertMessagesEqual(actual, expected_messages)
 
 
+class TestGzipASCFileFormat(ReaderWriterTest):
+    """Tests can.GzipASCWriter and can.GzipASCReader"""
+
+    def _setup_instance(self):
+        super()._setup_instance_helper(
+            can.GzipASCWriter,
+            can.GzipASCReader,
+            binary_file=True,
+            check_comments=True,
+            preserves_channel=False,
+            adds_default_channel=0,
+        )
+
+    def assertIncludesComments(self, filename):
+        """
+        Ensures that all comments are literally contained in the given file.
+
+        :param filename: the path-like object to use
+        """
+        if self.original_comments:
+            # read the entire outout file
+            with gzip.open(filename, "rt" if self.binary_file else "r") as file:
+                output_contents = file.read()
+            # check each, if they can be found in there literally
+            for comment in self.original_comments:
+                self.assertIn(comment, output_contents)
+
+
 class TestBlfFileFormat(ReaderWriterTest):
     """Tests can.BLFWriter and can.BLFReader.