243 lines
8.9 KiB
Python
243 lines
8.9 KiB
Python
"""Calendar-aware interval arithmetic (no external dependencies).
|
||
|
||
Used by the scheduling model so an interval can be expressed in days, weeks,
|
||
months, or years. Month/year steps are calendar-aware and clamp the day to the
|
||
target month's last day (e.g. Jan 31 + 1 month -> Feb 28/29), avoiding the
|
||
drift you get from approximating "monthly" as 30 days.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import calendar
|
||
from collections.abc import Iterator
|
||
from datetime import UTC, date, datetime, time, timedelta
|
||
from typing import Any
|
||
|
||
from homeassistant.util import dt as dt_util
|
||
|
||
from ..const import TIME_HHMMSS_PATTERN
|
||
from .workday import is_business_day
|
||
|
||
INTERVAL_UNITS = ("days", "weeks", "months", "years")
|
||
|
||
|
||
def parse_hhmm(value: Any) -> time | None:
|
||
"""``"HH:MM"`` or ``"HH:MM:SS"`` (HA's TimeSelector) → ``time`` to the
|
||
minute; ``None`` when absent or malformed (hours 0–23, minutes 0–59).
|
||
|
||
The one parser behind the status sensor, the calendar entity and the
|
||
model's sub-day overdue refinement — three hand-copied ``split(":")``
|
||
blocks before — and behind :func:`normalize_hhmm`.
|
||
"""
|
||
if not isinstance(value, str):
|
||
return None
|
||
text = value.strip()
|
||
if not TIME_HHMMSS_PATTERN.match(text):
|
||
return None
|
||
hours, minutes = text.split(":")[:2]
|
||
return time(int(hours), int(minutes))
|
||
|
||
|
||
def local_date_from_iso(value: Any) -> str | None:
|
||
"""The calendar date (``YYYY-MM-DD``) of an ISO timestamp *in HA's local
|
||
zone*. ``value[:10]`` is right for stamps written locally but off by one
|
||
around midnight for imported ``+00:00`` stamps; this is the one place
|
||
that does it properly. ``None`` when the value is not a timestamp."""
|
||
if not isinstance(value, str) or not value:
|
||
return None
|
||
try:
|
||
parsed = datetime.fromisoformat(value)
|
||
except ValueError:
|
||
return value[:10] if len(value) >= 10 and value[4] == "-" else None
|
||
if parsed.tzinfo is None:
|
||
return parsed.date().isoformat()
|
||
return dt_util.as_local(parsed).date().isoformat()
|
||
|
||
|
||
def normalize_hhmm(value: Any) -> str | None:
|
||
"""Canonical ``"HH:MM"`` for a stored time, dropping TimeSelector seconds;
|
||
``None`` when the value does not parse (the caller drops the field)."""
|
||
parsed = parse_hhmm(value)
|
||
return f"{parsed.hour:02d}:{parsed.minute:02d}" if parsed is not None else None
|
||
|
||
|
||
def parse_iso_date(value: str | None) -> date | None:
|
||
"""ISO date string → ``date``; ``None`` when absent or malformed."""
|
||
if not value:
|
||
return None
|
||
try:
|
||
return date.fromisoformat(value)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def parse_persisted_utc(value: str | None) -> datetime | None:
|
||
"""Persisted ISO timestamp → TZ-aware ``datetime``; ``None`` when absent
|
||
or malformed.
|
||
|
||
Live writes use ``dt_util.utcnow().isoformat()`` (TZ-aware), but older
|
||
payloads may be naive — those are assumed UTC, because subtracting a naive
|
||
datetime from ``dt_util.utcnow()`` raises ``TypeError``. One home for the
|
||
block the threshold and state-change triggers each hand-rolled (and the
|
||
runtime trigger FORGOT — a naive ``on_since`` crashed its elapsed math).
|
||
"""
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = datetime.fromisoformat(value)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
if parsed.tzinfo is None:
|
||
parsed = parsed.replace(tzinfo=UTC)
|
||
return parsed
|
||
|
||
|
||
def _add_months(anchor: date, months: int) -> date:
|
||
"""Advance ``anchor`` by ``months`` calendar months, clamping the day."""
|
||
total = anchor.month - 1 + months
|
||
year = anchor.year + total // 12
|
||
month = total % 12 + 1
|
||
last_day = calendar.monthrange(year, month)[1]
|
||
return date(year, month, min(anchor.day, last_day))
|
||
|
||
|
||
def add_interval(anchor: date, n: int, unit: str = "days") -> date:
|
||
"""Return ``anchor`` advanced by ``n`` of ``unit``.
|
||
|
||
``unit`` is one of days / weeks / months / years; anything else (or None)
|
||
falls back to days, which keeps existing day-based tasks working unchanged.
|
||
"""
|
||
if unit == "weeks":
|
||
return anchor + timedelta(weeks=n)
|
||
if unit == "months":
|
||
return _add_months(anchor, n)
|
||
if unit == "years":
|
||
return _add_months(anchor, n * 12)
|
||
return anchor + timedelta(days=n)
|
||
|
||
|
||
def nth_weekday_of_month(year: int, month: int, nth: int, weekday: int) -> date | None:
|
||
"""The ``nth`` ``weekday`` of ``(year, month)``; ``None`` if it doesn't exist.
|
||
|
||
``weekday`` is 0=Mon … 6=Sun (``date.weekday()``). ``nth`` is 1..5, or -1 for
|
||
the last occurrence. A 5th occurrence that the month doesn't have → ``None``.
|
||
"""
|
||
last_day = calendar.monthrange(year, month)[1]
|
||
if nth == -1:
|
||
anchor = date(year, month, last_day)
|
||
return anchor - timedelta(days=(anchor.weekday() - weekday) % 7)
|
||
first = date(year, month, 1)
|
||
day = 1 + (weekday - first.weekday()) % 7 + (nth - 1) * 7
|
||
if day > last_day:
|
||
return None
|
||
return date(year, month, day)
|
||
|
||
|
||
def next_weekday_in_set(ref: date, weekdays: tuple[int, ...], *, inclusive: bool) -> date | None:
|
||
"""Next date on/after ``ref`` whose weekday is in ``weekdays`` (0=Mon…6=Sun).
|
||
|
||
``inclusive`` includes ``ref`` itself; otherwise the search starts the next
|
||
day. ``None`` when ``weekdays`` is empty.
|
||
"""
|
||
if not weekdays:
|
||
return None
|
||
start = ref if inclusive else ref + timedelta(days=1)
|
||
for offset in range(7):
|
||
candidate = start + timedelta(days=offset)
|
||
if candidate.weekday() in weekdays:
|
||
return candidate
|
||
return None # pragma: no cover (unreachable: a non-empty weekday set always matches within 7 days)
|
||
|
||
|
||
def _iter_months(year: int, month: int, limit: int = 60) -> Iterator[tuple[int, int]]:
|
||
"""Yield ``(year, month)`` forward from the given month, ``limit`` times."""
|
||
for _ in range(limit):
|
||
yield year, month
|
||
month += 1
|
||
if month > 12:
|
||
month = 1
|
||
year += 1
|
||
|
||
|
||
def next_nth_weekday(
|
||
ref: date,
|
||
nth: int,
|
||
weekday: int,
|
||
months: tuple[int, ...] | None = None,
|
||
*,
|
||
inclusive: bool,
|
||
) -> date | None:
|
||
"""Next ``nth``-``weekday``-of-month occurrence on/after ``ref``.
|
||
|
||
Optionally restricted to ``months`` (1=Jan…12=Dec). ``None`` if no occurrence
|
||
is found within a bounded horizon (e.g. a 5th weekday restricted to a month
|
||
that never has one).
|
||
"""
|
||
for year, month in _iter_months(ref.year, ref.month):
|
||
if months and month not in months:
|
||
continue
|
||
occ = nth_weekday_of_month(year, month, nth, weekday)
|
||
if occ is not None and (occ >= ref if inclusive else occ > ref):
|
||
return occ
|
||
return None
|
||
|
||
|
||
def next_day_of_month(
|
||
ref: date,
|
||
day: int,
|
||
months: tuple[int, ...] | None = None,
|
||
*,
|
||
inclusive: bool,
|
||
) -> date | None:
|
||
"""Next ``day``-of-month occurrence on/after ``ref`` (day clamped to month).
|
||
|
||
``day`` 1..31 is clamped to the month's length (e.g. 31 → Feb 28/29);
|
||
``day == -1`` means the LAST day of the month (#83). Optionally restricted
|
||
to ``months``.
|
||
"""
|
||
for year, month in _iter_months(ref.year, ref.month):
|
||
if months and month not in months:
|
||
continue
|
||
month_len = calendar.monthrange(year, month)[1]
|
||
clamped = month_len if day == -1 else min(day, month_len)
|
||
candidate = date(year, month, clamped)
|
||
if candidate >= ref if inclusive else candidate > ref:
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def roll_back_to_business_day(d: date) -> date:
|
||
"""Roll a date back to the preceding business day (business days unchanged).
|
||
|
||
Used by the day-of-month schedule's ``business`` flag (#83): "last business
|
||
day of the month" = last day rolled back past non-business days. What
|
||
counts as a business day comes from :mod:`.workday`: plain Mon-Fri by
|
||
default, or the user's Workday integration configuration (public holidays,
|
||
custom working weekdays) when one is set up.
|
||
|
||
Bounded: a pathological provider that never yields a business day within
|
||
two weeks returns *d* unchanged rather than walking off into the past.
|
||
"""
|
||
candidate = d
|
||
for _ in range(14):
|
||
if is_business_day(candidate):
|
||
return candidate
|
||
candidate -= timedelta(days=1)
|
||
return d
|
||
|
||
|
||
def interval_span_days(n: int | None, unit: str = "days") -> int:
|
||
"""Length of one ``n``-``unit`` interval in days, calendar-aware.
|
||
|
||
Measured as the distance from a fixed reference date to that date advanced
|
||
by the interval, so months/years reflect real calendar lengths instead of
|
||
the raw count. Used so things like the warning window aren't capped by the
|
||
bare interval *count* for month/year tasks (issue #58/#59 — a 6-month task
|
||
must not collapse a 14-day warning to ``min(14, 6)``). Returns 0 when there
|
||
is no positive interval.
|
||
"""
|
||
if not n or n <= 0:
|
||
return 0
|
||
ref = date(2001, 1, 1) # common (non-leap) year; representative span
|
||
return (add_interval(ref, n, unit) - ref).days
|