Honey
889c062212
Screen.poll_interval was a fixed class attribute the scheduler read directly -- added Screen.get_poll_interval(settings) so a screen can make it settings-driven instead. WeatherScreen overrides it to read a new "update_minutes" field (default 30, matching the old hardcoded value). New number input on the Weather settings page. Caught a real edge case while testing: `raw or 30` treats an explicit 0 as falsy and silently replaces it with 30 instead of clamping to the 1-minute floor. Not reachable via the UI (min="1" on the input) but the method should still be correct standalone -- fixed with an explicit None-check instead.
206 lines
7.6 KiB
Python
206 lines
7.6 KiB
Python
"""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._rotation_index = 0
|
|
self._rotation_switched_at = time.time()
|
|
self._last_image_hash: 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):
|
|
"""Pin a screen indefinitely (persisted in config, turns rotation
|
|
off). Navidrome priority can still preempt it if something starts
|
|
playing -- that's the one thing this app promises to always show
|
|
reliably -- but rotation won't, until it's explicitly re-enabled."""
|
|
self.config.pin_screen(screen_id)
|
|
self._wake.set()
|
|
|
|
def request_resume_rotation(self):
|
|
self.config.resume_rotation()
|
|
self._wake.set()
|
|
|
|
def get_status(self) -> dict:
|
|
with self._status_lock:
|
|
status = dict(self._status)
|
|
rotation = self.config.get_rotation()
|
|
pinned = self.config.get_pinned_screen()
|
|
status["pinned_screen"] = pinned if (pinned and not rotation.get("enabled", True)) else None
|
|
return 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):
|
|
screen_id, state, did_check = self._select_screen()
|
|
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).
|
|
|
|
Priority order: Navidrome-when-playing (if enabled) always wins --
|
|
that's the one thing this app promises to show reliably. Then a
|
|
manual "show this now" pin, if rotation is off because of one (set
|
|
together in Config.pin_screen()). Then normal rotation.
|
|
"""
|
|
if self.config.get_navidrome_priority() and "navidrome" in self.screens:
|
|
screen_id, state, checked = self._poll_screen("navidrome")
|
|
if state is not None:
|
|
return screen_id, state, checked
|
|
|
|
rotation = self.config.get_rotation()
|
|
if not rotation.get("enabled", True):
|
|
pinned = self.config.get_pinned_screen()
|
|
if pinned and pinned in self.screens:
|
|
return self._poll_screen(pinned)
|
|
|
|
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.
|
|
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)]
|
|
return self._poll_screen(screen_id)
|
|
|
|
def _poll_screen(self, screen_id: str) -> tuple[str, object, bool]:
|
|
cache = self._caches[screen_id]
|
|
screen = self.screens[screen_id]
|
|
settings = self.config.get_screen_settings(screen_id)
|
|
checked = False
|
|
if cache.is_stale(screen.get_poll_interval(settings)):
|
|
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")
|