ctl.py

"""One-off display commands over a Unix datagram socket.

`piclockctl` (scripts/piclockctl, installed to /usr/local/bin by install.sh)
sends one-line datagrams -- "on", "off", "brightness N" -- to the socket at
PICLOCK_CTL_SOCKET. A daemon thread receives them and queues the parsed
command for the main loop, which applies it to DisplayControl and mirrors it
to HomeKit -- so a CLI command behaves exactly like a Siri command: the
override sticks until the day/night schedule's next on/off edge.

Fire-and-forget datagrams keep this tiny: no connections, no replies, one
command per datagram. The socket is mode 0660 and chgrp'd to CTL_GROUP
(install.sh creates the group and adds the service account to it), so only
root and clockctl members can command the display.

Fail-soft like HomeKit: the socket is optional operability, so a failure here
must never take down the clock -- app.run wraps startup, and the receive loop
logs and drops bad datagrams.
"""

from __future__ import annotations

import grp
import logging
import os
import socket
import threading
from typing import Callable

log = logging.getLogger("piclock")

#: Group allowed to write to the control socket (created by scripts/install.sh).
CTL_GROUP = "clockctl"

_MAX_DATAGRAM = 64  # longest valid command is "brightness 15"

#: A parsed command: ("on"|"off", None) or ("brightness", 0-15).
Command = tuple[str, "int | None"]


def parse(data: str) -> Command:
    """Parse one command datagram; raise ValueError on anything unrecognized."""
    words = data.strip().lower().split()
    if words == ["on"]:
        return "on", None
    if words == ["off"]:
        return "off", None
    if len(words) == 2 and words[0] == "brightness":
        try:
            level = int(words[1])
        except ValueError:
            raise ValueError(f"bad brightness {words[1]!r}") from None
        if not 0 <= level <= 15:
            raise ValueError(f"brightness {level} out of range 0-15")
        return "brightness", level
    raise ValueError(f"unknown command {data!r}")


class CtlSocket:
    """Listener handle: close() shuts the thread down and unlinks the socket."""

    def __init__(self, path: str, submit: Callable[[Command], None]) -> None:
        self._path = path
        try:
            os.unlink(path)  # stale socket left by an unclean stop
        except FileNotFoundError:
            pass
        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        self._sock.bind(path)
        os.chmod(path, 0o660)
        try:
            os.chown(path, -1, grp.getgrnam(CTL_GROUP).gr_gid)
        except (KeyError, PermissionError):
            # No clockctl group (dev machine) or not a member of it: the
            # socket still works for the service account and root.
            log.info("ctl socket: group %r unavailable, keeping default group", CTL_GROUP)
        self._submit = submit
        self._thread = threading.Thread(target=self._recv_loop, name="ctl", daemon=True)
        self._thread.start()
        log.info("ctl socket listening at %s", path)

    def _recv_loop(self) -> None:
        while True:
            try:
                data = self._sock.recv(_MAX_DATAGRAM)
            except OSError:  # socket closed by close()
                return
            try:
                # UnicodeDecodeError subclasses ValueError, so one except
                # covers both a non-UTF-8 datagram and an unknown command.
                self._submit(parse(data.decode("utf-8")))
            except ValueError as exc:
                log.warning("ctl socket: ignoring datagram (%s)", exc)

    def close(self) -> None:
        try:
            self._sock.close()
        finally:
            try:
                os.unlink(self._path)
            except FileNotFoundError:
                pass


def start_ctl(path: str, submit: Callable[[Command], None]) -> CtlSocket:
    """Bind the control socket at ``path`` and start the receive thread."""
    return CtlSocket(path, submit)