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 stat
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
        self._closed = threading.Event()
        try:
            mode = os.lstat(path).st_mode
        except FileNotFoundError:
            pass
        else:
            if not stat.S_ISSOCK(mode):
                raise ValueError(f"refusing to replace non-socket path {path!r}")
            os.unlink(path)  # stale socket left by an unclean stop
        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        bound = False
        try:
            self._sock.bind(path)
            bound = True
            os.chmod(path, 0o660)
            try:
                os.chown(path, -1, grp.getgrnam(CTL_GROUP).gr_gid)
            except (KeyError, PermissionError):
                log.info("ctl socket: group %r unavailable, keeping default group", CTL_GROUP)
            # Closing a socket in another thread need not interrupt recv().
            self._sock.settimeout(0.2)
        except Exception:
            self._sock.close()
            if bound:
                os.unlink(path)
            raise
        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 not self._closed.is_set():
            try:
                data = self._sock.recv(_MAX_DATAGRAM + 1)
            except socket.timeout:
                continue
            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.
                if len(data) > _MAX_DATAGRAM:
                    raise ValueError("datagram too long")
                self._submit(parse(data.decode("utf-8")))
            except ValueError as exc:
                log.warning("ctl socket: ignoring datagram (%s)", exc)

    def close(self) -> None:
        self._closed.set()
        self._thread.join(timeout=1)
        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)