16362801995

Committed 18 Jul 2025 05:17AM UTC coverage: 70.862% (+0.1%) from 70.763%

Build # 16362801995

Build Type

Pull #1920

github

Committed by

web-flow

Commit Message

Merge f9e8a3c29 into 958fc64ed

Pull Request Pull Request #1920: add FD support to slcan according to CANable 2.0 impementation

Run Details

6 of 45 new or added lines in 1 file covered. (13.33%)

838 existing lines in 35 files now uncovered.

7770 of 10965 relevant lines covered (70.86%)

13.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.48

/can/io/logger.py

"""
See the :class:`Logger` class.
"""

import gzip
import os
import pathlib
from abc import ABC, abstractmethod
from datetime import datetime
from types import TracebackType
from typing import (
    Any,
    Callable,
    ClassVar,
    Final,
    Literal,
    Optional,
    cast,
)

from typing_extensions import Self

from .._entry_points import read_entry_points
from ..message import Message
from ..typechecking import AcceptedIOType, FileLike, StringPathLike
from .asc import ASCWriter
from .blf import BLFWriter
from .canutils import CanutilsLogWriter
from .csv import CSVWriter
from .generic import (
    BinaryIOMessageWriter,
    FileIOMessageWriter,
    MessageWriter,
)
from .mf4 import MF4Writer
from .printer import Printer
from .sqlite import SqliteWriter
from .trc import TRCWriter

#: A map of file suffixes to their corresponding
#: :class:`can.io.generic.MessageWriter` class
MESSAGE_WRITERS: Final[dict[str, type[MessageWriter]]] = {
    ".asc": ASCWriter,
    ".blf": BLFWriter,
    ".csv": CSVWriter,
    ".db": SqliteWriter,
    ".log": CanutilsLogWriter,
    ".mf4": MF4Writer,
    ".trc": TRCWriter,
    ".txt": Printer,
}


def _update_writer_plugins() -> None:
    """Update available message writer plugins from entry points."""
    for entry_point in read_entry_points("can.io.message_writer"):
        if entry_point.key in MESSAGE_WRITERS:
            continue

        writer_class = entry_point.load()
        if issubclass(writer_class, MessageWriter):
            MESSAGE_WRITERS[entry_point.key] = writer_class


def _get_logger_for_suffix(suffix: str) -> type[MessageWriter]:
    try:
        return MESSAGE_WRITERS[suffix]
    except KeyError:
        raise ValueError(
            f'No write support for unknown log format "{suffix}"'
        ) from None


def _compress(
    filename: StringPathLike, **kwargs: Any
) -> tuple[type[MessageWriter], FileLike]:
    """
    Return the suffix and io object of the decompressed file.
    File will automatically recompress upon close.
    """
    suffixes = pathlib.Path(filename).suffixes
    if len(suffixes) != 2:
        raise ValueError(
            f"No write support for unknown log format \"{''.join(suffixes)}\""
        ) from None

    real_suffix = suffixes[-2].lower()
    if real_suffix in (".blf", ".db"):
        raise ValueError(
            f"The file type {real_suffix} is currently incompatible with gzip."
        )
    logger_type = _get_logger_for_suffix(real_suffix)
    append = kwargs.get("append", False)

    if issubclass(logger_type, BinaryIOMessageWriter):
        mode = "ab" if append else "wb"
    else:
        mode = "at" if append else "wt"

    return logger_type, gzip.open(filename, mode)


