app.py

"""PiClock main loop: read local time, render it, push it to the display.

Runs until SIGTERM/SIGINT (systemd stop or Ctrl-C), then blanks the display.
"""

from __future__ import annotations

import logging
import queue
import signal
import threading
import time
from datetime import datetime

from . import clock, ctl, schedule, timesync, wifi
from .config import Config
from .control import DisplayControl
from .display import make_display

log = logging.getLogger("piclock")

# How often to poll for NTP sync. Decoupled from the (much faster) display tick
# so a no-network boot doesn't spawn `timedatectl` 5x/sec forever; the display
# shows `----` meanwhile, so a few seconds' lag is invisible. Once synced we
# keep polling (the sync dot tracks losing sync too) but relax the interval —
# sync loss is rare and a minute's lag on the indicator is invisible.
_SYNC_POLL_SECONDS = 5.0
_SYNCED_POLL_SECONDS = 60.0

# Shown when the display is switched off (by the day/night schedule or HomeKit):
# all segments dark, so no LEDs are lit — this is the point, sparing the 7-seg's
# lifespan.
_BLANK_TEXT = "    "


def _dot_lit(nominal: bool, trouble: bool) -> bool:
    """Indicator polarity: status lights are lit while nominal, trouble lights
    only while off-nominal (see Config.trouble_dots)."""
    return (not nominal) if trouble else nominal


