"""This module provides abstract base classes for CAN message reading and writing operations
to various file formats.
.. note::
All classes in this module are abstract and should be subclassed to implement
specific file format handling.
"""
import locale
import os
from abc import ABC, abstractmethod
from collections.abc import Iterable
from contextlib import AbstractContextManager
from io import BufferedIOBase, TextIOWrapper
from pathlib import Path
from types import TracebackType
from typing import (
TYPE_CHECKING,
Any,
BinaryIO,
Generic,
Literal,
TextIO,
TypeVar,
)
from typing_extensions import Self
from ..listener import Listener
from ..message import Message
from ..typechecking import FileLike, StringPathLike
if TYPE_CHECKING:
from _typeshed import (
OpenBinaryModeReading,
OpenBinaryModeUpdating,
OpenBinaryModeWriting,
OpenTextModeReading,
OpenTextModeUpdating,
OpenTextModeWriting,
)
#: type parameter used in generic classes :class:`MessageReader` and :class:`MessageWriter`
_IoTypeVar = TypeVar("_IoTypeVar", bound=FileLike)
[docs]
class MessageWriter(AbstractContextManager["MessageWriter"], Listener, ABC):
"""Abstract base class for all CAN message writers.
This class serves as a foundation for implementing different message writer formats.
It combines context manager capabilities with the message listener interface.
:param file: Path-like object or string representing the output file location
:param kwargs: Additional keyword arguments for specific writer implementations
"""
@abstractmethod
def __init__(self, file: StringPathLike, **kwargs: Any) -> None:
pass
[docs]
@abstractmethod
def stop(self) -> None:
"""Stop handling messages and cleanup any resources."""
def __enter__(self) -> Self:
"""Enter the context manager."""
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> Literal[False]:
"""Exit the context manager and ensure proper cleanup."""
self.stop()
return False
[docs]
class SizedMessageWriter(MessageWriter, ABC):
"""Abstract base class for message writers that can report their file size.
This class extends :class:`MessageWriter` with the ability to determine the size
of the output file.
"""
[docs]
@abstractmethod
def file_size(self) -> int:
"""Get the current size of the output file in bytes.
:return: The size of the file in bytes
:rtype: int
"""
[docs]
class FileIOMessageWriter(SizedMessageWriter, Generic[_IoTypeVar]):
"""Base class for writers that operate on file descriptors.
This class provides common functionality for writers that work with file objects.
:param file: A path-like object or file object to write to
:param kwargs: Additional keyword arguments for specific writer implementations
:ivar file: The file object being written to
"""
file: _IoTypeVar
@abstractmethod
def __init__(self, file: StringPathLike | _IoTypeVar, **kwargs: Any) -> None:
pass
[docs]
def stop(self) -> None:
"""Close the file and stop writing."""
self.file.close()
[docs]
def file_size(self) -> int:
"""Get the current file size."""
return self.file.tell()
[docs]
class TextIOMessageWriter(FileIOMessageWriter[TextIO | TextIOWrapper], ABC):
"""Text-based message writer implementation.
:param file: Text file to write to
:param mode: File open mode for text operations
:param kwargs: Additional arguments like encoding
"""
def __init__(
self,
file: StringPathLike | TextIO | TextIOWrapper,
mode: "OpenTextModeUpdating | OpenTextModeWriting" = "w",
**kwargs: Any,
) -> None:
if isinstance(file, (str, os.PathLike)):
encoding: str = kwargs.get("encoding", locale.getpreferredencoding(False))
# pylint: disable=consider-using-with
self.file = Path(file).open(mode=mode, encoding=encoding)
else:
self.file = file
[docs]
class BinaryIOMessageWriter(FileIOMessageWriter[BinaryIO | BufferedIOBase], ABC):
"""Binary file message writer implementation.
:param file: Binary file to write to
:param mode: File open mode for binary operations
:param kwargs: Additional implementation specific arguments
"""
def __init__( # pylint: disable=unused-argument
self,
file: StringPathLike | BinaryIO | BufferedIOBase,
mode: "OpenBinaryModeUpdating | OpenBinaryModeWriting" = "wb",
**kwargs: Any,
) -> None:
if isinstance(file, (str, os.PathLike)):
# pylint: disable=consider-using-with,unspecified-encoding
self.file = Path(file).open(mode=mode)
else:
self.file = file
[docs]
class MessageReader(AbstractContextManager["MessageReader"], Iterable[Message], ABC):
"""Abstract base class for all CAN message readers.
This class serves as a foundation for implementing different message reader formats.
It combines context manager capabilities with iteration interface.
:param file: Path-like object or string representing the input file location
:param kwargs: Additional keyword arguments for specific reader implementations
"""
@abstractmethod
def __init__(self, file: StringPathLike, **kwargs: Any) -> None:
pass
[docs]
@abstractmethod
def stop(self) -> None:
"""Stop reading messages and cleanup any resources."""
def __enter__(self) -> Self:
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> Literal[False]:
self.stop()
return False
[docs]
class FileIOMessageReader(MessageReader, Generic[_IoTypeVar]):
"""Base class for readers that operate on file descriptors.
This class provides common functionality for readers that work with file objects.
:param file: A path-like object or file object to read from
:param kwargs: Additional keyword arguments for specific reader implementations
:ivar file: The file object being read from
"""
file: _IoTypeVar
@abstractmethod
def __init__(self, file: StringPathLike | _IoTypeVar, **kwargs: Any) -> None:
pass
[docs]
def stop(self) -> None:
self.file.close()
[docs]
class TextIOMessageReader(FileIOMessageReader[TextIO | TextIOWrapper], ABC):
"""Text-based message reader implementation.
:param file: Text file to read from
:param mode: File open mode for text operations
:param kwargs: Additional arguments like encoding
"""
def __init__(
self,
file: StringPathLike | TextIO | TextIOWrapper,
mode: "OpenTextModeReading" = "r",
**kwargs: Any,
) -> None:
if isinstance(file, (str, os.PathLike)):
encoding: str = kwargs.get("encoding", locale.getpreferredencoding(False))
# pylint: disable=consider-using-with
self.file = Path(file).open(mode=mode, encoding=encoding)
else:
self.file = file
[docs]
class BinaryIOMessageReader(FileIOMessageReader[BinaryIO | BufferedIOBase], ABC):
"""Binary file message reader implementation.
:param file: Binary file to read from
:param mode: File open mode for binary operations
:param kwargs: Additional implementation specific arguments
"""
def __init__( # pylint: disable=unused-argument
self,
file: StringPathLike | BinaryIO | BufferedIOBase,
mode: "OpenBinaryModeReading" = "rb",
**kwargs: Any,
) -> None:
if isinstance(file, (str, os.PathLike)):
# pylint: disable=consider-using-with,unspecified-encoding
self.file = Path(file).open(mode=mode)
else:
self.file = file