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

98.04

/can/io/generic.py

"""Contains generic base classes for file IO."""

import gzip
import locale
from abc import ABCMeta
from collections.abc import Iterable
from contextlib import AbstractContextManager
from types import TracebackType
from typing import (
    Any,
    BinaryIO,
    Literal,
    Optional,
    TextIO,
    Union,
    cast,
)

from typing_extensions import Self

from .. import typechecking
from ..listener import Listener
from ..message import Message


class BaseIOHandler(AbstractContextManager):
    """A generic file handler that can be used for reading and writing.

    Can be used as a context manager.

    :attr file:
        the file-like object that is kept internally, or `None` if none
        was opened
    """

    file: Optional[typechecking.FileLike]

    def __init__(
        self,
        file: Optional[typechecking.AcceptedIOType],
        mode: str = "rt",
        **kwargs: Any,
    ) -> 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[typechecking.FileLike]", file)
        else:
            encoding: Optional[str] = (
                None
                if "b" in mode
                else kwargs.get("encoding", locale.getpreferredencoding(False))
            )
            # pylint: disable=consider-using-with
            # file is some path-like object
            self.file = cast(
                "typechecking.FileLike", open(file, mode, encoding=encoding)
            )

        # for multiple inheritance
        super().__init__()

    def __enter__(self) -> Self:
        return self

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

    def stop(self) -> None:
        """Closes the underlying 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()


class MessageWriter(BaseIOHandler, Listener, metaclass=ABCMeta):
    """The base class for all writers."""

    file: Optional[typechecking.FileLike]


class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):
    """A specialized base class for all writers with file descriptors."""

    file: typechecking.FileLike

    def __init__(
        self, file: typechecking.AcceptedIOType, mode: str = "wt", **kwargs: Any
    ) -> None:
        # Not possible with the type signature, but be verbose for user-friendliness
        if file is None:
            raise ValueError("The given file cannot be None")

        super().__init__(file, mode, **kwargs)

    def file_size(self) -> int:
        """Return an estimate of the current file size in bytes."""
        return self.file.tell()


class TextIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):
    file: TextIO


class BinaryIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):
    file: Union[BinaryIO, gzip.GzipFile]


class MessageReader(BaseIOHandler, Iterable[Message], metaclass=ABCMeta):
    """The base class for all readers."""


class TextIOMessageReader(MessageReader, metaclass=ABCMeta):
    file: TextIO


class BinaryIOMessageReader(MessageReader, metaclass=ABCMeta):
    file: Union[BinaryIO, gzip.GzipFile]

1	"""Contains generic base classes for file IO."""	21✔
2
3	import gzip	21✔
4	import locale	21✔
5	from abc import ABCMeta	21✔
6	from collections.abc import Iterable	21✔
7	from contextlib import AbstractContextManager	21✔
8	from types import TracebackType	21✔
9	from typing import (	21✔
10	Any,
11	BinaryIO,
12	Literal,
13	Optional,
14	TextIO,
15	Union,
16	cast,
17	)
18
19	from typing_extensions import Self	21✔
20
21	from .. import typechecking	21✔
22	from ..listener import Listener	21✔
23	from ..message import Message	21✔
24
25
26	class BaseIOHandler(AbstractContextManager):	21✔
27	"""A generic file handler that can be used for reading and writing.	21✔
28
29	Can be used as a context manager.
30
31	:attr file:
32	the file-like object that is kept internally, or `None` if none
33	was opened
34	"""
35
36	file: Optional[typechecking.FileLike]	21✔
37
38	def __init__(	21✔
39	self,
40	file: Optional[typechecking.AcceptedIOType],
41	mode: str = "rt",
42	**kwargs: Any,
43	) -> None:
44	"""
45	:param file: a path-like object to open a file, a file-like object
46	to be used as a file or `None` to not use a file at all
47	:param mode: the mode that should be used to open the file, see
48	:func:`open`, ignored if file is `None`
49	"""
50	if file is None or (hasattr(file, "read") and hasattr(file, "write")):	21✔
51	# file is None or some file-like object
52	self.file = cast("Optional[typechecking.FileLike]", file)	21✔
53	else:
54	encoding: Optional[str] = (	21✔
55	None
56	if "b" in mode
57	else kwargs.get("encoding", locale.getpreferredencoding(False))
58	)
59	# pylint: disable=consider-using-with
60	# file is some path-like object
61	self.file = cast(	21✔
62	"typechecking.FileLike", open(file, mode, encoding=encoding)
63	)
64
65	# for multiple inheritance
66	super().__init__()	21✔
67
68	def __enter__(self) -> Self:	21✔
69	return self	21✔
70
71	def __exit__(	21✔
72	self,
73	exc_type: Optional[type[BaseException]],
74	exc_val: Optional[BaseException],
75	exc_tb: Optional[TracebackType],
76	) -> Literal[False]:
77	self.stop()	21✔
78	return False	21✔
79
80	def stop(self) -> None:	21✔
81	"""Closes the underlying file-like object and flushes it, if it was opened in write mode."""
82	if self.file is not None:	21✔
83	# this also implies a flush()
84	self.file.close()	21✔
85
86
87	class MessageWriter(BaseIOHandler, Listener, metaclass=ABCMeta):	21✔
88	"""The base class for all writers."""	21✔
89
90	file: Optional[typechecking.FileLike]	21✔
91
92
93	class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):	21✔
94	"""A specialized base class for all writers with file descriptors."""	21✔
95
96	file: typechecking.FileLike	21✔
97
98	def __init__(	21✔
99	self, file: typechecking.AcceptedIOType, mode: str = "wt", **kwargs: Any
100	) -> None:
101	# Not possible with the type signature, but be verbose for user-friendliness
102	if file is None:	21✔
UNCOV 103	raise ValueError("The given file cannot be None")	×
104
105	super().__init__(file, mode, **kwargs)	21✔
106
107	def file_size(self) -> int:	21✔
108	"""Return an estimate of the current file size in bytes."""
109	return self.file.tell()	21✔
110
111
112	class TextIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):	21✔
113	file: TextIO	21✔
114
115
116	class BinaryIOMessageWriter(FileIOMessageWriter, metaclass=ABCMeta):	21✔
117	file: Union[BinaryIO, gzip.GzipFile]	21✔
118
119
120	class MessageReader(BaseIOHandler, Iterable[Message], metaclass=ABCMeta):	21✔
121	"""The base class for all readers."""	21✔
122
123
124	class TextIOMessageReader(MessageReader, metaclass=ABCMeta):	21✔
125	file: TextIO	21✔
126
127
128	class BinaryIOMessageReader(MessageReader, metaclass=ABCMeta):	21✔
129	file: Union[BinaryIO, gzip.GzipFile]	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