def run(config: Config | None = None) -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )
    config = config or Config.from_env()

    stop = threading.Event()

    def _handle_signal(signum, _frame):
        log.info("received signal %s, shutting down", signal.Signals(signum).name)
        stop.set()

    signal.signal(signal.SIGTERM, _handle_signal)
    signal.signal(signal.SIGINT, _handle_signal)

    log.info("starting %s", config)
    display = make_display(config)
    synced = False
    # Latches True at the first sync and stays True: `----` is only for the
    # boot window where the clock has never been right. After that, losing NTP
    # keeps the (free-running, still approximately right) time on the display
    # and only the sync dot goes dark.
    ever_synced = False
    # Whether syncs are still *arriving* (timesyncd's stamp file is recent).
    # Distinct from `synced`, which is the kernel's is-the-clock-trustworthy
    # estimate and lags an outage by ~9 hours: `synced` gates the dashes and
    # the free-run warning, freshness only gates the sync dot. Starts True so
    # the dot can't flicker before the first check.
    sync_fresh = True

    # Display on/off + brightness state, shared by everything that controls the
    # display. The optional day/night schedule (evaluated below, in-process)
    # drives it; when HomeKit is enabled its Lightbulb accessory drives the same
    # object. Always present so the schedule works with or without HomeKit.
    control = DisplayControl(enabled=True, brightness=config.brightness)

    # Optional CLI control (piclockctl): the socket thread queues parsed
    # commands here and the loop drains them below — same pattern, and same
    # override semantics, as HomeKit. SimpleQueue is safe for the
    # ctl-thread-producer / loop-consumer handoff. Optional operability, so
    # startup is fail-soft.
    pending_commands: queue.SimpleQueue[ctl.Command] = queue.SimpleQueue()
    ctl_socket = None
    if config.ctl_socket:
        try:
            ctl_socket = ctl.start_ctl(config.ctl_socket, pending_commands.put)
        except Exception:  # optional feature; never crash the clock
            log.exception("control socket failed to start; continuing without it")

    # Optional HomeKit: its Lightbulb accessory drives the same `control`.
    # Imported lazily so the core needs no HomeKit deps. HomeKit is non-essential
    # to the clock's job, so a startup failure here (missing deps, unwritable
    # state dir, etc.) must never take the appliance down — log it and run the
    # clock without HomeKit, mirroring the display layer's fail-soft contract.
    homekit = None
    if config.homekit:
        try:
            from . import homekit as homekit_mod

            homekit = homekit_mod.start_homekit(config, control)
        except Exception:  # optional feature; never crash the clock
            log.exception("HomeKit failed to start; continuing without it")

    next_sync_check = 0.0  # monotonic deadline; 0 => check on the first iteration

    # The schedule is edge-triggered: each tick computes the phase for "now" and
    # applies it only when it *changes*. That keeps a HomeKit/Siri override in
    # force until the schedule's next edge, and — because the first tick counts
    # as an edge (last_phase starts as None) — a reboot or deploy mid-window
    # comes up in the correct state with no external reconciliation.
    last_phase: str | None = None

    try:
        while not stop.is_set():
            if time.monotonic() >= next_sync_check:
                was_synced, was_fresh = synced, sync_fresh
                synced = timesync.is_ntp_synced()
                age = timesync.last_sync_age()
                # No stamp file (no timesyncd, e.g. dev machine) -> fall back
                # to NTPSynchronized alone rather than darkening the dot.
                sync_fresh = age is None or age <= config.sync_fresh_seconds
                next_sync_check = time.monotonic() + (
                    _SYNCED_POLL_SECONDS if synced else _SYNC_POLL_SECONDS
                )
                if synced and not was_synced:
                    ever_synced = True
                    log.info("system clock synchronized")
                elif was_synced and not synced:
                    log.warning("NTP sync lost; free-running until it returns")
                if was_fresh and not sync_fresh:
                    log.warning(
                        "no successful NTP sync for %.0fs; sync dot dark", age
                    )
                elif sync_fresh and not was_fresh:
                    log.info("NTP syncs arriving again")

            now = datetime.now()

            if config.schedule:
                phase = schedule.phase(
                    now.time(),
                    off_at=config.off_at,
                    on_at=config.on_at,
                    dim_at=config.dim_at,
                )
                if phase != last_phase:
                    # What each edge asserts (and what it leaves alone — the
                    # dim edge must not relight a manually-switched-off
                    # display) is decided in schedule.edge, where it's tested.
                    enabled, brightness = schedule.edge(
                        last_phase,
                        phase,
                        wake_brightness=config.wake_brightness,
                        night_brightness=config.night_brightness,
                    )
                    last_phase = phase
                    if enabled is not None:
                        control.enabled = enabled
                    if brightness is not None:
                        control.brightness = brightness
                    log.info("schedule: display %s", phase)
                    if homekit is not None:
                        # Mirror the schedule's change back to the Home app tile.
                        homekit.notify_state(control.enabled, control.brightness)

            # Drained after the schedule so a command that races an edge in
            # the same tick wins; like any override, it then holds until the
            # schedule's next on/off edge.
            while True:
                try:
                    action, level = pending_commands.get_nowait()
                except queue.Empty:
                    break
                if action == "brightness":
                    control.brightness = level
                else:
                    control.enabled = action == "on"
                log.info(
                    "ctl: display %s",
                    action if level is None else f"brightness {level}",
                )
                if homekit is not None:
                    homekit.notify_state(control.enabled, control.brightness)

            if control.enabled:
                state = clock.render(
                    now,
                    synced,
                    ever_synced=ever_synced,
                    hour24=config.hour24,
                    blink=config.blink_colon,
                )
                text, colon = state.text, state.colon
                sync_ok = state.sync_dot and sync_fresh
                # Wi-Fi status is not a function of the time, so it joins here
                # rather than passing through render(). Polled per tick: it is
                # one sysfs read (~140 us), and per-second freshness is the
                # point of the indicator.
                wifi_ok = wifi.is_wifi_up(config.wifi_iface) if config.wifi_dot else True
                sync_dot = config.sync_dot and _dot_lit(sync_ok, config.trouble_dots)
                wifi_dot = config.wifi_dot and _dot_lit(wifi_ok, config.trouble_dots)
            else:
                # display off
                text, colon, sync_dot, wifi_dot = _BLANK_TEXT, False, False, False

            # The display backend is fault-tolerant (see make_display): show()
            # never raises on I/O errors, it logs and reconnects on its own. So a
            # missing or bumped display can't crash the loop.
            display.set_brightness(control.brightness)  # deduped; cheap per-tick
            display.show(text, colon, sync_dot, wifi_dot)

            stop.wait(config.tick_seconds)
    finally:
        if ctl_socket is not None:
            try:
                ctl_socket.close()
            except Exception:  # best effort on the way out
                log.exception("error while closing control socket")
        if homekit is not None:
            try:
                homekit.stop()
            except Exception:  # best effort on the way out
                log.exception("error while stopping HomeKit")
        try:
            display.close()
        except Exception:  # best effort on the way out
            log.exception("error while closing display")
        log.info("stopped")