6
6
import pathlib
7
7
from abc import ABC , abstractmethod
8
8
from datetime import datetime
9
+ import gzip
9
10
from typing import (
10
11
Any ,
11
12
Optional ,
12
13
Callable ,
13
14
cast ,
15
+ IO ,
14
16
Type ,
17
+ Tuple ,
15
18
)
16
19
from types import TracebackType
17
20
21
24
from ..message import Message
22
25
from ..listener import Listener
23
26
from .generic import BaseIOHandler , FileIOMessageWriter
24
- from .asc import ASCWriter , GzipASCWriter
27
+ from .asc import ASCWriter
25
28
from .blf import BLFWriter
26
29
from .canutils import CanutilsLogWriter
27
30
from .csv import CSVWriter
@@ -36,13 +39,14 @@ class Logger(BaseIOHandler, Listener): # pylint: disable=abstract-method
36
39
37
40
The format is determined from the file format which can be one of:
38
41
* .asc: :class:`can.ASCWriter`
39
- * .asc.gz: :class:`can.CompressedASCWriter`
40
42
* .blf :class:`can.BLFWriter`
41
43
* .csv: :class:`can.CSVWriter`
42
44
* .db: :class:`can.SqliteWriter`
43
45
* .log :class:`can.CanutilsLogWriter`
44
46
* .txt :class:`can.Printer`
45
47
48
+ Or any of the above compressed using gzip (.gz)
49
+
46
50
The **filename** may also be *None*, to fall back to :class:`can.Printer`.
47
51
48
52
The log files may be incomplete until `stop()` is called due to buffering.
@@ -55,7 +59,6 @@ class Logger(BaseIOHandler, Listener): # pylint: disable=abstract-method
55
59
fetched_plugins = False
56
60
message_writers = {
57
61
".asc" : ASCWriter ,
58
- ".asc.gz" : GzipASCWriter ,
59
62
".blf" : BLFWriter ,
60
63
".csv" : CSVWriter ,
61
64
".db" : SqliteWriter ,
@@ -85,7 +88,11 @@ def __new__( # type: ignore
85
88
)
86
89
Logger .fetched_plugins = True
87
90
88
- suffix = "" .join (s .lower () for s in pathlib .PurePath (filename ).suffixes )
91
+ suffix = pathlib .PurePath (filename ).suffix .lower ()
92
+
93
+ if suffix == ".gz" :
94
+ suffix , filename = Logger .compress (filename )
95
+
89
96
try :
90
97
return cast (
91
98
Listener , Logger .message_writers [suffix ](filename , * args , ** kwargs )
@@ -95,6 +102,17 @@ def __new__( # type: ignore
95
102
f'No write support for this unknown log format "{ suffix } "'
96
103
) from None
97
104
105
+ @staticmethod
106
+ def compress (filename : StringPathLike ) -> Tuple [str , IO [Any ]]:
107
+ """
108
+ Return the suffix and io object of the decompressed file.
109
+ File will automatically recompress upon close.
110
+ """
111
+ real_suffix = pathlib .Path (filename ).suffixes [- 2 ].lower ()
112
+ mode = "ab" if real_suffix == ".blf" else "at"
113
+
114
+ return real_suffix , gzip .open (filename , mode )
115
+
98
116
99
117
class BaseRotatingLogger (Listener , BaseIOHandler , ABC ):
100
118
"""
0 commit comments