"""Decides what should be on screen right now, renders it, and only pushes to the panel when the rendered image actually changed. Design goals (in reaction to the InkyPi issues that motivated this rewrite): - One short tick (a few seconds), not a long sleep-until-interval -- config changes and manual actions take effect almost immediately, no signaling machinery needed. - Navidrome can preempt the rotation the instant something's playing, without needing to be "its turn" in a round-robin. - A screen is only ever fully re-rendered+pushed to hardware when its rendered image differs from what's currently displayed. """ import hashlib import logging import threading import time from pathlib import Path from PIL import Image logger = logging.getLogger(__name__) TICK_SECONDS = 5 PREVIEW_PATH = Path(__file__).resolve().parent / "web" / "static" / "preview.png" class ScreenCache: """Remembers a screen's last check() result so we don't re-poll faster than its own poll_interval, even if the rotation/tick cadence is tighter.""" def __init__(self): self.state = None self.checked_at: float = 0.0 def is_stale(self, poll_interval: int) -> bool: return (time.time() - self.checked_at) >= poll_interval class Scheduler: def __init__(self, config, screens: dict, display): self.config = config self.screens = screens self.display = display self._caches = {screen_id: ScreenCache() for screen_id in screens} self._current_screen_id: str | None = None self._rotation_index = 0 self._rotation_switched_at = time.time() self._last_image_hash: str | None = None self._manual_screen: str | None = None self._force_once = False self._wake = threading.Event() self._running = False self._thread: threading.Thread | None = None self._status_lock = threading.Lock() self._status = {"current_screen": None, "last_render_at": None, "last_display_write_at": None} # -------------------------------------------------------------- public def start(self): if self._thread and self._thread.is_alive(): return self._running = True self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() logger.info("Scheduler started") def stop(self): self._running = False self._wake.set() if self._thread: self._thread.join() def request_manual_refresh(self): self._force_once = True self._wake.set() def request_switch(self, screen_id: str): self._manual_screen = screen_id self._wake.set() def get_status(self) -> dict: with self._status_lock: return dict(self._status) # ------------------------------------------------------------- internal def _run(self): while self._running: try: self._tick() except Exception: logger.exception("Scheduler tick failed") self._wake.wait(timeout=TICK_SECONDS) self._wake.clear() def _tick(self): did_check = False if self._manual_screen: screen_id = self._manual_screen self._manual_screen = None state = self._check(screen_id) did_check = True else: screen_id, state, checked = self._select_screen() did_check = checked self._current_screen_id = screen_id screen = self.screens[screen_id] settings = self.config.get_screen_settings(screen_id) resolution = self.display.resolution image = screen.render(state, settings, resolution) image_hash = hashlib.sha256(image.tobytes()).hexdigest() self._save_preview(image) with self._status_lock: self._status["current_screen"] = screen_id self._status["last_render_at"] = time.time() if image_hash != self._last_image_hash or self._force_once: logger.info("Content changed (or forced) -- writing to display. screen=%s", screen_id) self.display.show(image) self.display.sleep() self._last_image_hash = image_hash self._force_once = False with self._status_lock: self._status["last_display_write_at"] = time.time() else: logger.debug("No change for screen=%s, skipping display write", screen_id) if did_check: self.config.save() def _select_screen(self) -> tuple[str, object, bool]: """Returns (screen_id, state, did_check_happen).""" if self.config.get_navidrome_priority() and "navidrome" in self.screens: cache = self._caches["navidrome"] navidrome = self.screens["navidrome"] checked = False if cache.is_stale(navidrome.poll_interval): cache.state = self._check("navidrome") cache.checked_at = time.time() checked = True if cache.state is not None: return "navidrome", cache.state, checked return self._rotation_screen() def _rotation_screen(self) -> tuple[str, object, bool]: rotation = self.config.get_rotation() screen_ids = [s for s in rotation["screens"] if s in self.screens] if not screen_ids: # Nothing configured to rotate through -- fall back to whatever # exists so we're never stuck showing nothing. screen_ids = list(self.screens.keys()) now = time.time() # With rotation disabled, just pin to whatever's currently selected -- # never auto-advance. (Manual "show this screen now" still works via # request_switch(); it just won't stick past the next tick.) if rotation.get("enabled", True): interval_seconds = rotation["interval_minutes"] * 60 if now - self._rotation_switched_at >= interval_seconds: self._rotation_index = (self._rotation_index + 1) % len(screen_ids) self._rotation_switched_at = now screen_id = screen_ids[self._rotation_index % len(screen_ids)] cache = self._caches[screen_id] screen = self.screens[screen_id] checked = False if cache.is_stale(screen.poll_interval): cache.state = self._check(screen_id) cache.checked_at = time.time() checked = True return screen_id, cache.state, checked def _check(self, screen_id: str): screen = self.screens[screen_id] settings = self.config.get_screen_settings(screen_id) try: return screen.check(settings) except Exception: logger.exception("check() failed for screen=%s", screen_id) return None def _save_preview(self, image: Image.Image): try: PREVIEW_PATH.parent.mkdir(parents=True, exist_ok=True) image.save(PREVIEW_PATH) except OSError: logger.exception("Could not write preview image")