musicmouse/python-backend/musicmouse/config.py

"""Config schema and loading.

Validation is strict on purpose: unknown keys are rejected (a typo'd setting that is
silently ignored is worse than a startup failure), and every problem in the file is
reported at once rather than one per run.
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Annotated, Any, Final, Literal, Self

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    PlainValidator,
    ValidationError,
    ValidationInfo,
    field_validator,
    model_validator,
)
from ruamel.yaml import YAML
from ruamel.yaml.error import YAMLError

from musicmouse.color import ColorRGBW, parse_color
from musicmouse.hardware import NO_FIGURE_TAG, RFID_TAG_LENGTH
from musicmouse.library.podcast_feeds import DEFAULT_EPISODE_LIMIT

_log = logging.getLogger(__name__)

__all__ = [
    "SIMULATE",
    "Config",
    "ConfigError",
    "Digit",
    "FigureColors",
    "FigureConfig",
    "GeneralConfig",
    "HaConfig",
    "HaDeviceConfig",
    "LibraryConfig",
    "LircConfig",
    "MqttConfig",
    "RemoteSlotConfig",
    "TippenConfig",
    "WebConfig",
    "format_validation_error",
    "load_config",
]

#: Number keys on the IR remote, as lircd's ``BTN_0``..``BTN_9`` map to them.
type Digit = Literal["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]

DEFAULT_AUDIO_EXTENSIONS = (".mp3", ".ogg", ".oga", ".opus", ".flac", ".wav", ".m4a", ".aac")

#: Stand-in value for ``serial_port`` and ``alsa_device``. Running without the mouse or
#: without sound is a supported setup, but it has to be *asked for*: a missing key is an
#: error, so a config that lost a line fails loudly instead of booting into a silent
#: mouse that looks like it is working.
SIMULATE: Final = "simulate"


class ConfigError(Exception):
    """Raised with an already human-readable, multi-line message."""


def _parse_tag_id(value: Any) -> bytes:
    if isinstance(value, bytes):
        raw = value
    else:
        if not isinstance(value, str):
            raise ValueError(f"expected a hex string, got {type(value).__name__}")
        text = value.strip().replace(":", "").replace(" ", "")
        try:
            raw = bytes.fromhex(text)
        except ValueError:
            raise ValueError(f"{value!r} is not a valid hex string") from None
    if len(raw) != RFID_TAG_LENGTH:
        raise ValueError(
            f"expected {RFID_TAG_LENGTH} bytes ({RFID_TAG_LENGTH * 2} hex digits), "
            f"got {len(raw)} ({raw.hex()!r})"
        )
    if raw == NO_FIGURE_TAG:
        raise ValueError("the all-zero tag id is reserved for 'no figure on the reader'")
    return raw


Color = Annotated[ColorRGBW, PlainValidator(parse_color)]
TagId = Annotated[bytes, PlainValidator(_parse_tag_id)]

_COLOR_ROLES = ("primary", "secondary", "bg", "accent")


class _Strict(BaseModel):
    model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)


class FigureColors(_Strict):
    """The four colours of a figure, given in config as a list of colour strings."""

    primary: Color
    secondary: Color
    bg: Color
    accent: Color

    @model_validator(mode="before")
    @classmethod
    def _accept_sequence(cls, data: Any) -> Any:
        if isinstance(data, (list, tuple)):
            if len(data) != len(_COLOR_ROLES):
                raise ValueError(
                    f"expected exactly {len(_COLOR_ROLES)} colors "
                    f"({', '.join(_COLOR_ROLES)}), got {len(data)}"
                )
            return dict(zip(_COLOR_ROLES, data, strict=True))
        return data


def _resolve_folder(
    folder: Path, info: ValidationInfo, *, must_exist: bool, kind: Literal["dir", "file"] = "dir"
) -> Path:
    """Make a configured path absolute against the config file, and optionally check it."""
    context = info.context or {}
    base = context.get("config_dir")
    if base is not None and not folder.is_absolute():
        folder = (Path(base) / folder).resolve()
    if must_exist and context.get("check_paths", True):
        exists = folder.is_file() if kind == "file" else folder.is_dir()
        if not exists:
            noun = "file" if kind == "file" else "directory"
            raise ValueError(f"no such {noun}: {folder}")
    return folder


class MqttConfig(_Strict):
    server: str
    port: int = Field(default=1883, ge=1, le=65535)
    user: str | None = None
    password: str | None = None
    base_topic: str = "musicmouse"
    discovery_prefix: str = "homeassistant"
    device_id: str = "musicmouse"
    device_name: str = "Music Mouse"
    reconnect_interval: float = Field(default=10.0, gt=0)


class LircConfig(_Strict):
    """TCP client for lircd's classic network protocol - see ``ansible/roles/pi_lirc``.

    Omit the whole section to run without an IR remote.
    """

    host: str
    #: This deployment's lircd listens on 2222 (see the ansible role); lircd's own
    #: default is 8765, so this is worth overriding rather than assuming.
    port: int = Field(default=2222, ge=1, le=65535)
    #: Only button events from this remote are acted on - other remotes registered
    #: with the same lircd (an LED remote, say) are ignored.
    remote_name: str = "Hauppauge"
    reconnect_interval: float = Field(default=5.0, gt=0)


class LibraryConfig(_Strict):
    """Where the music lives.

    One path. The shelves underneath it - ``Figuren``, ``Musik``, ``Hoerbuecher``,
    ``Kinderpodcasts`` - are fixed names, not settings; see
    :mod:`musicmouse.library.sections`.
    """

    root: Path
    #: Scan results, extracted cover art and track analysis. Relative to this file.
    cache: Path = Path(".musicmouse-cache")
    #: How many tracks background analysis may work on at once, each in its own worker
    #: process. Omit for one per core bar one (see
    #: :func:`musicmouse.library.workers.default_worker_count`); set it to 1 to keep
    #: analysis to a single process on a machine that has other work to do.
    analysis_workers: int | None = Field(default=None, ge=1)

    @field_validator("root")
    @classmethod
    def _resolve_root(cls, folder: Path, info: ValidationInfo) -> Path:
        return _resolve_folder(folder, info, must_exist=True)

    @field_validator("cache")
    @classmethod
    def _resolve_cache(cls, folder: Path, info: ValidationInfo) -> Path:
        return _resolve_folder(folder, info, must_exist=False)

    @property
    def figure_folder(self) -> Path:
        return self.root / "Figuren"


class WebConfig(_Strict):
    """The web front-end. Omit the whole section to run without it."""

    #: A LAN appliance with no auth; binding to all interfaces is the point.
    host: str = "0.0.0.0"
    port: int = Field(default=8080, ge=1, le=65535)
    #: Built frontend to serve at ``/``. Omit to expose only the JSON API.
    static_dir: Path | None = None

    @field_validator("static_dir")
    @classmethod
    def _resolve_static(cls, folder: Path | None, info: ValidationInfo) -> Path | None:
        return None if folder is None else _resolve_folder(folder, info, must_exist=False)


class HaDeviceConfig(_Strict):
    """One Home Assistant entity to expose to the room-control page ("Mein Zimmer")."""

    entity_id: str
    name: str | None = None


class HaConfig(_Strict):
    """Home Assistant integration for the room-control page ("Mein Zimmer").

    The backend never calls Home Assistant itself - it only hands the browser the
    server URL, the token, and these two ordered lists. Control happens directly from
    the browser to Home Assistant's own REST API, so this token grants full HA control
    to anything on the LAN that can reach musicmouse. See config.yml.example.
    """

    url: str
    token: str
    #: Order is preserved and drives the device card grid on the room page.
    devices: list[HaDeviceConfig] = Field(default_factory=list)
    #: Order is preserved and drives the scene pill row on the room page.
    scenes: list[HaDeviceConfig] = Field(default_factory=list)

    @model_validator(mode="after")
    def _check_something_configured(self) -> Self:
        if not self.devices and not self.scenes:
            raise ValueError("configure at least one device or scene, or omit the ha section")
        return self


class TippenConfig(_Strict):
    """The typing game. Omit the whole section to run without it.

    The curriculum is content, not device settings, so it lives in its own file
    (``curriculum_file``) rather than inline here - see ``tippen-curriculum.yml.example``.
    ``progress_file`` is written by the app itself, not hand-edited, and defaults to a
    name next to ``config.yml`` if not given a folder of its own.
    """

    curriculum_file: Path
    progress_file: Path = Path("tippen-progress.json")

    @field_validator("curriculum_file")
    @classmethod
    def _resolve_curriculum_file(cls, path: Path, info: ValidationInfo) -> Path:
        return _resolve_folder(path, info, must_exist=True, kind="file")

    @field_validator("progress_file")
    @classmethod
    def _resolve_progress_file(cls, path: Path, info: ValidationInfo) -> Path:
        return _resolve_folder(path, info, must_exist=False, kind="file")


class GeneralConfig(_Strict):
    library: LibraryConfig

    #: Serial port the ESP32 is on, or ``"simulate"`` to run without the mouse: the
    #: web front-end is a complete way to drive the player on its own. Required.
    serial_port: str
    baudrate: int = Field(default=115200, gt=0)
    reconnect_interval: float = Field(default=5.0, gt=0)

    #: ALSA output device passed to VLC, e.g. ``"hw:0,0"`` or ``"default"``, or
    #: ``"simulate"`` for a player that makes no sound. Required.
    alsa_device: str

    mqtt: MqttConfig | None = None
    web: WebConfig | None = None
    ha: HaConfig | None = None
    lirc: LircConfig | None = None
    tippen: TippenConfig | None = None

    min_volume: int = Field(default=0, ge=0, le=200)
    max_volume: int = Field(default=100, ge=0, le=200)
    initial_volume: int = Field(default=50, ge=0, le=200)
    volume_increment: int = Field(default=5, ge=1, le=100)
    button_leds_brightness: float = Field(default=0.5, ge=0, le=1)

    audio_extensions: tuple[str, ...] = DEFAULT_AUDIO_EXTENSIONS

    #: How many episodes of each podcast show to keep, newest first. A show that has
    #: published for years grows without bound and will eventually fill the device's SD
    #: card. ``null`` keeps every episode, and minding the free space is then on you.
    #:
    #: Lowering this *deletes* the episodes that fall outside the window on the next
    #: poll, and an episode that has aged out of its feed cannot be fetched again.
    podcast_episode_limit: int | None = Field(default=DEFAULT_EPISODE_LIMIT, ge=1)

    @property
    def serial_simulated(self) -> bool:
        return self.serial_port == SIMULATE

    @property
    def audio_simulated(self) -> bool:
        return self.alsa_device == SIMULATE

    @model_validator(mode="after")
    def _check_volumes(self) -> Self:
        if self.min_volume > self.max_volume:
            raise ValueError(
                f"min_volume ({self.min_volume}) must not exceed max_volume ({self.max_volume})"
            )
        if not self.min_volume <= self.initial_volume <= self.max_volume:
            raise ValueError(
                f"initial_volume ({self.initial_volume}) must lie between "
                f"min_volume ({self.min_volume}) and max_volume ({self.max_volume})"
            )
        return self


class FigureConfig(_Strict):
    #: RFID tag id as hex, e.g. "04a1b2c3d4".
    id: TagId
    colors: FigureColors
    #: What this figure holds. Unlike the other shelves a figure folder is named after
    #: the figurine rather than its contents, so nothing on disk says whether it is an
    #: album or an audiobook - and the browse view draws the two differently.
    kind: Literal["music", "book"] = "music"


class RemoteSlotConfig(_Strict):
    """What a number key on the IR remote plays.

    ``"album"``: ``target`` is an ``Album.id``, always started from track 0 - a music
    album or an audiobook. ``"series"``: ``target`` is a podcast show name (an
    ``Album.series``); resolved to that show's newest episode fresh on every press,
    since a podcast show is not itself one playable thing in this library - each
    episode is its own album.
    """

    target_kind: Literal["album", "series"]
    target: str


class Config(_Strict):
    general: GeneralConfig
    figures: dict[str, FigureConfig] = Field(min_length=1)
    #: Number key (0-9) -> what it plays. Empty by default: a fresh install has no
    #: assignments, and that is not an error.
    remote: dict[Digit, RemoteSlotConfig] = Field(default_factory=dict)

    @model_validator(mode="after")
    def _check_unique_tag_ids(self) -> Self:
        seen: dict[bytes, str] = {}
        for name, figure in self.figures.items():
            if (other := seen.get(figure.id)) is not None:
                raise ValueError(
                    f"figures {other!r} and {name!r} both use tag id {figure.id.hex()}"
                )
            seen[figure.id] = name
        return self

    @property
    def tag_map(self) -> dict[bytes, str]:
        """Tag id -> figure name, as handed to the device."""
        return {figure.id: name for name, figure in self.figures.items()}

    @property
    def figure_kinds(self) -> dict[str, Literal["music", "book"]]:
        """Figure name -> what it holds, as handed to the library scanner."""
        return {name: figure.kind for name, figure in self.figures.items()}

    def folder_for(self, figure: str) -> Path:
        return self.general.library.figure_folder / figure


def format_validation_error(error: ValidationError) -> str:
    """Render a pydantic error as one short ``path: message`` line per problem."""
    lines: list[str] = []
    for entry in error.errors():
        location = ".".join(
            f"[{part}]" if isinstance(part, int) else str(part) for part in entry["loc"]
        ).replace(".[", "[")
        message = entry["msg"]
        for prefix in ("Value error, ", "Assertion failed, "):
            message = message.removeprefix(prefix)
        if entry["type"] == "extra_forbidden":
            message = "unknown option (check the spelling against config.yml.example)"
        elif entry["type"] == "missing":
            message = "required (see config.yml.example)"
        lines.append(f"  {location or '<root>'}: {message}")
    plural = "s" if len(lines) != 1 else ""
    return f"{len(lines)} problem{plural} in the config file:\n" + "\n".join(lines)


def load_config(path: Path, *, check_paths: bool = True) -> Config:
    """Load and validate a config file.

    Raises:
        ConfigError: with a message that can be printed straight to the terminal.
    """
    path = Path(path)
    if path.is_dir():
        raise ConfigError(
            f"{path} is a directory. Pass the config file itself, e.g. {path / 'config.yml'}"
        )
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise ConfigError(f"Cannot read config file {path}: {exc.strerror}") from exc

    try:
        data = YAML(typ="safe").load(text)
    except YAMLError as exc:
        raise ConfigError(f"{path} is not valid YAML:\n  {exc}") from exc

    if data is None:
        raise ConfigError(f"{path} is empty")
    if not isinstance(data, dict):
        raise ConfigError(
            f"{path} must contain a mapping at the top level, got {type(data).__name__}"
        )

    context = {"config_dir": path.parent, "check_paths": check_paths}
    try:
        config = Config.model_validate(data, context=context)
    except ValidationError as exc:
        raise ConfigError(f"{path}\n{format_validation_error(exc)}") from exc

    if check_paths:
        for name in config.figures:
            folder = config.folder_for(name)
            if not folder.is_dir():
                _log.warning("Figure %r has no media folder at %s", name, folder)
    return config