forked from hardbyte/python-can
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgeneric.py
More file actions
109 lines (87 loc) · 3.25 KB
/
Copy pathgeneric.py
File metadata and controls
109 lines (87 loc) · 3.25 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
"""Contains generic base classes for file IO."""
import locale
from abc import ABCMeta
from typing import (
Optional,
cast,
Iterable,
Type,
ContextManager,
Any,
)
from typing_extensions import Literal
from types import TracebackType
import can
import can.typechecking
class BaseIOHandler(ContextManager, metaclass=ABCMeta):
"""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[can.typechecking.FileLike]
def __init__(
self,
file: Optional[can.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[can.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(
can.typechecking.FileLike,
open(
cast(can.typechecking.StringPathLike, file), mode, encoding=encoding
),
)
# for multiple inheritance
super().__init__()
def __enter__(self) -> "BaseIOHandler":
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, can.Listener, metaclass=ABCMeta):
"""The base class for all writers."""
file: Optional[can.typechecking.FileLike]
class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):
"""A specialized base class for all writers with file descriptors."""
file: can.typechecking.FileLike
def __init__(
self, file: can.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 MessageReader(BaseIOHandler, Iterable[can.Message], metaclass=ABCMeta):
"""The base class for all readers."""