Source code for

# coding: utf-8

Contains a generic class for file IO.

from abc import ABCMeta
from typing import Optional, cast

import can
import can.typechecking

[docs]class BaseIOHandler(metaclass=ABCMeta): """A generic file handler that can be used for reading and writing. Can be used as a context manager. :attr Optional[FileLike] file: the file-like object that is kept internally, or None if none was opened """ def __init__(self, file: can.typechecking.AcceptedIOType, mode: str = "rt") -> None: """ :param file: a path-like object to open a file, a file-like object to be used as a file or `None` to not use a file at all :param mode: the mode that should be used to open the file, see :func:`open`, ignored if *file* is `None` """ if file is None or (hasattr(file, "read") and hasattr(file, "write")): # file is None or some file-like object self.file = cast(Optional[can.typechecking.FileLike], file) else: # file is some path-like object self.file = open(cast(can.typechecking.StringPathLike, file), mode) # for multiple inheritance super().__init__() def __enter__(self) -> "BaseIOHandler": return self def __exit__(self, *args) -> None: self.stop()
[docs] def stop(self) -> None: """Closes the undelying file-like object and flushes it, if it was opened in write mode.""" if self.file is not None: # this also implies a flush() self.file.close()
# pylint: disable=abstract-method,too-few-public-methods
[docs]class MessageWriter(BaseIOHandler, can.Listener, metaclass=ABCMeta): """The base class for all writers."""
# pylint: disable=too-few-public-methods
[docs]class MessageReader(BaseIOHandler, metaclass=ABCMeta): """The base class for all readers."""