improve io typing

Author: zariiii9003

can/__init__.py

@@ -24,13 +24,6 @@
  CanTimeoutError,
 )
 
-from .io import Logger, SizedRotatingLogger, Printer, LogReader, MessageSync
-from .io import ASCWriter, ASCReader
-from .io import BLFReader, BLFWriter
-from .io import CanutilsLogReader, CanutilsLogWriter
-from .io import CSVWriter, CSVReader
-from .io import SqliteWriter, SqliteReader
-
 from .util import set_logging_level
 
 from .message import Message
@@ -42,6 +35,13 @@
 from .interface import Bus, detect_available_configs
 from .bit_timing import BitTiming
 
+from .io import Logger, SizedRotatingLogger, Printer, LogReader, MessageSync
+from .io import ASCWriter, ASCReader
+from .io import BLFReader, BLFWriter
+from .io import CanutilsLogReader, CanutilsLogWriter
+from .io import CSVWriter, CSVReader
+from .io import SqliteWriter, SqliteReader
+
 from .broadcastmanager import (
  CyclicSendTaskABC,
  LimitedDurationCyclicSendTaskABC,

can/io/asc.py

@@ -12,12 +12,10 @@
 import time
 import logging
 
-from .. import typechecking
 from ..message import Message
-from ..listener import Listener
 from ..util import channel2int
-from .generic import BaseIOHandler, FileIOMessageWriter
-from ..typechecking import AcceptedIOType
+from .generic import FileIOMessageWriter, MessageReader
+from ..typechecking import StringPathLike
 
 
 CAN_MSG_EXT = 0x80000000
@@ -28,7 +26,7 @@
 logger = logging.getLogger("can.io.asc")
 
 
-class ASCReader(BaseIOHandler):
+class ASCReader(MessageReader):
  """
  Iterator of CAN messages from a ASC logging file. Meta data (comments,
  bus statistics, J1939 Transport Protocol messages) is ignored.
@@ -40,7 +38,7 @@ class ASCReader(BaseIOHandler):
 
  def __init__(
  self,
- file: AcceptedIOType,
+ file: Union[StringPathLike, TextIO],
  base: str = "hex",
  relative_timestamp: bool = True,
  ) -> None:
@@ -248,7 +246,7 @@ def __iter__(self) -> Generator[Message, None, None]:
  self.stop()
 
 
-class ASCWriter(FileIOMessageWriter, Listener):
+class ASCWriter(FileIOMessageWriter):
  """Logs CAN data to an ASCII log file (.asc).
 
  The measurement starts with the timestamp of the first registered message.
@@ -287,7 +285,7 @@ class ASCWriter(FileIOMessageWriter, Listener):
 
  def __init__(
  self,
- file: AcceptedIOType,
+ file: Union[StringPathLike, TextIO],
  channel: int = 1,
  ) -> None:
  """

can/io/blf.py

@@ -17,13 +17,12 @@
 import datetime
 import time
 import logging
-from typing import List, BinaryIO
+from typing import List, BinaryIO, Generator, Union
 
 from ..message import Message
-from ..listener import Listener
 from ..util import len2dlc, dlc2len, channel2int
-from ..typechecking import AcceptedIOType
-from .generic import BaseIOHandler, FileIOMessageWriter
+from ..typechecking import StringPathLike
+from .generic import FileIOMessageWriter, MessageReader
 
 
 class BLFParseError(Exception):
@@ -131,7 +130,7 @@ def systemtime_to_timestamp(systemtime):
  return 0
 
 
-class BLFReader(BaseIOHandler):
+class BLFReader(MessageReader):
  """
  Iterator of CAN messages from a Binary Logging File.
 
@@ -141,7 +140,7 @@ class BLFReader(BaseIOHandler):
 
  file: BinaryIO
 
- def __init__(self, file: AcceptedIOType) -> None:
+ def __init__(self, file: Union[StringPathLike, BinaryIO]) -> None:
  """
  :param file: a path-like object or as file-like object to read from
  If this is a file-like object, is has to opened in binary
@@ -162,7 +161,7 @@ def __init__(self, file: AcceptedIOType) -> None:
  self._tail = b""
  self._pos = 0
 
- def __iter__(self):
+ def __iter__(self) -> Generator[Message, None, None]:
  while True:
  data = self.file.read(OBJ_HEADER_BASE_STRUCT.size)
  if not data:
@@ -349,7 +348,7 @@ def _parse_data(self, data):
  pos = next_pos
 
 
-class BLFWriter(FileIOMessageWriter, Listener):
+class BLFWriter(FileIOMessageWriter):
  """
  Logs CAN data to a Binary Logging File compatible with Vector's tools.
  """
@@ -364,7 +363,7 @@ class BLFWriter(FileIOMessageWriter, Listener):
 
  def __init__(
  self,
- file: AcceptedIOType,
+ file: Union[StringPathLike, BinaryIO],
  append: bool = False,
  channel: int = 1,
  compression_level: int = -1,

can/io/canutils.py

@@ -5,11 +5,11 @@
 """
 
 import logging
+from typing import Generator, TextIO, Union
 
 from can.message import Message
-from can.listener import Listener
-from .generic import BaseIOHandler, FileIOMessageWriter
-from ..typechecking import AcceptedIOType
+from .generic import FileIOMessageWriter, MessageReader
+from ..typechecking import AcceptedIOType, StringPathLike
 
 log = logging.getLogger("can.io.canutils")
 
@@ -22,7 +22,7 @@
 CANFD_ESI = 0x02
 
 
-class CanutilsLogReader(BaseIOHandler):
+class CanutilsLogReader(MessageReader):
  """
  Iterator over CAN messages from a .log Logging File (candump -L).
 
@@ -32,30 +32,37 @@ class CanutilsLogReader(BaseIOHandler):
  ``(0.0) vcan0 001#8d00100100820100``
  """
 
- def __init__(self, file: AcceptedIOType) -> None:
+ file: TextIO
+
+ def __init__(self, file: Union[StringPathLike, TextIO]) -> None:
  """
  :param file: a path-like object or as file-like object to read from
  If this is a file-like object, is has to opened in text
  read mode, not binary read mode.
  """
  super().__init__(file, mode="r")
 
- def __iter__(self):
+ def __iter__(self) -> Generator[Message, None, None]:
  for line in self.file:
 
  # skip empty lines
  temp = line.strip()
  if not temp:
  continue
 
- timestamp, channel, frame = temp.split()
- timestamp = float(timestamp[1:-1])
- canId, data = frame.split("#", maxsplit=1)
- if channel.isdigit():
- channel = int(channel)
+ channel_string: str
+ timestamp_string, channel_string, frame = temp.split()
+ timestamp = float(timestamp_string[1:-1])
+ can_id_string, data = frame.split("#", maxsplit=1)
+
+ channel: Union[int, str]
+ if channel_string.isdigit():
+ channel = int(channel_string)
+ else:
+ channel = channel_string
 
- isExtended = len(canId) > 3
- canId = int(canId, 16)
+ is_extended = len(can_id_string) > 3
+ can_id = int(can_id_string, 16)
 
  is_fd = False
  brs = False
@@ -69,43 +76,43 @@ def __iter__(self):
  data = data[2:]
 
  if data and data[0].lower() == "r":
- isRemoteFrame = True
+ is_remote_frame = True
 
  if len(data) > 1:
  dlc = int(data[1:])
  else:
  dlc = 0
 
- dataBin = None
+ data_bin = None
  else:
- isRemoteFrame = False
+ is_remote_frame = False
 
  dlc = len(data) // 2
- dataBin = bytearray()
+ data_bin = bytearray()
  for i in range(0, len(data), 2):
- dataBin.append(int(data[i : (i + 2)], 16))
+ data_bin.append(int(data[i : (i + 2)], 16))
 
- if canId & CAN_ERR_FLAG and canId & CAN_ERR_BUSERROR:
+ if can_id & CAN_ERR_FLAG and can_id & CAN_ERR_BUSERROR:
  msg = Message(timestamp=timestamp, is_error_frame=True)
  else:
  msg = Message(
  timestamp=timestamp,
- arbitration_id=canId & 0x1FFFFFFF,
- is_extended_id=isExtended,
- is_remote_frame=isRemoteFrame,
+ arbitration_id=can_id & 0x1FFFFFFF,
+ is_extended_id=is_extended,
+ is_remote_frame=is_remote_frame,
  is_fd=is_fd,
  bitrate_switch=brs,
  error_state_indicator=esi,
  dlc=dlc,
- data=dataBin,
+ data=data_bin,
  channel=channel,
  )
  yield msg
 
  self.stop()
 
 
-class CanutilsLogWriter(FileIOMessageWriter, Listener):
+class CanutilsLogWriter(FileIOMessageWriter):
  """Logs CAN data to an ASCII log file (.log).
  This class is is compatible with "candump -L".
 
@@ -115,7 +122,10 @@ class CanutilsLogWriter(FileIOMessageWriter, Listener):
  """
 
  def __init__(
- self, file: AcceptedIOType, channel: str = "vcan0", append: bool = False
+ self,
+ file: Union[StringPathLike, TextIO],
+ channel: str = "vcan0",
+ append: bool = False,
  ):
  """
  :param file: a path-like object or as file-like object to write to

can/io/csv.py

@@ -10,15 +10,14 @@
 """
 
 from base64 import b64encode, b64decode
-from typing import TextIO
+from typing import TextIO, Generator, Union
 
 from can.message import Message
-from can.listener import Listener
-from .generic import BaseIOHandler, FileIOMessageWriter
-from ..typechecking import AcceptedIOType
+from .generic import FileIOMessageWriter, MessageReader
+from ..typechecking import StringPathLike
 
 
-class CSVReader(BaseIOHandler):
+class CSVReader(MessageReader):
  """Iterator over CAN messages from a .csv file that was
  generated by :class:`~can.CSVWriter` or that uses the same
  format as described there. Assumes that there is a header
@@ -27,15 +26,17 @@ class CSVReader(BaseIOHandler):
  Any line separator is accepted.
  """
 
- def __init__(self, file: AcceptedIOType) -> None:
+ file: TextIO
+
+ def __init__(self, file: Union[StringPathLike, TextIO]) -> None:
  """
  :param file: a path-like object or as file-like object to read from
  If this is a file-like object, is has to opened in text
  read mode, not binary read mode.
  """
  super().__init__(file, mode="r")
 
- def __iter__(self):
+ def __iter__(self) -> Generator[Message, None, None]:
  # skip the header line
  try:
  next(self.file)
@@ -62,7 +63,7 @@ def __iter__(self):
  self.stop()
 
 
-class CSVWriter(FileIOMessageWriter, Listener):
+class CSVWriter(FileIOMessageWriter):
  """Writes a comma separated text file with a line for
  each message. Includes a header line.
 
@@ -85,7 +86,9 @@ class CSVWriter(FileIOMessageWriter, Listener):
 
  file: TextIO
 
- def __init__(self, file: AcceptedIOType, append: bool = False) -> None:
+ def __init__(
+ self, file: Union[StringPathLike, TextIO], append: bool = False
+ ) -> None:
  """
  :param file: a path-like object or a file-like object to write to.
  If this is a file-like object, is has to open in text

can/io/generic.py

@@ -5,9 +5,6 @@
  Optional,
  cast,
  Iterable,
- Union,
- TextIO,
- BinaryIO,
  Type,
  ContextManager,
 )
@@ -76,21 +73,23 @@ def stop(self) -> None:
 class MessageWriter(BaseIOHandler, can.Listener, metaclass=ABCMeta):
  """The base class for all writers."""
 
+ file: Optional[can.typechecking.FileLike]
+
 
 # pylint: disable=abstract-method,too-few-public-methods
 class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):
  """A specialized base class for all writers with file descriptors."""
 
- file: Union[TextIO, BinaryIO]
+ file: can.typechecking.FileLike
 
  def __init__(self, file: can.typechecking.AcceptedIOType, mode: str = "rt") -> None:
- # Not possible with the type signature, but be verbose for user friendliness
+ # 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)
 
 
 # pylint: disable=too-few-public-methods
-class MessageReader(BaseIOHandler, Iterable, metaclass=ABCMeta):
+class MessageReader(BaseIOHandler, Iterable[can.Message], metaclass=ABCMeta):
  """The base class for all readers."""