Coverage for core / src / sensorkit / common / time.py: 96%
74 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-02 00:03 +0000
1# SPDX-License-Identifier: Apache-2.0
2"""Time range parsing utilities supporting symbolic and absolute time specifications."""
4import functools
5import re
6import zoneinfo
7from datetime import UTC, datetime, timedelta
8from typing import Callable
10from dateutil.parser import parse as parse_datetime
11from dateutil.tz import tzlocal, tzoffset, tzutc
13_pattern = re.compile(
14 r"(?P<symbol>[A-Za-z_]+)\s*(?:(?P<op>[+-])\s*(?P<deltaspec>.*))?|(?P<timespec>.+)"
15)
16_ref_dt = datetime(year=2000, month=1, day=1, tzinfo=UTC)
17_ref_date = _ref_dt.date()
19type SymbolParseHandler = Callable[[datetime, datetime, bool], datetime]
22@functools.cache
23def _timezones():
24 now = datetime.now(UTC)
25 return {
26 info.tzname(now): info
27 for info in (
28 zoneinfo.ZoneInfo(zone) for zone in zoneinfo.available_timezones()
29 )
30 }
33def clear_timezone_cache():
34 """Invalidate the cached timezone name-to-ZoneInfo mapping so it is rebuilt on next access."""
35 _timezones.cache_clear()
38def get_tzinfo(name: str | None, offset: int | None = None):
39 """Resolve a timezone name and optional UTC offset (in seconds) to a `tzinfo` object.
41 Returns the local timezone when both arguments are absent. Raises `TimeRangeError` for
42 unrecognized timezone names.
43 """
44 if not name and offset is None:
45 return tzlocal()
47 if name:
48 tz = _timezones().get(name)
50 if not tz:
51 raise TimeRangeError(f"unknown timezone: {name}")
52 else:
53 tz = tzutc()
55 if offset is not None:
56 offset = int(tz.utcoffset(datetime.now()).total_seconds()) - offset
57 return tzoffset(f"UTC{offset // 36:+d}", offset)
59 return tz
62def parse_time_of_day(spec: str, range_min: datetime, range_max: datetime):
63 """Parse a time-of-day string into the first matching `datetime` within `[range_min, range_max]`.
65 Raises `TimeRangeParseError` if `spec` contains a date component, or `TimeRangeError` if
66 the resolved time falls outside the allowed range.
67 """
68 # Parse the input spec.
69 dt = parse_datetime(spec, default=_ref_dt, tzinfos=get_tzinfo)
71 # Make sure the spec did not contain a date component.
72 if dt.date() != _ref_date:
73 raise TimeRangeParseError("input must be time of day, not date")
75 # Find the first time in the allowed range that matches the parsed time.
76 if range_min.tzinfo != dt.tzinfo:
77 # Make sure timezones match before combining.
78 range_min = range_min.astimezone(dt.tzinfo)
80 dt = datetime.combine(range_min.date(), dt.timetz())
82 if dt < range_min:
83 dt += timedelta(days=1)
85 if dt > range_max:
86 raise TimeRangeError(f"input is outside allowed range ({dt} > {range_max})")
88 return dt
91def parse_time_delta(spec: str):
92 """Parse a time-of-day string as a `timedelta` relative to midnight of the reference date."""
93 # Use a default reference datetime to extract the delta.
94 delta_dt = parse_datetime(spec, default=_ref_dt)
96 if delta_dt.date() != _ref_date:
97 raise TimeRangeParseError("delta must be time of day, not date")
99 return delta_dt - _ref_dt
102def parse_spec(
103 spec: str,
104 range_min: datetime,
105 range_max: datetime,
106 symbol_handlers: dict[str, SymbolParseHandler] | None = None,
107 latest_match: bool = False,
108) -> datetime:
109 """Resolve a time spec string to a `datetime` within a one-day window.
111 The spec may be an absolute time-of-day string or a symbol (optionally followed by a
112 `+`/`-` delta). Symbol resolution is delegated to `symbol_handlers`.
113 """
114 assert range_max - range_min <= timedelta(days=1)
116 # Parse the input spec.
117 match = _pattern.match(spec)
119 if not match:
120 raise TimeRangeParseError("input not recognized")
122 if timespec := match["timespec"]:
123 return parse_time_of_day(timespec.upper(), range_min, range_max)
124 else:
125 symbol_handlers = symbol_handlers or {}
126 handler_func = symbol_handlers.get(match["symbol"])
128 if not handler_func:
129 raise TimeRangeParseError(f"symbol '{match['symbol']}' not recognized")
131 if deltaspec := match["deltaspec"]:
132 delta = parse_time_delta(deltaspec.lower()) * (-1 if match["op"] == "-" else 1)
133 else:
134 delta = timedelta(0)
136 # Use the matching handler to evaluate the range into a datetime.
137 dt = handler_func(
138 range_min - delta,
139 range_max - delta,
140 latest_match,
141 )
143 if dt is None:
144 raise TimeRangeError(
145 f"could not evaluate '{match['symbol']}'"
146 f" between {range_min - delta} and {range_max - delta}"
147 )
149 return dt + delta
152def parse_time_range(
153 start_spec: str,
154 end_spec: str,
155 time_ref: datetime | None = None,
156 symbol_handlers: dict[str, SymbolParseHandler] | None = None,
157):
158 """Parse a start and end spec into a `(start_dt, end_dt)` tuple relative to `time_ref`.
160 The end time is the earliest match within one day after `time_ref`; the start time is the
161 latest match within one day before the resolved end time. `time_ref` defaults to the current
162 UTC time if not provided.
163 """
164 if time_ref:
165 if time_ref.tzinfo is None:
166 raise ValueError("time_ref must have a timezone")
167 else:
168 time_ref = datetime.now(UTC)
170 # End time is the earliest spec match within one day following the input reference time.
171 end_dt = parse_spec(
172 end_spec,
173 time_ref,
174 time_ref + timedelta(days=1),
175 symbol_handlers=symbol_handlers,
176 )
178 # Start time is the latest spec match within one day preceding the end time.
179 start_dt = parse_spec(
180 start_spec,
181 end_dt - timedelta(days=1),
182 end_dt,
183 symbol_handlers=symbol_handlers,
184 latest_match=True,
185 )
187 return start_dt, end_dt
190class TimeRangeError(Exception):
191 """Raised when a resolved time is outside the permitted range or cannot be evaluated."""
194class TimeRangeParseError(TimeRangeError):
195 """Raised when a time spec string cannot be parsed or is structurally invalid."""