diff --git a/frigate/test/test_dst_transitions.py b/frigate/test/test_dst_transitions.py new file mode 100644 index 0000000000..66afb40976 --- /dev/null +++ b/frigate/test/test_dst_transitions.py @@ -0,0 +1,54 @@ +"""Tests for get_dst_transitions.""" + +import datetime +import unittest + +from frigate.util.time import get_dst_transitions + + +class TestDstTransitions(unittest.TestCase): + def test_dst_transition_splits_periods_at_the_transition(self): + start = datetime.datetime(2026, 3, 7, 12, tzinfo=datetime.UTC).timestamp() + end = start + 2 * 86400 + spring = datetime.datetime(2026, 3, 8, 7, tzinfo=datetime.UTC).timestamp() + self.assertEqual( + get_dst_transitions("America/New_York", start, end), + [(start, spring, -18000), (spring, end, -14400)], + ) + + def test_dst_transition_is_not_reported_a_day_late(self): + # local midnight on the day of the change used to report the + # transition a full day after it actually happened + start = datetime.datetime(2024, 3, 10, 5, tzinfo=datetime.UTC).timestamp() + end = start + 3 * 86400 + spring = datetime.datetime(2024, 3, 10, 7, tzinfo=datetime.UTC).timestamp() + self.assertEqual( + get_dst_transitions("America/New_York", start, end), + [(start, spring, -18000), (spring, end, -14400)], + ) + + def test_dst_transition_after_the_last_daily_probe_is_found(self): + start = datetime.datetime(2024, 11, 2, 12, tzinfo=datetime.UTC).timestamp() + end = datetime.datetime(2024, 11, 3, 10, tzinfo=datetime.UTC).timestamp() + fall = datetime.datetime(2024, 11, 3, 6, tzinfo=datetime.UTC).timestamp() + self.assertEqual( + get_dst_transitions("America/New_York", start, end), + [(start, fall, -14400), (fall, end, -18000)], + ) + + def test_no_transition_returns_a_single_period(self): + start = datetime.datetime(2026, 6, 1, tzinfo=datetime.UTC).timestamp() + end = start + 5 * 86400 + self.assertEqual( + get_dst_transitions("America/New_York", start, end), + [(start, end, -14400)], + ) + + def test_invalid_zone_retains_utc_fallback(self): + self.assertEqual( + get_dst_transitions("Invalid/Timezone", 100, 200), [(100, 200, 0)] + ) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/frigate/util/time.py b/frigate/util/time.py index 861bec17f6..8d2223ce7c 100644 --- a/frigate/util/time.py +++ b/frigate/util/time.py @@ -2,6 +2,7 @@ import datetime import logging +import math from zoneinfo import ZoneInfoNotFoundError import pytz @@ -43,9 +44,33 @@ def is_current_hour(timestamp: int) -> bool: return timestamp < start_of_next_hour +def _utc_offset(tz: datetime.tzinfo, timestamp: float) -> float: + dt = datetime.datetime.fromtimestamp(timestamp, tz=datetime.UTC) + return dt.astimezone(tz).utcoffset().total_seconds() + + +def _find_transition( + tz: datetime.tzinfo, lo: float, hi: float, lo_offset: float +) -> float: + """Bisect (lo, hi] to the second where the UTC offset first differs from lo_offset.""" + # whole seconds, so the midpoint always advances (a fractional bound can + # otherwise leave the midpoint sitting on lo) and lands on the transition + low = math.floor(lo) + high = math.ceil(hi) + + while high - low > 1: + mid = (low + high) // 2 + if _utc_offset(tz, mid) == lo_offset: + low = mid + else: + high = mid + + return float(high) + + def get_dst_transitions( tz_name: str, start_time: float, end_time: float -) -> list[tuple[float, float]]: +) -> list[tuple[float, float, float]]: """ Find DST transition points and return time periods with consistent offsets. @@ -66,28 +91,24 @@ def get_dst_transitions( periods = [] current = start_time - - # Get initial offset - dt = datetime.datetime.utcfromtimestamp(current).replace(tzinfo=pytz.UTC) - local_dt = dt.astimezone(tz) - prev_offset = local_dt.utcoffset().total_seconds() period_start = start_time + prev_offset = _utc_offset(tz, current) - # Check each day for offset changes - while current <= end_time: - dt = datetime.datetime.utcfromtimestamp(current).replace(tzinfo=pytz.UTC) - local_dt = dt.astimezone(tz) - current_offset = local_dt.utcoffset().total_seconds() + # Probe at most a day ahead, capped at end_time so a transition after the + # last full day is still seen instead of silently kept in the last period. + while current < end_time: + next_probe = min(current + 86400, end_time) + next_offset = _utc_offset(tz, next_probe) - if current_offset != prev_offset: - # Found a transition - close previous period - periods.append((period_start, current, prev_offset)) - period_start = current - prev_offset = current_offset + if next_offset != prev_offset: + transition = _find_transition(tz, current, next_probe, prev_offset) + periods.append((period_start, transition, prev_offset)) + period_start = transition + prev_offset = _utc_offset(tz, transition) + current = transition + else: + current = next_probe - current += 86400 # Check daily - - # Add final period periods.append((period_start, end_time, prev_offset)) return periods