|
6 | 6 | - under `test/data/logfile.asc` |
7 | 7 | """ |
8 | 8 | import gzip |
9 | | -from typing import cast, Any, Generator, IO, List, Optional, Dict, Union |
| 9 | +from typing import cast, Any, Generator, IO, List, Optional, Dict, Union, TextIO |
10 | 10 |
|
11 | 11 | from datetime import datetime |
12 | 12 | import time |
@@ -34,6 +34,8 @@ class ASCReader(BaseIOHandler): |
34 | 34 | bus statistics, J1939 Transport Protocol messages) is ignored. |
35 | 35 | """ |
36 | 36 |
|
| 37 | + file: TextIO |
| 38 | + |
37 | 39 | FORMAT_START_OF_FILE_DATE = "%a %b %d %I:%M:%S.%f %p %Y" |
38 | 40 |
|
39 | 41 | def __init__( |
@@ -205,19 +207,18 @@ def _process_fd_can_frame(self, line: str, msg_kwargs: Dict[str, Any]) -> Messag |
205 | 207 | return Message(**msg_kwargs) |
206 | 208 |
|
207 | 209 | def __iter__(self) -> Generator[Message, None, None]: |
208 | | - # This is guaranteed to not be None since we raise ValueError in __init__ |
209 | | - self.file = cast(IO[Any], self.file) |
210 | 210 | self._extract_header() |
211 | 211 |
|
212 | 212 | for line in self.file: |
213 | 213 | temp = line.strip() |
214 | 214 | if not temp or not temp[0].isdigit(): |
215 | 215 | # Could be a comment |
216 | 216 | continue |
217 | | - msg_kwargs = {} |
| 217 | + |
| 218 | + msg_kwargs: Dict[str, Union[float, bool, int]] = {} |
218 | 219 | try: |
219 | | - timestamp, channel, rest_of_message = temp.split(None, 2) |
220 | | - timestamp = float(timestamp) + self.start_time |
| 220 | + _timestamp, channel, rest_of_message = temp.split(None, 2) |
| 221 | + timestamp = float(_timestamp) + self.start_time |
221 | 222 | msg_kwargs["timestamp"] = timestamp |
222 | 223 | if channel == "CANFD": |
223 | 224 | msg_kwargs["is_fd"] = True |
@@ -250,6 +251,8 @@ class ASCWriter(FileIOMessageWriter, Listener): |
250 | 251 | It the first message does not have a timestamp, it is set to zero. |
251 | 252 | """ |
252 | 253 |
|
| 254 | + file: TextIO |
| 255 | + |
253 | 256 | FORMAT_MESSAGE = "{channel} {id:<15} {dir:<4} {dtype} {data}" |
254 | 257 | FORMAT_MESSAGE_FD = " ".join( |
255 | 258 | [ |
@@ -319,8 +322,6 @@ def log_event(self, message: str, timestamp: Optional[float] = None) -> None: |
319 | 322 | if not message: # if empty or None |
320 | 323 | logger.debug("ASCWriter: ignoring empty message") |
321 | 324 | return |
322 | | - # This is guaranteed to not be None since we raise ValueError in __init__ |
323 | | - self.file = cast(IO[Any], self.file) |
324 | 325 |
|
325 | 326 | # this is the case for the very first message: |
326 | 327 | if not self.header_written: |
|
0 commit comments