def Logger(  # noqa: N802
    filename: Optional[StringPathLike], **kwargs: Any
) -> MessageWriter:
    """Find and return the appropriate :class:`~can.io.generic.MessageWriter` instance
    for a given file suffix.

    The format is determined from the file suffix which can be one of:
      * .asc :class:`can.ASCWriter`
      * .blf :class:`can.BLFWriter`
      * .csv: :class:`can.CSVWriter`
      * .db :class:`can.SqliteWriter`
      * .log :class:`can.CanutilsLogWriter`
      * .mf4 :class:`can.MF4Writer`
        (optional, depends on `asammdf <https://github.com/danielhrisca/asammdf>`_)
      * .trc :class:`can.TRCWriter`
      * .txt :class:`can.Printer`

    Any of these formats can be used with gzip compression by appending
    the suffix .gz (e.g. filename.asc.gz). However, third-party tools might not
    be able to read these files.

    The **filename** may also be *None*, to fall back to :class:`can.Printer`.

    The log files may be incomplete until `stop()` is called due to buffering.

    :param filename:
        the filename/path of the file to write to,
        may be a path-like object or None to
        instantiate a :class:`~can.Printer`
    :raises ValueError:
        if the filename's suffix is of an unknown file type

    .. note::
        This function itself is just a dispatcher, and any positional and keyword
        arguments are passed on to the returned instance.
    """

    if filename is None:
        return Printer(**kwargs)

    _update_writer_plugins()

    suffix = pathlib.PurePath(filename).suffix.lower()
    file_or_filename: AcceptedIOType = filename
    if suffix == ".gz":
        logger_type, file_or_filename = _compress(filename, **kwargs)
    else:
        logger_type = _get_logger_for_suffix(suffix)
    return logger_type(file=file_or_filename, **kwargs)


