find DST transitions to the second

`get_dst_transitions` probed the offset once every 24 hours from the start time and reported a change at the first probe after it, up to a day late, so events, review items and recordings near a transition were grouped into days with the old offset. A transition after the last daily probe wasn't found at all. The end of the range is probed too now, and a probe that sees the offset change bisects the interval to the second of the transition.
This commit is contained in:
Josh Hawkins
2026-09-18 06:55:36 -05:00
parent 54ba07917d
commit 51eaae4857
2 changed files with 94 additions and 19 deletions
+54
View File
@@ -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)
+40 -19
View File
@@ -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