Skip to content

Commit 5cada6d

Browse files
authored
Add typing annotations for ASC module (#818)
This works towards PEP 561 compatibility.
1 parent de0613e commit 5cada6d

File tree

1 file changed

+37
-13
lines changed

1 file changed

+37
-13
lines changed

can/io/asc.py

Lines changed: 37 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66
- under `test/data/logfile.asc`
77
"""
88

9+
from typing import cast, Any, Generator, IO, List, Optional, Tuple, Union
10+
from can import typechecking
11+
912
from datetime import datetime
1013
import time
1114
import logging
@@ -32,7 +35,11 @@ class ASCReader(BaseIOHandler):
3235
TODO: turn relative timestamps back to absolute form
3336
"""
3437

35-
def __init__(self, file, base="hex"):
38+
def __init__(
39+
self,
40+
file: Union[typechecking.FileLike, typechecking.StringPathLike],
41+
base: str = "hex",
42+
) -> None:
3643
"""
3744
:param file: a path-like object or as file-like object to read from
3845
If this is a file-like object, is has to opened in text
@@ -42,10 +49,13 @@ def __init__(self, file, base="hex"):
4249
this value will be overwritten. Default "hex".
4350
"""
4451
super().__init__(file, mode="r")
52+
53+
if not self.file:
54+
raise ValueError("The given file cannot be None")
4555
self.base = base
4656

4757
@staticmethod
48-
def _extract_can_id(str_can_id, base):
58+
def _extract_can_id(str_can_id: str, base: int) -> Tuple[int, bool]:
4959
if str_can_id[-1:].lower() == "x":
5060
is_extended = True
5161
can_id = int(str_can_id[0:-1], base)
@@ -55,13 +65,15 @@ def _extract_can_id(str_can_id, base):
5565
return can_id, is_extended
5666

5767
@staticmethod
58-
def _check_base(base):
68+
def _check_base(base: str) -> int:
5969
if base not in ["hex", "dec"]:
6070
raise ValueError('base should be either "hex" or "dec"')
6171
return BASE_DEC if base == "dec" else BASE_HEX
6272

63-
def __iter__(self):
73+
def __iter__(self) -> Generator[Message, None, None]:
6474
base = self._check_base(self.base)
75+
# This is guaranteed to not be None since we raise ValueError in __init__
76+
self.file = cast(IO[Any], self.file)
6577
for line in self.file:
6678
# logger.debug("ASCReader: parsing line: '%s'", line.splitlines()[0])
6779
if line.split(" ")[0] == "base":
@@ -194,7 +206,11 @@ class ASCWriter(BaseIOHandler, Listener):
194206
FORMAT_DATE = "%a %b %d %I:%M:%S.{} %p %Y"
195207
FORMAT_EVENT = "{timestamp: 9.6f} {message}\n"
196208

197-
def __init__(self, file, channel=1):
209+
def __init__(
210+
self,
211+
file: Union[typechecking.FileLike, typechecking.StringPathLike],
212+
channel: int = 1,
213+
) -> None:
198214
"""
199215
:param file: a path-like object or as file-like object to write to
200216
If this is a file-like object, is has to opened in text
@@ -203,6 +219,9 @@ def __init__(self, file, channel=1):
203219
have a channel set
204220
"""
205221
super().__init__(file, mode="w")
222+
if not self.file:
223+
raise ValueError("The given file cannot be None")
224+
206225
self.channel = channel
207226

208227
# write start of file header
@@ -213,24 +232,29 @@ def __init__(self, file, channel=1):
213232

214233
# the last part is written with the timestamp of the first message
215234
self.header_written = False
216-
self.last_timestamp = None
217-
self.started = None
235+
self.last_timestamp = 0.0
236+
self.started = 0.0
218237

219-
def stop(self):
238+
def stop(self) -> None:
239+
# This is guaranteed to not be None since we raise ValueError in __init__
240+
self.file = cast(IO[Any], self.file)
220241
if not self.file.closed:
221242
self.file.write("End TriggerBlock\n")
222243
super().stop()
223244

224-
def log_event(self, message, timestamp=None):
245+
def log_event(self, message: str, timestamp: Optional[float] = None) -> None:
225246
"""Add a message to the log file.
226247
227-
:param str message: an arbitrary message
228-
:param float timestamp: the absolute timestamp of the event
248+
:param message: an arbitrary message
249+
:param timestamp: the absolute timestamp of the event
229250
"""
230251

231252
if not message: # if empty or None
232253
logger.debug("ASCWriter: ignoring empty message")
233254
return
255+
# This is guaranteed to not be None since we raise ValueError in __init__
256+
self.file = cast(IO[Any], self.file)
257+
234258
# this is the case for the very first message:
235259
if not self.header_written:
236260
self.last_timestamp = timestamp or 0.0
@@ -251,14 +275,14 @@ def log_event(self, message, timestamp=None):
251275
line = self.FORMAT_EVENT.format(timestamp=timestamp, message=message)
252276
self.file.write(line)
253277

254-
def on_message_received(self, msg):
278+
def on_message_received(self, msg: Message) -> None:
255279

256280
if msg.is_error_frame:
257281
self.log_event("{} ErrorFrame".format(self.channel), msg.timestamp)
258282
return
259283
if msg.is_remote_frame:
260284
dtype = "r"
261-
data = []
285+
data: List[str] = []
262286
else:
263287
dtype = "d {}".format(msg.dlc)
264288
data = ["{:02X}".format(byte) for byte in msg.data]

0 commit comments

Comments
 (0)