class BaseRotatingLogger(MessageWriter, ABC):
    """
    Base class for rotating CAN loggers. This class is not meant to be
    instantiated directly. Subclasses must implement the :meth:`should_rollover`
    and :meth:`do_rollover` methods according to their rotation strategy.

    The rotation behavior can be further customized by the user by setting
    the :attr:`namer` and :attr:`rotator` attributes after instantiating the subclass.

    These attributes as well as the methods :meth:`rotation_filename` and :meth:`rotate`
    and the corresponding docstrings are carried over from the python builtin
    :class:`~logging.handlers.BaseRotatingHandler`.

    Subclasses must set the `_writer` attribute upon initialization.
    """

    _supported_formats: ClassVar[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`.
    namer: Optional[Callable[[StringPathLike], StringPathLike]] = None

    #: If this attribute is set to a callable, the :meth:`~BaseRotatingLogger.rotate` method
    #: delegates to this callable. The parameters passed to the callable are those
    #: passed to :meth:`~BaseRotatingLogger.rotate`.
    rotator: Optional[Callable[[StringPathLike, StringPathLike], None]] = None

    #: An integer counter to track the number of rollovers.
    rollover_count: int = 0

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**{**kwargs, "file": None})

        self.writer_kwargs = kwargs

    @property
    @abstractmethod
    def writer(self) -> FileIOMessageWriter:
        """This attribute holds an instance of a writer class which manages the actual file IO."""
        raise NotImplementedError

    def rotation_filename(self, default_name: StringPathLike) -> StringPathLike:
        """Modify the filename of a log file when rotating.

        This is provided so that a custom filename can be provided.
        The default implementation calls the :attr:`namer` attribute of the
        handler, if it's callable, passing the default name to
        it. If the attribute isn't callable (the default is :obj:`None`), the name
        is returned unchanged.

        :param default_name:
            The default name for the log file.
        """
        if not callable(self.namer):
            return default_name

        return self.namer(default_name)  # pylint: disable=not-callable

    def rotate(self, source: StringPathLike, dest: StringPathLike) -> None:
        """When rotating, rotate the current log.

        The default implementation calls the :attr:`rotator` attribute of the
        handler, if it's callable, passing the `source` and `dest` arguments to
        it. If the attribute isn't callable (the default is :obj:`None`), the source
        is simply renamed to the destination.

        :param source:
            The source filename. This is normally the base
            filename, e.g. `"test.log"`
        :param dest:
            The destination filename. This is normally
            what the source is rotated to, e.g. `"test_#001.log"`.
        """
        if not callable(self.rotator):
            if os.path.exists(source):
                os.rename(source, dest)
        else:
            self.rotator(source, dest)  # pylint: disable=not-callable

    def on_message_received(self, msg: Message) -> None:
        """This method is called to handle the given message.

        :param msg:
            the delivered message
        """
        if self.should_rollover(msg):
            self.do_rollover()
            self.rollover_count += 1

        self.writer.on_message_received(msg)

    def _get_new_writer(self, filename: StringPathLike) -> FileIOMessageWriter:
        """Instantiate a new writer.

        .. note::
            The :attr:`self.writer` should be closed prior to calling this function.

        :param filename:
            Path-like object that specifies the location and name of the log file.
            The log file format is defined by the suffix of `filename`.
        :return:
            An instance of a writer class.
        """
        suffixes = pathlib.Path(filename).suffixes
        for suffix_length in range(len(suffixes), 0, -1):
            suffix = "".join(suffixes[-suffix_length:]).lower()
            if suffix not in self._supported_formats:
                continue
            logger = Logger(filename=filename, **self.writer_kwargs)
            if isinstance(logger, FileIOMessageWriter):
                return logger
            elif isinstance(logger, Printer) and logger.file is not None:
                return cast("FileIOMessageWriter", logger)

        raise ValueError(
            f'The log format of "{pathlib.Path(filename).name}" '
            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.

        Carry out any final tasks to ensure
        data is persisted and cleanup any open resources.
        """
        self.writer.stop()

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> Literal[False]:
        return self.writer.__exit__(exc_type, exc_val, exc_tb)

    @abstractmethod
    def should_rollover(self, msg: Message) -> bool:
        """Determine if the rollover conditions are met."""

    @abstractmethod
    def do_rollover(self) -> None:
        """Perform rollover."""


class SizedRotatingLogger(BaseRotatingLogger):
    """Log CAN messages to a sequence of files with a given maximum size.

    The logger creates a log file with the given `base_filename`. When the
    size threshold is reached the current log file is closed and renamed
    by adding a timestamp and the rollover count. A new log file is then
    created and written to.

    This behavior can be customized by setting the
    :attr:`~can.io.BaseRotatingLogger.namer` and
    :attr:`~can.io.BaseRotatingLogger.rotator`
    attribute.

    Example::

        from can import Notifier, SizedRotatingLogger
        from can.interfaces.vector import VectorBus

        bus = VectorBus(channel=[0], app_name="CANape", fd=True)

        logger = SizedRotatingLogger(
            base_filename="my_logfile.asc",
            max_bytes=5 * 1024 ** 2,  # =5MB
        )
        logger.rollover_count = 23  # start counter at 23

        notifier = Notifier(bus=bus, listeners=[logger])

    The SizedRotatingLogger currently supports the formats
      * .asc: :class:`can.ASCWriter`
      * .blf :class:`can.BLFWriter`
      * .csv: :class:`can.CSVWriter`
      * .log :class:`can.CanutilsLogWriter`
      * .txt :class:`can.Printer` (if pointing to a file)

    .. note::
        The :class:`can.SqliteWriter` is not supported yet.

    The log files on disk may be incomplete due to buffering until
    :meth:`~can.Listener.stop` is called.
    """

    _supported_formats: ClassVar[set[str]] = {".asc", ".blf", ".csv", ".log", ".txt"}

    def __init__(
        self,
        base_filename: StringPathLike,
        max_bytes: int = 0,
        **kwargs: Any,
    ) -> None:
        """
        :param base_filename:
            A path-like object for the base filename. The log file format is defined by
            the suffix of `base_filename`.
        :param max_bytes:
            The size threshold at which a new log file shall be created. If set to 0, no
            rollover will be performed.
        """
        super().__init__(**kwargs)

        self.base_filename = os.path.abspath(base_filename)
        self.max_bytes = max_bytes

        self._writer = self._get_new_writer(self.base_filename)

    @property
    def writer(self) -> FileIOMessageWriter:
        return self._writer

    def should_rollover(self, msg: Message) -> bool:
        if self.max_bytes <= 0:
            return False

        if self.writer.file_size() >= self.max_bytes:
            return True

        return False

    def do_rollover(self) -> None:
        if self.writer:
            self.writer.stop()

        sfn = self.base_filename
        dfn = self.rotation_filename(self._default_name())
        self.rotate(sfn, dfn)

        self._writer = self._get_new_writer(self.base_filename)

    def _default_name(self) -> StringPathLike:
        """Generate the default rotation filename."""
        path = pathlib.Path(self.base_filename)
        new_name = (
            path.stem.split(".")[0]
            + "_"
            + datetime.now().strftime("%Y-%m-%dT%H%M%S")
            + "_"
            + f"#{self.rollover_count:03}"
            + "".join(path.suffixes[-2:])
        )
        return str(path.parent / new_name)

1	"""	21✔
2	See the :class:`Logger` class.
3	"""
4
5	import gzip	21✔
6	import os	21✔
7	import pathlib	21✔
8	from abc import ABC, abstractmethod	21✔
9	from datetime import datetime	21✔
10	from types import TracebackType	21✔
11	from typing import (	21✔
12	Any,
13	Callable,
14	ClassVar,
15	Final,
16	Literal,
17	Optional,
18	cast,
19	)
20
21	from typing_extensions import Self	21✔
22
23	from .._entry_points import read_entry_points	21✔
24	from ..message import Message	21✔
25	from ..typechecking import AcceptedIOType, FileLike, StringPathLike	21✔
26	from .asc import ASCWriter	21✔
27	from .blf import BLFWriter	21✔
28	from .canutils import CanutilsLogWriter	21✔
29	from .csv import CSVWriter	21✔
30	from .generic import (	21✔
31	BinaryIOMessageWriter,
32	FileIOMessageWriter,
33	MessageWriter,
34	)
35	from .mf4 import MF4Writer	21✔
36	from .printer import Printer	21✔
37	from .sqlite import SqliteWriter	21✔
38	from .trc import TRCWriter	21✔
39
40	#: A map of file suffixes to their corresponding
41	#: :class:`can.io.generic.MessageWriter` class
42	MESSAGE_WRITERS: Final[dict[str, type[MessageWriter]]] = {	21✔
43	".asc": ASCWriter,
44	".blf": BLFWriter,
45	".csv": CSVWriter,
46	".db": SqliteWriter,
47	".log": CanutilsLogWriter,
48	".mf4": MF4Writer,
49	".trc": TRCWriter,
50	".txt": Printer,
51	}
52
53
54	def _update_writer_plugins() -> None:	21✔
55	"""Update available message writer plugins from entry points."""
56	for entry_point in read_entry_points("can.io.message_writer"):	21✔
UNCOV 57	if entry_point.key in MESSAGE_WRITERS:	×
UNCOV 58	continue	×
59
UNCOV 60	writer_class = entry_point.load()	×
61	if issubclass(writer_class, MessageWriter):	×
62	MESSAGE_WRITERS[entry_point.key] = writer_class	×
63
64
65	def _get_logger_for_suffix(suffix: str) -> type[MessageWriter]:	21✔
66	try:	21✔
67	return MESSAGE_WRITERS[suffix]	21✔
68	except KeyError:	21✔
69	raise ValueError(	21✔
70	f'No write support for unknown log format "{suffix}"'
71	) from None
72
73
74	def _compress(	21✔
75	filename: StringPathLike, **kwargs: Any
76	) -> tuple[type[MessageWriter], FileLike]:
77	"""
78	Return the suffix and io object of the decompressed file.
79	File will automatically recompress upon close.
80	"""
81	suffixes = pathlib.Path(filename).suffixes	21✔
82	if len(suffixes) != 2:	21✔
UNCOV 83	raise ValueError(	×
84	f"No write support for unknown log format \"{''.join(suffixes)}\""
85	) from None
86
87	real_suffix = suffixes[-2].lower()	21✔
88	if real_suffix in (".blf", ".db"):	21✔
UNCOV 89	raise ValueError(	×
90	f"The file type {real_suffix} is currently incompatible with gzip."
91	)
92	logger_type = _get_logger_for_suffix(real_suffix)	21✔
93	append = kwargs.get("append", False)	21✔
94
95	if issubclass(logger_type, BinaryIOMessageWriter):	21✔
UNCOV 96	mode = "ab" if append else "wb"	×
97	else:
98	mode = "at" if append else "wt"	21✔
99
100	return logger_type, gzip.open(filename, mode)	21✔
101
102
103	def Logger( # noqa: N802	21✔
104	filename: Optional[StringPathLike], **kwargs: Any
105	) -> MessageWriter:
106	"""Find and return the appropriate :class:`~can.io.generic.MessageWriter` instance
107	for a given file suffix.
108
109	The format is determined from the file suffix which can be one of:
110	* .asc :class:`can.ASCWriter`
111	* .blf :class:`can.BLFWriter`
112	* .csv: :class:`can.CSVWriter`
113	* .db :class:`can.SqliteWriter`
114	* .log :class:`can.CanutilsLogWriter`
115	* .mf4 :class:`can.MF4Writer`
116	(optional, depends on `asammdf <https://github.com/danielhrisca/asammdf>`_)
117	* .trc :class:`can.TRCWriter`
118	* .txt :class:`can.Printer`
119
120	Any of these formats can be used with gzip compression by appending
121	the suffix .gz (e.g. filename.asc.gz). However, third-party tools might not
122	be able to read these files.
123
124	The filename may also be None, to fall back to :class:`can.Printer`.
125
126	The log files may be incomplete until `stop()` is called due to buffering.
127
128	:param filename:
129	the filename/path of the file to write to,
130	may be a path-like object or None to
131	instantiate a :class:`~can.Printer`
132	:raises ValueError:
133	if the filename's suffix is of an unknown file type
134
135	.. note::
136	This function itself is just a dispatcher, and any positional and keyword
137	arguments are passed on to the returned instance.
138	"""
139
140	if filename is None:	21✔
141	return Printer(**kwargs)	21✔
142
143	_update_writer_plugins()	21✔
144
145	suffix = pathlib.PurePath(filename).suffix.lower()	21✔
146	file_or_filename: AcceptedIOType = filename	21✔
147	if suffix == ".gz":	21✔
148	logger_type, file_or_filename = _compress(filename, **kwargs)	21✔
149	else:
150	logger_type = _get_logger_for_suffix(suffix)	21✔
151	return logger_type(file=file_or_filename, **kwargs)	21✔
152
153
154	class BaseRotatingLogger(MessageWriter, ABC):	21✔
155	"""	21✔
156	Base class for rotating CAN loggers. This class is not meant to be
157	instantiated directly. Subclasses must implement the :meth:`should_rollover`
158	and :meth:`do_rollover` methods according to their rotation strategy.
159
160	The rotation behavior can be further customized by the user by setting
161	the :attr:`namer` and :attr:`rotator` attributes after instantiating the subclass.
162
163	These attributes as well as the methods :meth:`rotation_filename` and :meth:`rotate`
164	and the corresponding docstrings are carried over from the python builtin
165	:class:`~logging.handlers.BaseRotatingHandler`.
166
167	Subclasses must set the `_writer` attribute upon initialization.
168	"""
169
170	_supported_formats: ClassVar[set[str]] = set()	21✔
171
172	#: If this attribute is set to a callable, the :meth:`~BaseRotatingLogger.rotation_filename`
173	#: method delegates to this callable. The parameters passed to the callable are
174	#: those passed to :meth:`~BaseRotatingLogger.rotation_filename`.
175	namer: Optional[Callable[[StringPathLike], StringPathLike]] = None	21✔
176
177	#: If this attribute is set to a callable, the :meth:`~BaseRotatingLogger.rotate` method
178	#: delegates to this callable. The parameters passed to the callable are those
179	#: passed to :meth:`~BaseRotatingLogger.rotate`.
180	rotator: Optional[Callable[[StringPathLike, StringPathLike], None]] = None	21✔
181
182	#: An integer counter to track the number of rollovers.
183	rollover_count: int = 0	21✔
184
185	def __init__(self, **kwargs: Any) -> None:	21✔
186	super().__init__({kwargs, "file": None})	21✔
187
188	self.writer_kwargs = kwargs	21✔
189
190	@property	21✔
191	@abstractmethod	21✔
192	def writer(self) -> FileIOMessageWriter:	21✔
193	"""This attribute holds an instance of a writer class which manages the actual file IO."""
194	raise NotImplementedError
195
196	def rotation_filename(self, default_name: StringPathLike) -> StringPathLike:	21✔
197	"""Modify the filename of a log file when rotating.
198
199	This is provided so that a custom filename can be provided.
200	The default implementation calls the :attr:`namer` attribute of the
201	handler, if it's callable, passing the default name to
202	it. If the attribute isn't callable (the default is :obj:`None`), the name
203	is returned unchanged.
204
205	:param default_name:
206	The default name for the log file.
207	"""
208	if not callable(self.namer):	21✔
209	return default_name	21✔
210
211	return self.namer(default_name) # pylint: disable=not-callable	21✔
212
213	def rotate(self, source: StringPathLike, dest: StringPathLike) -> None:	21✔
214	"""When rotating, rotate the current log.
215
216	The default implementation calls the :attr:`rotator` attribute of the
217	handler, if it's callable, passing the `source` and `dest` arguments to
218	it. If the attribute isn't callable (the default is :obj:`None`), the source
219	is simply renamed to the destination.
220
221	:param source:
222	The source filename. This is normally the base
223	filename, e.g. `"test.log"`
224	:param dest:
225	The destination filename. This is normally
226	what the source is rotated to, e.g. `"test_#001.log"`.
227	"""
228	if not callable(self.rotator):	21✔
229	if os.path.exists(source):	21✔
230	os.rename(source, dest)	21✔
231	else:
232	self.rotator(source, dest) # pylint: disable=not-callable	21✔
233
234	def on_message_received(self, msg: Message) -> None:	21✔
235	"""This method is called to handle the given message.
236
237	:param msg:
238	the delivered message
239	"""
240	if self.should_rollover(msg):	21✔
241	self.do_rollover()	21✔
242	self.rollover_count += 1	21✔
243
244	self.writer.on_message_received(msg)	21✔
245
246	def _get_new_writer(self, filename: StringPathLike) -> FileIOMessageWriter:	21✔
247	"""Instantiate a new writer.
248
249	.. note::
250	The :attr:`self.writer` should be closed prior to calling this function.
251
252	:param filename:
253	Path-like object that specifies the location and name of the log file.
254	The log file format is defined by the suffix of `filename`.
255	:return:
256	An instance of a writer class.
257	"""
258	suffixes = pathlib.Path(filename).suffixes	21✔
259	for suffix_length in range(len(suffixes), 0, -1):	21✔
260	suffix = "".join(suffixes[-suffix_length:]).lower()	21✔
261	if suffix not in self._supported_formats:	21✔
262	continue	21✔
263	logger = Logger(filename=filename, **self.writer_kwargs)	21✔
264	if isinstance(logger, FileIOMessageWriter):	21✔
265	return logger	21✔
266	elif isinstance(logger, Printer) and logger.file is not None:	21✔
267	return cast("FileIOMessageWriter", logger)	21✔
268
UNCOV 269	raise ValueError(	×
270	f'The log format of "{pathlib.Path(filename).name}" '
271	f"is not supported by {self.__class__.__name__}. "
272	f"{self.__class__.__name__} supports the following formats: "
273	f"{', '.join(self._supported_formats)}"
274	)
275
276	def stop(self) -> None:	21✔
277	"""Stop handling new messages.
278
279	Carry out any final tasks to ensure
280	data is persisted and cleanup any open resources.
281	"""
282	self.writer.stop()	21✔
283
284	def __enter__(self) -> Self:	21✔
285	return self	21✔
286
287	def __exit__(	21✔
288	self,
289	exc_type: Optional[type[BaseException]],
290	exc_val: Optional[BaseException],
291	exc_tb: Optional[TracebackType],
292	) -> Literal[False]:
293	return self.writer.__exit__(exc_type, exc_val, exc_tb)	21✔
294
295	@abstractmethod	21✔
296	def should_rollover(self, msg: Message) -> bool:	21✔
297	"""Determine if the rollover conditions are met."""
298
299	@abstractmethod	21✔
300	def do_rollover(self) -> None:	21✔
301	"""Perform rollover."""
302
303
304	class SizedRotatingLogger(BaseRotatingLogger):	21✔
305	"""Log CAN messages to a sequence of files with a given maximum size.	21✔
306
307	The logger creates a log file with the given `base_filename`. When the
308	size threshold is reached the current log file is closed and renamed
309	by adding a timestamp and the rollover count. A new log file is then
310	created and written to.
311
312	This behavior can be customized by setting the
313	:attr:`~can.io.BaseRotatingLogger.namer` and
314	:attr:`~can.io.BaseRotatingLogger.rotator`
315	attribute.
316
317	Example::
318
319	from can import Notifier, SizedRotatingLogger
320	from can.interfaces.vector import VectorBus
321
322	bus = VectorBus(channel=[0], app_name="CANape", fd=True)
323
324	logger = SizedRotatingLogger(
325	base_filename="my_logfile.asc",
326	max_bytes=5 * 1024 ** 2, # =5MB
327	)
328	logger.rollover_count = 23 # start counter at 23
329
330	notifier = Notifier(bus=bus, listeners=[logger])
331
332	The SizedRotatingLogger currently supports the formats
333	* .asc: :class:`can.ASCWriter`
334	* .blf :class:`can.BLFWriter`
335	* .csv: :class:`can.CSVWriter`
336	* .log :class:`can.CanutilsLogWriter`
337	* .txt :class:`can.Printer` (if pointing to a file)
338
339	.. note::
340	The :class:`can.SqliteWriter` is not supported yet.
341
342	The log files on disk may be incomplete due to buffering until
343	:meth:`~can.Listener.stop` is called.
344	"""
345
346	_supported_formats: ClassVar[set[str]] = {".asc", ".blf", ".csv", ".log", ".txt"}	21✔
347
348	def __init__(	21✔
349	self,
350	base_filename: StringPathLike,
351	max_bytes: int = 0,
352	**kwargs: Any,
353	) -> None:
354	"""
355	:param base_filename:
356	A path-like object for the base filename. The log file format is defined by
357	the suffix of `base_filename`.
358	:param max_bytes:
359	The size threshold at which a new log file shall be created. If set to 0, no
360	rollover will be performed.
361	"""
362	super().__init__(**kwargs)	21✔
363
364	self.base_filename = os.path.abspath(base_filename)	21✔
365	self.max_bytes = max_bytes	21✔
366
367	self._writer = self._get_new_writer(self.base_filename)	21✔
368
369	@property	21✔
370	def writer(self) -> FileIOMessageWriter:	21✔
371	return self._writer	21✔
372
373	def should_rollover(self, msg: Message) -> bool:	21✔
374	if self.max_bytes <= 0:	21✔
UNCOV 375	return False	×
376
377	if self.writer.file_size() >= self.max_bytes:	21✔
378	return True	21✔
379
380	return False	21✔
381
382	def do_rollover(self) -> None:	21✔
383	if self.writer:	21✔
384	self.writer.stop()	21✔
385
386	sfn = self.base_filename	21✔
387	dfn = self.rotation_filename(self._default_name())	21✔
388	self.rotate(sfn, dfn)	21✔
389
390	self._writer = self._get_new_writer(self.base_filename)	21✔
391
392	def _default_name(self) -> StringPathLike:	21✔
393	"""Generate the default rotation filename."""
394	path = pathlib.Path(self.base_filename)	21✔
395	new_name = (	21✔
396	path.stem.split(".")[0]
397	+ "_"
398	+ datetime.now().strftime("%Y-%m-%dT%H%M%S")
399	+ "_"
400	+ f"#{self.rollover_count:03}"
401	+ "".join(path.suffixes[-2:])
402	)
403	return str(path.parent / new_name)	21✔

hardbyte / python-can / 16362801995

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous