diff --git a/Sensor/adr_sensor/utils/timestamp_utils.py b/Sensor/adr_sensor/utils/timestamp_utils.py index d7263b59..e7b406d9 100644 --- a/Sensor/adr_sensor/utils/timestamp_utils.py +++ b/Sensor/adr_sensor/utils/timestamp_utils.py @@ -2,9 +2,18 @@ Timestamp utilities for standardized datetime handling across parsers. """ +import re from datetime import datetime, timezone from typing import Optional, Union +# Python < 3.11 fromisoformat() only accepts 3 or 6 fractional-second digits, +# but RFC 3339 writers (e.g. Go's RFC3339Nano) emit anywhere from 1 to 9. +_FRACTIONAL_SECONDS = re.compile(r"(?<=:\d\d)\.(\d+)") + + +def _fraction_to_microseconds(match: "re.Match[str]") -> str: + return "." + (match.group(1) + "000000")[:6] + def normalize_timestamp(timestamp: Union[str, int, float, datetime]) -> datetime: """Convert any timestamp format to a timezone-aware datetime in UTC. @@ -28,6 +37,7 @@ def normalize_timestamp(timestamp: Union[str, int, float, datetime]) -> datetime elif isinstance(timestamp, str): try: iso_str = timestamp.replace("Z", "+00:00") + iso_str = _FRACTIONAL_SECONDS.sub(_fraction_to_microseconds, iso_str, count=1) dt = datetime.fromisoformat(iso_str) except ValueError: try: diff --git a/Sensor/tests/test_utils.py b/Sensor/tests/test_utils.py index c14c69c5..143e19ef 100644 --- a/Sensor/tests/test_utils.py +++ b/Sensor/tests/test_utils.py @@ -91,6 +91,22 @@ def test_float_timestamp(self): dt = normalize_timestamp(1704067200.5) assert dt.year == 2024 + @pytest.mark.parametrize( + ("value", "microsecond"), + [ + ("2025-07-16T13:32:09.5Z", 500000), + ("2025-07-16T13:32:09.12345Z", 123450), + ("2025-07-16T13:32:09.123456789Z", 123456), + ("2025-07-16T13:32:09.123456789-07:00", 123456), + ("2025-07-16 13:32:09.1234567", 123456), + ], + ) + def test_iso_string_with_any_fraction_length(self, value, microsecond): + # Python < 3.11 fromisoformat() only accepts 3 or 6 fractional digits. + dt = normalize_timestamp(value) + assert dt.tzinfo == timezone.utc + assert (dt.minute, dt.second, dt.microsecond) == (32, 9, microsecond) + class TestFormatTimestampForFilename: def test_format(self):