Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 20 additions & 14 deletions can/io/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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`.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
13 changes: 10 additions & 3 deletions can/io/printer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
2 changes: 2 additions & 0 deletions test/test_rotating_loggers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down