From 71ead7a5758bd54218471634fda2b23688dda55b Mon Sep 17 00:00:00 2001 From: Andreas Lauser Date: Tue, 29 Jul 2025 15:35:23 +0200 Subject: [PATCH 1/3] remove the freezegun dependency from the logreader unit test this makes the logreader unit test slightly less comprehensive because it now does not check if the correct local time zone is used if no time zone is passed to the parser, but since this bit of functionality is fairly trivial, IMO we can simply trust that it works correctly. Signed-off-by: Andreas Lauser Approved-by: Gerrit Ecke --- tests/test_logreader.py | 100 ++++++++++++++++++++-------------------- tox.ini | 1 - 2 files changed, 50 insertions(+), 51 deletions(-) diff --git a/tests/test_logreader.py b/tests/test_logreader.py index 779cba68..c08cbe2b 100644 --- a/tests/test_logreader.py +++ b/tests/test_logreader.py @@ -3,7 +3,6 @@ import unittest import pytest -from freezegun import freeze_time import cantools @@ -12,15 +11,15 @@ def utc_plus(offset: int) -> datetime.timezone: return datetime.timezone(datetime.timedelta(hours=offset)) -@freeze_time(tz_offset=0) class TestLogreaderFormats(unittest.TestCase): def test_empty_line(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) + outp = parser.parse("") self.assertIsNone(outp) def test_candump(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -74,7 +73,7 @@ def test_candump(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.MISSING) def test_timestamped_candump(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("(000.000000) vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -115,7 +114,7 @@ def test_timestamped_candump(self): self.assertEqual(outp.timestamp.microseconds, 679614) self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.RELATIVE) - outp = parser.parse("(1613749650.388103) can1 0AD [08] A6 55 3B CF 3F 1A F5 2A") + outp = cantools.logreader.Parser(tz=utc_plus(0)).parse("(1613749650.388103) can1 0AD [08] A6 55 3B CF 3F 1A F5 2A") self.assertEqual(outp.channel, 'can1') self.assertEqual(outp.frame_id, 0xad) self.assertEqual(outp.is_extended_frame, False) @@ -151,7 +150,7 @@ def test_timestamped_candump(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.RELATIVE) def test_candump_log(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("(1579857014.345944) can2 486#82967A6B006B07F8") self.assertEqual(outp.channel, 'can2') @@ -185,37 +184,36 @@ def test_candump_log(self): self.assertEqual(outp.timestamp.second, 24) self.assertEqual(outp.timestamp.microsecond, 501098) - with freeze_time(tz_offset=2): - outp = parser.parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") - self.assertEqual(outp.channel, 'vcan0') - self.assertEqual(outp.frame_id, 0x123) - self.assertEqual(outp.is_extended_frame, True) - self.assertEqual(outp.data, b'\x12\x34\x56\x78\x90\xab\xcd\xef') - self.assertEqual(outp.is_remote_frame, False) - self.assertEqual(outp.timestamp.year, 2025) - self.assertEqual(outp.timestamp.month, 7) - self.assertEqual(outp.timestamp.day, 19) - self.assertEqual(outp.timestamp.hour, 11) - self.assertEqual(outp.timestamp.minute, 48) - self.assertEqual(outp.timestamp.second, 59) - self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) - - outp = cantools.logreader.Parser(tz=utc_plus(0)).parse( - "(1613656104.501098) can3 14C##155B53476F7B82EEEB8E97236AC252B8BBB5B80A6A7734B2F675C6D2CEEC869D3") - self.assertEqual(outp.channel, 'can3') - self.assertEqual(outp.frame_id, 0x14c) - self.assertEqual(outp.is_extended_frame, False) - self.assertEqual( - outp.data, b'\x55\xB5\x34\x76\xF7\xB8\x2E\xEE\xB8\xE9\x72\x36\xAC\x25\x2B\x8B\xBB\x5B\x80\xA6\xA7\x73\x4B\x2F\x67\x5C\x6D\x2C\xEE\xC8\x69\xD3') - self.assertEqual(outp.is_remote_frame, False) - self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) - self.assertEqual(outp.timestamp.year, 2021) - self.assertEqual(outp.timestamp.month, 2) - self.assertEqual(outp.timestamp.day, 18) - self.assertEqual(outp.timestamp.hour, 13) - self.assertEqual(outp.timestamp.minute, 48) - self.assertEqual(outp.timestamp.second, 24) - self.assertEqual(outp.timestamp.microsecond, 501098) + outp = cantools.logreader.Parser(tz=utc_plus(2)).parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") + self.assertEqual(outp.channel, 'vcan0') + self.assertEqual(outp.frame_id, 0x123) + self.assertEqual(outp.is_extended_frame, True) + self.assertEqual(outp.data, b'\x12\x34\x56\x78\x90\xab\xcd\xef') + self.assertEqual(outp.is_remote_frame, False) + self.assertEqual(outp.timestamp.year, 2025) + self.assertEqual(outp.timestamp.month, 7) + self.assertEqual(outp.timestamp.day, 19) + self.assertEqual(outp.timestamp.hour, 11) + self.assertEqual(outp.timestamp.minute, 48) + self.assertEqual(outp.timestamp.second, 59) + self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) + + outp = cantools.logreader.Parser(tz=utc_plus(0)).parse( + "(1613656104.501098) can3 14C##155B53476F7B82EEEB8E97236AC252B8BBB5B80A6A7734B2F675C6D2CEEC869D3") + self.assertEqual(outp.channel, 'can3') + self.assertEqual(outp.frame_id, 0x14c) + self.assertEqual(outp.is_extended_frame, False) + self.assertEqual( + outp.data, b'\x55\xB5\x34\x76\xF7\xB8\x2E\xEE\xB8\xE9\x72\x36\xAC\x25\x2B\x8B\xBB\x5B\x80\xA6\xA7\x73\x4B\x2F\x67\x5C\x6D\x2C\xEE\xC8\x69\xD3') + self.assertEqual(outp.is_remote_frame, False) + self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) + self.assertEqual(outp.timestamp.year, 2021) + self.assertEqual(outp.timestamp.month, 2) + self.assertEqual(outp.timestamp.day, 18) + self.assertEqual(outp.timestamp.hour, 13) + self.assertEqual(outp.timestamp.minute, 48) + self.assertEqual(outp.timestamp.second, 24) + self.assertEqual(outp.timestamp.microsecond, 501098) outp = cantools.logreader.Parser(tz=utc_plus(2)).parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") self.assertEqual(outp.channel, 'vcan0') @@ -246,7 +244,7 @@ def test_candump_log(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_absolute_timestamp(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("(2020-12-19 12:04:45.485261) vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -292,7 +290,7 @@ def test_candump_log_absolute_timestamp(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_ascii(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse(" can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -303,7 +301,7 @@ def test_candump_log_ascii(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.MISSING) def test_candump_log_ascii_timestamped(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse(" (1621271100.919019) can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -314,7 +312,7 @@ def test_candump_log_ascii_timestamped(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_ascii_absolute(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("(2020-12-19 12:04:45.485261) can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -332,7 +330,7 @@ def test_candump_log_ascii_absolute(self): self.assertEqual(outp.timestamp.microsecond, 485261) def test_pcan_traceV10(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("1) 1841 0001 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -355,7 +353,7 @@ def test_pcan_traceV10(self): self.assertEqual(outp.timestamp.microseconds, 844000) def test_pcan_traceV11(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("1) 6357.2 Rx 0401 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -378,7 +376,7 @@ def test_pcan_traceV11(self): self.assertEqual(outp.timestamp.microseconds, 352700) def test_pcan_traceV12(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("1) 6357.213 1 Rx 0401 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -401,7 +399,7 @@ def test_pcan_traceV12(self): self.assertEqual(outp.timestamp.microseconds, 352743) def test_pcan_traceV13(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse("1) 6357.213 1 Rx 0401 - 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -424,7 +422,7 @@ def test_pcan_traceV13(self): self.assertEqual(outp.timestamp.microseconds, 352743) def test_pcan_traceV20(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse(" 1 1059.900 DT 0300 Rx 7 00 00 00 00 04 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -447,7 +445,7 @@ def test_pcan_traceV20(self): self.assertEqual(outp.timestamp.microseconds, 336543) def test_pcan_traceV21(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) outp = parser.parse(" 1 1059.900 DT 1 0300 Rx - 7 00 00 00 00 04 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -854,13 +852,15 @@ def test_pcan_traceV21(self): next(frame_iter) def test_none(self): - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) + frame_iter = iter(parser) with pytest.raises(StopIteration): next(frame_iter) def test_data_frame_repr() -> None: - parser = cantools.logreader.Parser() + parser = cantools.logreader.Parser(tz=utc_plus(0)) + outp = parser.parse("vcan0 0C8 [8] F0 00 00 00 00 00 00 00") assert repr(outp) == r"DataFrame(channel = 'vcan0', frame_id = 200, is_extended_frame = False, data = b'\xf0\x00\x00\x00\x00\x00\x00\x00', is_remote_frame = False, timestamp = None, timestamp_format = )" diff --git a/tox.ini b/tox.ini index 2aff0381..fbf0efdb 100644 --- a/tox.ini +++ b/tox.ini @@ -9,7 +9,6 @@ deps = coverage==7.3.* parameterized==0.9.* windows-curses; platform_system=="Windows" and platform_python_implementation=="CPython" and python_version<"3.13" - freezegun>=1.5.2,<1.6.0 extras = plot From c784c133e9e02a9bf727f99bbb4f91f465fa881a Mon Sep 17 00:00:00 2001 From: Andreas Lauser Date: Wed, 30 Jul 2025 11:46:44 +0200 Subject: [PATCH 2/3] Logreader: Convert to old-school type hints this makes type hints functional when using python 3.9. (python 3.9 support currently is on limited time because 3.9 it is close to its upstream EOL, but converting old-school to new-school type hints can be done automatically by ruff once the minimum python version is raised in `pyproject.toml`...) Signed-off-by: Andreas Lauser --- src/cantools/logreader.py | 68 +++++++++++++++++++-------------------- 1 file changed, 34 insertions(+), 34 deletions(-) diff --git a/src/cantools/logreader.py b/src/cantools/logreader.py index fd373417..464343bb 100644 --- a/src/cantools/logreader.py +++ b/src/cantools/logreader.py @@ -3,14 +3,14 @@ import datetime import enum import re -import typing +from typing import Literal, Optional, Union, TYPE_CHECKING, overload -if typing.TYPE_CHECKING: +if TYPE_CHECKING: import io from collections.abc import Iterator -TimestampType = typing.Union[datetime.datetime, datetime.timedelta, None] +TimestampType = Optional[Union[datetime.datetime, datetime.timedelta]] class TimestampFormat(enum.Enum): """Describes a type of timestamp. ABSOLUTE is referring to UNIX time @@ -30,7 +30,7 @@ def __init__(self, channel: str, is_extended_frame: bool, data: bytes, is_remote_frame: bool, - timestamp: 'TimestampType', + timestamp: TimestampType, timestamp_format: TimestampFormat): """Constructor for DataFrame @@ -55,9 +55,9 @@ def __repr__(self) -> str: class BasePattern: - pattern: 're.Pattern[str]' + pattern: re.Pattern[str] - def match(self, line: str) -> 'DataFrame|None': + def match(self, line: str) -> Optional[DataFrame]: mo = self.pattern.match(line) if mo: return self.unpack(mo) @@ -65,13 +65,13 @@ def match(self, line: str) -> 'DataFrame|None': return None @abc.abstractmethod - def unpack(self, match_object: 're.Match[str]') -> 'DataFrame|None': + def unpack(self, match_object: re.Match[str]) -> Optional[DataFrame]: raise NotImplementedError() class CandumpBasePattern(BasePattern): - def unpack(self, match_object: 're.Match[str]') -> DataFrame: + def unpack(self, match_object: re.Match[str]) -> DataFrame: channel = match_object.group('channel') frame_id = int(match_object.group('can_id'), 16) is_extended_frame = len(match_object.group('can_id')) > 3 @@ -88,7 +88,7 @@ def unpack(self, match_object: 're.Match[str]') -> DataFrame: return DataFrame(channel=channel, frame_id=frame_id, is_extended_frame=is_extended_frame, data=data, is_remote_frame=is_remote_frame, timestamp=timestamp, timestamp_format=timestamp_format) @abc.abstractmethod - def parse_timestamp(self, match_object: 're.Match[str]') -> 'tuple[TimestampType, TimestampFormat]': + def parse_timestamp(self, match_object: re.Match[str]) -> tuple[TimestampType, TimestampFormat]: raise NotImplementedError() @@ -101,7 +101,7 @@ class CandumpDefaultPattern(CandumpBasePattern): pattern = re.compile( r'^\s*?(?P[a-zA-Z0-9]+)\s+(?P[0-9A-F]+)\s+\[\d+\]\s*(?Premote request|[0-9A-F ]*).*?$') - def parse_timestamp(self, match_object: 're.Match[str]') -> 'tuple[TimestampType, TimestampFormat]': + def parse_timestamp(self, match_object: re.Match[str]) -> tuple[TimestampType, TimestampFormat]: timestamp = None timestamp_format = TimestampFormat.MISSING return timestamp, timestamp_format @@ -116,12 +116,12 @@ class CandumpTimestampedPattern(CandumpBasePattern): pattern = re.compile( r'^\s*?\((?P[\d.]+)\)\s+(?P[a-zA-Z0-9]+)\s+(?P[0-9A-F]+)\s+\[\d+\]\s*(?Premote request|[0-9A-F ]*).*?$') - def __init__(self, tz: 'datetime.tzinfo|None') -> None: + def __init__(self, tz: Optional[datetime.tzinfo]) -> None: self.tz = tz - def parse_timestamp(self, match_object: 're.Match[str]') -> 'tuple[TimestampType, TimestampFormat]': + def parse_timestamp(self, match_object: re.Match[str]) -> tuple[TimestampType, TimestampFormat]: seconds = float(match_object.group('timestamp')) - timestamp: typing.Union[datetime.timedelta, datetime.datetime] + timestamp: Union[datetime.timedelta, datetime.datetime] if seconds < 662688000: # 1991-01-01 00:00:00, "Released in 1991, the Mercedes-Benz W140 was the first production vehicle to feature a CAN-based multiplex wiring system." timestamp = datetime.timedelta(seconds=seconds) timestamp_format = TimestampFormat.RELATIVE @@ -138,10 +138,10 @@ class CandumpDefaultLogPattern(CandumpBasePattern): pattern = re.compile( r'^\s*?\((?P[\d.]+?)\)\s+?(?P[a-zA-Z0-9]+)\s+?(?P[0-9A-F]+?)#(#[0-9A-F])?(?PR|([0-9A-Fa-f]{2})*)(\s+[RT])?$') - def __init__(self, tz: 'datetime.tzinfo|None') -> None: + def __init__(self, tz: Optional[datetime.tzinfo]) -> None: self.tz = tz - def parse_timestamp(self, match_object: 're.Match[str]') -> 'tuple[TimestampType, TimestampFormat]': + def parse_timestamp(self, match_object: re.Match[str]) -> tuple[TimestampType, TimestampFormat]: timestamp = datetime.datetime.fromtimestamp(float(match_object.group('timestamp')), self.tz) timestamp_format = TimestampFormat.ABSOLUTE return timestamp, timestamp_format @@ -156,7 +156,7 @@ class CandumpAbsoluteLogPattern(CandumpBasePattern): pattern = re.compile( r'^\s*?\((?P\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\)\s+(?P[a-zA-Z0-9]+)\s+(?P[0-9A-F]+)\s+\[\d+\]\s*(?Premote request|[0-9A-F ]*).*?$') - def parse_timestamp(self, match_object: 're.Match[str]') -> 'tuple[TimestampType, TimestampFormat]': + def parse_timestamp(self, match_object: re.Match[str]) -> tuple[TimestampType, TimestampFormat]: timestamp = datetime.datetime.strptime(match_object.group('timestamp'), "%Y-%m-%d %H:%M:%S.%f") timestamp_format = TimestampFormat.ABSOLUTE return timestamp, timestamp_format @@ -173,7 +173,7 @@ class PCANTracePatternV10(BasePattern): pattern = re.compile( r'^\s*?\d+\)\s*?(?P\d+)\s+(?P[0-9A-F]+)\s+(?P[0-9])\s+(?PRTR|[0-9A-F ]*)$') - def unpack(self, match_object: 're.Match[str]') -> 'DataFrame|None': + def unpack(self, match_object: re.Match[str]) -> Optional[DataFrame]: channel = self.parse_channel(match_object) frame_id = int(match_object.group('can_id'), 16) is_extended_frame = len(match_object.group('can_id')) > 4 @@ -185,10 +185,10 @@ def unpack(self, match_object: 're.Match[str]') -> 'DataFrame|None': return DataFrame(channel=channel, frame_id=frame_id, is_extended_frame=is_extended_frame, data=data, is_remote_frame=is_remote_frame, timestamp=timestamp, timestamp_format=timestamp_format) - def parse_channel(self, match_object: 're.Match[str]') -> str: + def parse_channel(self, match_object: re.Match[str]) -> str: return 'pcanx' - def parse_data(self, match_object: 're.Match[str]') -> 'tuple[bytes, bool]': + def parse_data(self, match_object: re.Match[str]) -> tuple[bytes, bool]: data = match_object.group('can_data') if data == 'RTR': is_remote_frame = True @@ -211,7 +211,7 @@ class PCANTracePatternV11(PCANTracePatternV10): pattern = re.compile( r'^\s*?\d+\)\s*?(?P\d+.\d+)\s+(?P\w+)\s+(?P[0-9A-F]+)\s+(?P[0-9])\s+(?PRTR|[0-9A-F ]*)$') - def unpack(self, match_object: 're.Match[str]') -> 'DataFrame|None': + def unpack(self, match_object: re.Match[str]) -> Optional[DataFrame]: if match_object.group('type') in ('Error', 'Warng'): # yes, they really spell Warning without the 'in' return None @@ -228,7 +228,7 @@ class PCANTracePatternV12(PCANTracePatternV11): pattern = re.compile( r'^\s*?\d+\)\s*?(?P\d+.\d+)\s+(?P[0-9])\s+(?P\w+)\s+(?P[0-9A-F]+)\s+(?P[0-9])\s+(?PRTR|[0-9A-F ]*)$') - def parse_channel(self, match_object: 're.Match[str]') -> str: + def parse_channel(self, match_object: re.Match[str]) -> str: return 'pcan' + match_object.group('channel') @@ -254,10 +254,10 @@ class PCANTracePatternV20(PCANTracePatternV13): pattern = re.compile( r'^\s*?\d+?\s*?(?P\d+.\d+)\s+(?P\w+)\s+(?P[0-9A-F]+)\s+(?P\w+)\s+(?P[0-9]+)(\s+(?P[0-9A-F ]*))?$') - def parse_channel(self, match_object: 're.Match[str]') -> str: + def parse_channel(self, match_object: re.Match[str]) -> str: return 'pcanx' - def parse_data(self, match_object: 're.Match[str]') -> 'tuple[bytes, bool]': + def parse_data(self, match_object: re.Match[str]) -> tuple[bytes, bool]: if match_object.group('type') == 'RR': is_remote_frame = True data = bytes(0) @@ -277,7 +277,7 @@ class PCANTracePatternV21(PCANTracePatternV20): pattern = re.compile( r'^\s*?\d+?\s*?(?P\d+.\d+)\s+(?P\w+)\s+(?P[0-9])\s+(?P[0-9A-F]+)\s+(?P.+)\s+-\s+(?P[0-9]+)(\s+(?P[0-9A-F ]*))?$') - def parse_channel(self, match_object: 're.Match[str]') -> str: + def parse_channel(self, match_object: re.Match[str]) -> str: return 'pcan' + match_object.group('channel') @@ -292,12 +292,12 @@ class Parser: print(f'{frame.timestamp}: {frame.frame_id}') """ - def __init__(self, stream: 'io.TextIOBase|None' = None, *, tz: 'datetime.tzinfo|None' = None) -> None: + def __init__(self, stream: Optional[io.TextIOBase] = None, *, tz: Optional[datetime.tzinfo] = None) -> None: self.stream = stream - self.pattern: typing.Optional[BasePattern] = None + self.pattern: Optional[BasePattern] = None self.tz = tz - def detect_pattern(self, line: str) -> 'BasePattern|None': + def detect_pattern(self, line: str) -> Optional[BasePattern]: for p in [CandumpDefaultPattern(), CandumpTimestampedPattern(self.tz), CandumpDefaultLogPattern(self.tz), CandumpAbsoluteLogPattern(), PCANTracePatternV21(), PCANTracePatternV20(), PCANTracePatternV13(), PCANTracePatternV12(), PCANTracePatternV11(), PCANTracePatternV10()]: mo = p.pattern.match(line) if mo: @@ -305,22 +305,22 @@ def detect_pattern(self, line: str) -> 'BasePattern|None': return None - def parse(self, line: str) -> 'DataFrame|None': + def parse(self, line: str) -> Optional[DataFrame]: if self.pattern is None: self.pattern = self.detect_pattern(line) if self.pattern is None: return None return self.pattern.match(line) - @typing.overload - def iterlines(self) -> 'Iterator[tuple[str, DataFrame]]': + @overload + def iterlines(self) -> Iterator[tuple[str, DataFrame]]: pass - @typing.overload - def iterlines(self, keep_unknowns: 'typing.Literal[True]') -> 'Iterator[tuple[str, DataFrame|None]]': + @overload + def iterlines(self, keep_unknowns: Literal[True]) -> Iterator[tuple[str, Optional[DataFrame]]]: pass - def iterlines(self, keep_unknowns: bool = False) -> 'Iterator[tuple[str, DataFrame|None]]': + def iterlines(self, keep_unknowns: bool = False) -> Iterator[tuple[str, Optional[DataFrame]]]: """Returns an generator that yields (str, DataFrame) tuples with the raw log entry and a parsed log entry. If keep_unknowns=True, (str, None) tuples will be returned for log entries that couldn't be decoded. @@ -341,7 +341,7 @@ def iterlines(self, keep_unknowns: bool = False) -> 'Iterator[tuple[str, DataFra else: continue - def __iter__(self) -> 'Iterator[DataFrame]': + def __iter__(self) -> Iterator[DataFrame]: """Returns DataFrame log entries. Non-parseable log entries is discarded.""" for _, frame in self.iterlines(): From 8d744e4f9486a0c142b1142e8cc524794769681f Mon Sep 17 00:00:00 2001 From: Andreas Lauser Date: Fri, 8 Aug 2025 21:00:34 +0200 Subject: [PATCH 3/3] test_logreader: use the freezegun module again this makes the test slightly more comprehensive again. IMO this is acceptable because I missed the fact that the freezgun is only a dependency for running the unit tests, not for normal operation. Signed-off-by: Andreas Lauser --- tests/test_logreader.py | 100 ++++++++++++++++++++-------------------- tox.ini | 1 + 2 files changed, 51 insertions(+), 50 deletions(-) diff --git a/tests/test_logreader.py b/tests/test_logreader.py index c08cbe2b..779cba68 100644 --- a/tests/test_logreader.py +++ b/tests/test_logreader.py @@ -3,6 +3,7 @@ import unittest import pytest +from freezegun import freeze_time import cantools @@ -11,15 +12,15 @@ def utc_plus(offset: int) -> datetime.timezone: return datetime.timezone(datetime.timedelta(hours=offset)) +@freeze_time(tz_offset=0) class TestLogreaderFormats(unittest.TestCase): def test_empty_line(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) - + parser = cantools.logreader.Parser() outp = parser.parse("") self.assertIsNone(outp) def test_candump(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -73,7 +74,7 @@ def test_candump(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.MISSING) def test_timestamped_candump(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("(000.000000) vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -114,7 +115,7 @@ def test_timestamped_candump(self): self.assertEqual(outp.timestamp.microseconds, 679614) self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.RELATIVE) - outp = cantools.logreader.Parser(tz=utc_plus(0)).parse("(1613749650.388103) can1 0AD [08] A6 55 3B CF 3F 1A F5 2A") + outp = parser.parse("(1613749650.388103) can1 0AD [08] A6 55 3B CF 3F 1A F5 2A") self.assertEqual(outp.channel, 'can1') self.assertEqual(outp.frame_id, 0xad) self.assertEqual(outp.is_extended_frame, False) @@ -150,7 +151,7 @@ def test_timestamped_candump(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.RELATIVE) def test_candump_log(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("(1579857014.345944) can2 486#82967A6B006B07F8") self.assertEqual(outp.channel, 'can2') @@ -184,36 +185,37 @@ def test_candump_log(self): self.assertEqual(outp.timestamp.second, 24) self.assertEqual(outp.timestamp.microsecond, 501098) - outp = cantools.logreader.Parser(tz=utc_plus(2)).parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") - self.assertEqual(outp.channel, 'vcan0') - self.assertEqual(outp.frame_id, 0x123) - self.assertEqual(outp.is_extended_frame, True) - self.assertEqual(outp.data, b'\x12\x34\x56\x78\x90\xab\xcd\xef') - self.assertEqual(outp.is_remote_frame, False) - self.assertEqual(outp.timestamp.year, 2025) - self.assertEqual(outp.timestamp.month, 7) - self.assertEqual(outp.timestamp.day, 19) - self.assertEqual(outp.timestamp.hour, 11) - self.assertEqual(outp.timestamp.minute, 48) - self.assertEqual(outp.timestamp.second, 59) - self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) - - outp = cantools.logreader.Parser(tz=utc_plus(0)).parse( - "(1613656104.501098) can3 14C##155B53476F7B82EEEB8E97236AC252B8BBB5B80A6A7734B2F675C6D2CEEC869D3") - self.assertEqual(outp.channel, 'can3') - self.assertEqual(outp.frame_id, 0x14c) - self.assertEqual(outp.is_extended_frame, False) - self.assertEqual( - outp.data, b'\x55\xB5\x34\x76\xF7\xB8\x2E\xEE\xB8\xE9\x72\x36\xAC\x25\x2B\x8B\xBB\x5B\x80\xA6\xA7\x73\x4B\x2F\x67\x5C\x6D\x2C\xEE\xC8\x69\xD3') - self.assertEqual(outp.is_remote_frame, False) - self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) - self.assertEqual(outp.timestamp.year, 2021) - self.assertEqual(outp.timestamp.month, 2) - self.assertEqual(outp.timestamp.day, 18) - self.assertEqual(outp.timestamp.hour, 13) - self.assertEqual(outp.timestamp.minute, 48) - self.assertEqual(outp.timestamp.second, 24) - self.assertEqual(outp.timestamp.microsecond, 501098) + with freeze_time(tz_offset=2): + outp = parser.parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") + self.assertEqual(outp.channel, 'vcan0') + self.assertEqual(outp.frame_id, 0x123) + self.assertEqual(outp.is_extended_frame, True) + self.assertEqual(outp.data, b'\x12\x34\x56\x78\x90\xab\xcd\xef') + self.assertEqual(outp.is_remote_frame, False) + self.assertEqual(outp.timestamp.year, 2025) + self.assertEqual(outp.timestamp.month, 7) + self.assertEqual(outp.timestamp.day, 19) + self.assertEqual(outp.timestamp.hour, 11) + self.assertEqual(outp.timestamp.minute, 48) + self.assertEqual(outp.timestamp.second, 59) + self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) + + outp = cantools.logreader.Parser(tz=utc_plus(0)).parse( + "(1613656104.501098) can3 14C##155B53476F7B82EEEB8E97236AC252B8BBB5B80A6A7734B2F675C6D2CEEC869D3") + self.assertEqual(outp.channel, 'can3') + self.assertEqual(outp.frame_id, 0x14c) + self.assertEqual(outp.is_extended_frame, False) + self.assertEqual( + outp.data, b'\x55\xB5\x34\x76\xF7\xB8\x2E\xEE\xB8\xE9\x72\x36\xAC\x25\x2B\x8B\xBB\x5B\x80\xA6\xA7\x73\x4B\x2F\x67\x5C\x6D\x2C\xEE\xC8\x69\xD3') + self.assertEqual(outp.is_remote_frame, False) + self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) + self.assertEqual(outp.timestamp.year, 2021) + self.assertEqual(outp.timestamp.month, 2) + self.assertEqual(outp.timestamp.day, 18) + self.assertEqual(outp.timestamp.hour, 13) + self.assertEqual(outp.timestamp.minute, 48) + self.assertEqual(outp.timestamp.second, 24) + self.assertEqual(outp.timestamp.microsecond, 501098) outp = cantools.logreader.Parser(tz=utc_plus(2)).parse("(1752918539.271062) vcan0 00000123#1234567890ABCDEF") self.assertEqual(outp.channel, 'vcan0') @@ -244,7 +246,7 @@ def test_candump_log(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_absolute_timestamp(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("(2020-12-19 12:04:45.485261) vcan0 0C8 [8] F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'vcan0') @@ -290,7 +292,7 @@ def test_candump_log_absolute_timestamp(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_ascii(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse(" can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -301,7 +303,7 @@ def test_candump_log_ascii(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.MISSING) def test_candump_log_ascii_timestamped(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse(" (1621271100.919019) can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -312,7 +314,7 @@ def test_candump_log_ascii_timestamped(self): self.assertEqual(outp.timestamp_format, cantools.logreader.TimestampFormat.ABSOLUTE) def test_candump_log_ascii_absolute(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("(2020-12-19 12:04:45.485261) can1 123 [8] 31 30 30 2E 35 20 46 4D '100.5 FM'") self.assertEqual(outp.channel, 'can1') @@ -330,7 +332,7 @@ def test_candump_log_ascii_absolute(self): self.assertEqual(outp.timestamp.microsecond, 485261) def test_pcan_traceV10(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("1) 1841 0001 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -353,7 +355,7 @@ def test_pcan_traceV10(self): self.assertEqual(outp.timestamp.microseconds, 844000) def test_pcan_traceV11(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("1) 6357.2 Rx 0401 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -376,7 +378,7 @@ def test_pcan_traceV11(self): self.assertEqual(outp.timestamp.microseconds, 352700) def test_pcan_traceV12(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("1) 6357.213 1 Rx 0401 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -399,7 +401,7 @@ def test_pcan_traceV12(self): self.assertEqual(outp.timestamp.microseconds, 352743) def test_pcan_traceV13(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse("1) 6357.213 1 Rx 0401 - 8 F0 00 00 00 00 00 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -422,7 +424,7 @@ def test_pcan_traceV13(self): self.assertEqual(outp.timestamp.microseconds, 352743) def test_pcan_traceV20(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse(" 1 1059.900 DT 0300 Rx 7 00 00 00 00 04 00 00") self.assertEqual(outp.channel, 'pcanx') @@ -445,7 +447,7 @@ def test_pcan_traceV20(self): self.assertEqual(outp.timestamp.microseconds, 336543) def test_pcan_traceV21(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) + parser = cantools.logreader.Parser() outp = parser.parse(" 1 1059.900 DT 1 0300 Rx - 7 00 00 00 00 04 00 00") self.assertEqual(outp.channel, 'pcan1') @@ -852,15 +854,13 @@ def test_pcan_traceV21(self): next(frame_iter) def test_none(self): - parser = cantools.logreader.Parser(tz=utc_plus(0)) - + parser = cantools.logreader.Parser() frame_iter = iter(parser) with pytest.raises(StopIteration): next(frame_iter) def test_data_frame_repr() -> None: - parser = cantools.logreader.Parser(tz=utc_plus(0)) - + parser = cantools.logreader.Parser() outp = parser.parse("vcan0 0C8 [8] F0 00 00 00 00 00 00 00") assert repr(outp) == r"DataFrame(channel = 'vcan0', frame_id = 200, is_extended_frame = False, data = b'\xf0\x00\x00\x00\x00\x00\x00\x00', is_remote_frame = False, timestamp = None, timestamp_format = )" diff --git a/tox.ini b/tox.ini index fbf0efdb..2aff0381 100644 --- a/tox.ini +++ b/tox.ini @@ -9,6 +9,7 @@ deps = coverage==7.3.* parameterized==0.9.* windows-curses; platform_system=="Windows" and platform_python_implementation=="CPython" and python_version<"3.13" + freezegun>=1.5.2,<1.6.0 extras = plot