Paperwall/config.py

"""Single JSON config file, loaded once and kept in memory.

Unlike InkyPi's Config (which we spent a long time debugging), writes happen
immediately on every change from the web UI, and the scheduler re-reads the
in-memory structure fresh on every tick -- there's no "did the setting
actually take effect" gap, because there's no long sleep to get stuck in.
"""

import json
import logging
import threading
from pathlib import Path

logger = logging.getLogger(__name__)

DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent / "config.json"

DEFAULT_ROTATION = {
    "screens": ["clock", "weather"],
    "interval_minutes": 5,
    "enabled": True,
}


class Config:
    def __init__(self, screens: dict, path: Path = DEFAULT_CONFIG_PATH):
        self.path = path
        self._lock = threading.Lock()
        self._data = self._load(screens)

    def _load(self, screens: dict) -> dict:
        if self.path.exists():
            with open(self.path) as f:
                data = json.load(f)
        else:
            data = {}

        data.setdefault("rotation", dict(DEFAULT_ROTATION))
        # Migrate configs written before rotation had minutes/enabled.
        if "interval_seconds" in data["rotation"] and "interval_minutes" not in data["rotation"]:
            data["rotation"]["interval_minutes"] = max(1, round(data["rotation"].pop("interval_seconds") / 60))
        data["rotation"].setdefault("interval_minutes", DEFAULT_ROTATION["interval_minutes"])
        data["rotation"].setdefault("enabled", DEFAULT_ROTATION["enabled"])
        data.setdefault("navidrome_priority", True)
        data.setdefault("pinned_screen", None)
        data.setdefault("schedules", [])
        data.setdefault("settings", {})
        for screen_id, screen in screens.items():
            data["settings"].setdefault(screen_id, {})
            for key, value in screen.default_settings().items():
                data["settings"][screen_id].setdefault(key, value)

        return data

    def save(self) -> None:
        with self._lock:
            tmp_path = self.path.with_suffix(".json.tmp")
            with open(tmp_path, "w") as f:
                json.dump(self._data, f, indent=2)
            tmp_path.replace(self.path)
            logger.debug("Config written to %s", self.path)

    # ------------------------------------------------------------ screens

    def get_screen_settings(self, screen_id: str) -> dict:
        """Returns the live, mutable settings dict for a screen -- screens are
        allowed to write ephemeral state into it (e.g. Navidrome's last-known
        track for the idle fallback)."""
        with self._lock:
            return self._data["settings"].setdefault(screen_id, {})

    def update_screen_settings(self, screen_id: str, updates: dict) -> None:
        with self._lock:
            self._data["settings"].setdefault(screen_id, {}).update(updates)
        self.save()

    # ----------------------------------------------------------- rotation

    def get_rotation(self) -> dict:
        with self._lock:
            return dict(self._data["rotation"])

    def set_rotation(self, screens: list[str], interval_minutes: int, enabled: bool) -> None:
        with self._lock:
            self._data["rotation"] = {"screens": screens, "interval_minutes": interval_minutes, "enabled": enabled}
            if enabled:
                # Re-enabling rotation (e.g. via the settings checkbox)
                # supersedes any manual pin -- that's the intended way to
                # get back to automatic switching.
                self._data["pinned_screen"] = None
        self.save()

    def get_pinned_screen(self) -> str | None:
        with self._lock:
            return self._data.get("pinned_screen")

    def pin_screen(self, screen_id: str) -> None:
        """Pin a screen indefinitely -- turns rotation off so it doesn't
        fight with the pin. Stays until the pin is explicitly cleared or
        rotation is re-enabled (settings checkbox, or resume_rotation())."""
        with self._lock:
            self._data["pinned_screen"] = screen_id
            self._data["rotation"]["enabled"] = False
        self.save()

    def resume_rotation(self) -> None:
        with self._lock:
            self._data["rotation"]["enabled"] = True
            self._data["pinned_screen"] = None
        self.save()

    # ----------------------------------------------------------- schedules

    def get_schedules(self) -> list[dict]:
        """Each rule: {"screen_id", "start", "end"} with start/end as
        "HH:MM" local time. A rule being in this list *is* it being
        enabled -- there's no separate flag, disabling one is just removing
        it (the settings form does this by leaving its screen dropdown on
        "-- none --")."""
        with self._lock:
            return list(self._data["schedules"])

    def set_schedules(self, schedules: list[dict]) -> None:
        with self._lock:
            self._data["schedules"] = schedules
        self.save()

    def get_navidrome_priority(self) -> bool:
        with self._lock:
            return bool(self._data["navidrome_priority"])

    def set_navidrome_priority(self, enabled: bool) -> None:
        with self._lock:
            self._data["navidrome_priority"] = enabled
        self.save()