espresso_frame/server/app/quiet_hours.py

"""Quiet-hours math, extracted verbatim from the old main.py. Everything
takes the frame-like object duck-typed on quiet_hours_enabled/start/end,
timezone, and refresh_interval_s -- both the old FrameConfig and the
Frame ORM model satisfy it."""

from __future__ import annotations

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo, available_timezones

# Populated once from the OS's zoneinfo database (installed via the
# `tzdata` apt package in the Dockerfile) and offered as a <select> in the
# web UI's "Timezone" field.
ALL_TIMEZONES = sorted(available_timezones())


def valid_hhmm(s: str) -> bool:
    try:
        datetime.strptime(s, "%H:%M")
        return True
    except ValueError:
        return False


def _zoneinfo(name: str) -> ZoneInfo:
    """Falls back to UTC for an unrecognized zone name -- defensive only;
    the config-save route validates against ALL_TIMEZONES before saving,
    so this only matters for state hand-edited or written by an older
    version of this code."""
    try:
        return ZoneInfo(name)
    except Exception:
        return ZoneInfo("UTC")


def _quiet_hours_state(now: datetime, start_str: str, end_str: str) -> tuple[bool, datetime | None]:
    """Whether `now` falls inside the quiet-hours window, and the next
    boundary: if inside, when it ends; if outside, when it next starts.
    Handles a window that wraps past midnight (e.g. 22:00-07:00). Returns
    (False, None) for a degenerate window (start == end)."""
    sh, sm = (int(x) for x in start_str.split(":"))
    eh, em = (int(x) for x in end_str.split(":"))
    start = now.replace(hour=sh, minute=sm, second=0, microsecond=0)
    end = now.replace(hour=eh, minute=em, second=0, microsecond=0)

    if start == end:
        return False, None

    if start < end:
        # Same-day window, e.g. 13:00-15:00. End is exclusive, so being
        # exactly at `end` counts as already outside the window.
        if start <= now < end:
            return True, end
        if now < start:
            return False, start
        return False, start + timedelta(days=1)

    # Wraps midnight, e.g. 22:00-07:00.
    if now >= start:
        return True, end + timedelta(days=1)
    if now < end:
        return True, end
    return False, start


def _quiet_hours_span_s(start_str: str, end_str: str) -> int:
    """Duration of the quiet-hours window in seconds, wrap-aware."""
    sh, sm = (int(x) for x in start_str.split(":"))
    eh, em = (int(x) for x in end_str.split(":"))
    span_min = ((eh * 60 + em) - (sh * 60 + sm)) % (24 * 60)
    return span_min * 60


def effective_refresh_interval_s(cfg) -> int:
    """The refresh interval actually handed to the device: its configured
    value, unless quiet hours are enabled, in which case it's clamped so
    the device sleeps through the whole window instead of waking inside
    it. A device already mid-sleep when quiet hours begin can still land
    one wake inside the window (nothing server-side can prevent that
    without touching the firmware) -- but from that wake on, it's told to
    sleep exactly until the window ends."""
    if not cfg.quiet_hours_enabled:
        return cfg.refresh_interval_s
    now = datetime.now(_zoneinfo(cfg.timezone))
    in_quiet, boundary = _quiet_hours_state(now, cfg.quiet_hours_start, cfg.quiet_hours_end)
    if boundary is None:
        return cfg.refresh_interval_s
    seconds_to_boundary = max(1, int((boundary - now).total_seconds()))
    if in_quiet:
        return seconds_to_boundary
    return min(cfg.refresh_interval_s, seconds_to_boundary)


def in_quiet_hours(cfg) -> bool:
    """Whether quiet hours are in effect right now -- separate from
    effective_refresh_interval_s, which only shapes what the *device* is
    told to sleep for. This instead gates photo_queue.get_current()'s
    time-based advance, since that check runs independent of the device
    (also triggered by the web UI's queue endpoint, e.g. an open browser
    tab polling overnight) and would otherwise happily advance the
    current photo mid-quiet-hours on raw elapsed time alone."""
    if not cfg.quiet_hours_enabled:
        return False
    now = datetime.now(_zoneinfo(cfg.timezone))
    in_quiet, _ = _quiet_hours_state(now, cfg.quiet_hours_start, cfg.quiet_hours_end)
    return in_quiet


def max_expected_gap_s(cfg) -> int:
    """Longest gap between wakes the device might legitimately have --
    normally just refresh_interval_s, but quiet hours can make the real
    gap much longer, and the "overdue" check shouldn't mistake a device
    quietly sleeping through the night for a dead one."""
    gap = cfg.refresh_interval_s
    if cfg.quiet_hours_enabled:
        gap = max(gap, _quiet_hours_span_s(cfg.quiet_hours_start, cfg.quiet_hours_end))
    return gap