frame_push.py
Rama main · réplica de lectura del repositorio original.
"""
Samsung Frame TV push output (Track: Integrations).
Renders the currently-selected artwork to a full-colour TV-resolution JPEG and pushes it into a
Samsung Frame TV's Art Mode over the LAN (Samsung's unofficial art WebSocket API via `samsungtvws`).
This is a third output target alongside the browser Canvas and the e-ink pull API.
Design note (no hardware here): the TV is hidden behind the small `FrameClient` interface so the only
hardware-dependent code is the thin `SamsungFrameClient` wrapper. Everything else — config, render,
push orchestration (dedupe / delete-old / persist), and the scheduler tick — is exercised in tests via
`FakeFrameClient` and an injected selector. See memory `no-frame-tv-for-testing`.
"""
from __future__ import annotations
import asyncio
import logging
import time
import traceback
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Awaitable, Callable, Optional, Tuple
from database import SessionLocal
from epaper import pick_crop_for_aspect, render_fullcolor
from models import SettingsModel
logger = logging.getLogger("artwork-display-api.frame")
# Token persists the TV pairing (first connect triggers a TV-side authorize prompt) across restarts.
# ./data is the app's persisted volume (same place as the SQLite DB).
DEFAULT_TOKEN_FILE = Path("data") / "frame_tv_token.json"
FRAME_SETTING_KEYS = (
"frame_enabled", "frame_host", "frame_port", "frame_playlist",
"frame_interval_sec", "frame_width", "frame_height", "frame_matte", "frame_display_id",
# internal state written by the pusher:
"frame_last_content_id", "frame_last_artwork_id", "frame_last_push_at",
)
_DEFAULTS = {
"frame_enabled": "false",
"frame_host": "",
"frame_port": "8001",
"frame_playlist": "",
"frame_interval_sec": "900",
"frame_width": "3840",
"frame_height": "2160",
"frame_matte": "none",
"frame_display_id": "frame-tv",
}
# select_fn(playlist_name) -> (image_path, artwork_id, focal[, aspect_crops]) or None, where focal is the normalized
# (x, y) framing anchor. Injected by app.py so this module never imports app (avoids a circular
# import) and stays unit-testable.
SelectFn = Callable[[str], Awaitable[Optional[Tuple[Path, int, Tuple[float, float]]]]]
# --------------------------------------------------------------------------- config
_cache: dict = {"data": None, "ts": 0.0}
_CACHE_TTL = 30.0
def _read_rows() -> dict:
db = SessionLocal()
try:
return {
s.setting_key: s.setting_value
for s in db.query(SettingsModel).filter(SettingsModel.setting_key.in_(FRAME_SETTING_KEYS)).all()
}
finally:
db.close()
def _truthy(v: str) -> bool:
return str(v).strip().lower() in ("true", "1", "yes", "on")
def _int(v, fallback: int) -> int:
try:
return int(str(v).strip())
except (TypeError, ValueError):
return fallback
def get_frame_config(force: bool = False) -> dict:
"""Resolve Frame settings (DB rows over defaults), 30s TTL cache like ai_client.get_ai_config."""
now = time.monotonic()
if not force and _cache["data"] is not None and (now - _cache["ts"]) < _CACHE_TTL:
return _cache["data"]
rows = _read_rows()
def val(k):
return rows.get(k, _DEFAULTS.get(k, ""))
cfg = {
"enabled": _truthy(val("frame_enabled")),
"host": val("frame_host").strip(),
"port": _int(val("frame_port"), 8001),
"playlist": val("frame_playlist").strip(),
"interval_sec": max(60, _int(val("frame_interval_sec"), 900)),
"width": _int(val("frame_width"), 3840),
"height": _int(val("frame_height"), 2160),
"matte": (val("frame_matte").strip() or "none"),
"display_id": (val("frame_display_id").strip() or "frame-tv"),
"last_content_id": rows.get("frame_last_content_id") or None,
"last_artwork_id": _int(rows.get("frame_last_artwork_id"), 0) or None,
"last_push_at": float(rows.get("frame_last_push_at") or 0.0) or None,
}
_cache["data"] = cfg
_cache["ts"] = now
return cfg
def invalidate_frame_cache() -> None:
_cache["data"] = None
_cache["ts"] = 0.0
def _persist_state(content_id: str, artwork_id: int) -> None:
"""Record what we last pushed so the next cycle can dedupe + delete the prior upload."""
db = SessionLocal()
try:
updates = {
"frame_last_content_id": content_id,
"frame_last_artwork_id": str(artwork_id),
"frame_last_push_at": repr(time.time()),
}
for k, v in updates.items():
row = db.query(SettingsModel).filter(SettingsModel.setting_key == k).first()
if row:
row.setting_value = v
else:
db.add(SettingsModel(setting_key=k, setting_value=v))
db.commit()
finally:
db.close()
invalidate_frame_cache()
# --------------------------------------------------------------------------- TV abstraction
class FrameClient(ABC):
"""Minimal interface over a Frame TV's Art Mode, so the pusher is hardware-agnostic + testable."""
@abstractmethod
async def push(self, image: bytes, file_type: str, matte: str) -> str:
"""Upload an image; return its content_id."""
@abstractmethod
async def show(self, content_id: str) -> None:
"""Select the uploaded image as the current art."""
@abstractmethod
async def ensure_artmode(self) -> None:
"""Make sure the TV is in Art Mode."""
@abstractmethod
async def delete(self, content_id: str) -> None:
"""Remove a previously-uploaded image (housekeeping)."""
@abstractmethod
async def test(self) -> dict:
"""Light connectivity check; raises on failure."""
class SamsungFrameClient(FrameClient):
"""Thin wrapper over the synchronous `samsungtvws` library, called via asyncio.to_thread.
THE ONLY HARDWARE-DEPENDENT CODE. One connection is reused across the calls within a single push
cycle; the scheduler builds a fresh client each tick (see _run_one_cycle), so there is no
long-lived socket to reset. `tv_factory` is injectable purely so tests can verify the mapping
without the real library or a TV."""
def __init__(self, host: str, port: int = 8001, token_file: Optional[str] = None, tv_factory=None):
self.host = host
self.port = port
self.token_file = token_file or str(DEFAULT_TOKEN_FILE)
self._tv_factory = tv_factory or self._default_factory
self._tv = None
def _default_factory(self):
from samsungtvws import SamsungTVWS # lazy: optional dep, only needed when enabled
DEFAULT_TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
return SamsungTVWS(host=self.host, port=self.port, token_file=self.token_file)
def _conn(self):
if self._tv is None:
self._tv = self._tv_factory()
return self._tv
async def push(self, image: bytes, file_type: str, matte: str) -> str:
return await asyncio.to_thread(
lambda: self._conn().art().upload(image, file_type=file_type, matte=matte)
)
async def show(self, content_id: str) -> None:
await asyncio.to_thread(lambda: self._conn().art().select_image(content_id, show=True))
async def ensure_artmode(self) -> None:
await asyncio.to_thread(lambda: self._conn().art().set_artmode(True))
async def delete(self, content_id: str) -> None:
await asyncio.to_thread(lambda: self._conn().art().delete(content_id))
async def test(self) -> dict:
return await asyncio.to_thread(lambda: {"artmode": self._conn().art().get_artmode()})
class FakeFrameClient(FrameClient):
"""In-memory FrameClient for tests: records calls, returns deterministic content_ids."""
def __init__(self, fail_on: Optional[str] = None):
self.calls: list = []
self.uploaded: list = []
self.deleted: list = []
self.artmode = False
self._n = 0
self._fail_on = fail_on # method name to raise in, to exercise error paths
def _maybe_fail(self, name):
if self._fail_on == name:
raise RuntimeError(f"simulated failure in {name}")
async def push(self, image: bytes, file_type: str, matte: str) -> str:
self._maybe_fail("push")
self._n += 1
cid = f"MY-F{self._n:04d}"
self.calls.append(("push", file_type, matte, len(image)))
self.uploaded.append(cid)
return cid
async def show(self, content_id: str) -> None:
self._maybe_fail("show")
self.calls.append(("show", content_id))
async def ensure_artmode(self) -> None:
self._maybe_fail("artmode")
self.artmode = True
self.calls.append(("artmode",))
async def delete(self, content_id: str) -> None:
self._maybe_fail("delete")
self.deleted.append(content_id)
self.calls.append(("delete", content_id))
async def test(self) -> dict:
self._maybe_fail("test")
return {"ok": True}
def _default_client_factory(cfg: dict) -> FrameClient:
return SamsungFrameClient(cfg["host"], cfg["port"], token_file=str(DEFAULT_TOKEN_FILE))
# --------------------------------------------------------------------------- orchestration
async def push_once(
cfg: dict,
select_fn: SelectFn,
client: FrameClient,
*,
force: bool = False,
persist: Optional[Callable[[str, int], None]] = None,
) -> dict:
"""Render + push the current artwork to the Frame. Returns a status dict; never raises for
expected conditions (no artwork, unchanged). Caller decides how to surface client errors."""
sel = await select_fn(cfg.get("playlist", ""))
if not sel:
return {"status": "skipped", "reason": "no artwork available"}
# 4th element (per-shape crop presets) is optional: a selector that predates them — or any
# third-party one — still returns a 3-tuple and simply gets the focal cover.
path, artwork_id, focal = sel[0], sel[1], sel[2]
aspect_crops = sel[3] if len(sel) > 3 else None
if not force and artwork_id == cfg.get("last_artwork_id") and cfg.get("last_content_id"):
return {"status": "unchanged", "artwork_id": artwork_id}
# A Frame can't Ken Burns its way out of a bad crop any more than e-ink can, so it takes the
# authored box too — picked for the Frame's own panel size.
crop_box = pick_crop_for_aspect(aspect_crops, cfg["width"], cfg["height"])
image = await asyncio.to_thread(
render_fullcolor, Path(path), cfg["width"], cfg["height"], "cover", focal, 90, crop_box
)
content_id = await client.push(image, "jpg", cfg.get("matte", "none"))
await client.show(content_id)
await client.ensure_artmode()
old = cfg.get("last_content_id")
if old and old != content_id:
try:
await client.delete(old)
except Exception as e: # housekeeping only — don't fail the push
logger.warning(f"[Frame] could not delete prior upload {old}: {e}")
(persist or _persist_state)(content_id, artwork_id)
return {"status": "pushed", "content_id": content_id, "artwork_id": artwork_id, "bytes": len(image)}
async def run_test_push(select_fn: SelectFn, client_factory=_default_client_factory) -> dict:
"""One-shot 'Test / Push now' for the settings endpoint. Returns a structured result instead of
raising, so the GUI can show a clean message with or without a TV present."""
cfg = get_frame_config(force=True)
if not cfg["host"]:
return {"status": "error", "reason": "No Frame TV host/IP configured."}
try:
client = client_factory(cfg)
return await push_once(cfg, select_fn, client, force=True)
except Exception as e:
logger.warning(f"[Frame] test push failed: {e}")
return {"status": "error", "reason": str(e) or e.__class__.__name__}
async def _run_one_cycle(select_fn: SelectFn, client_factory=_default_client_factory) -> dict:
"""A single scheduler tick (factored out so it's testable without the infinite loop)."""
cfg = get_frame_config(force=True)
if not cfg["enabled"] or not cfg["host"]:
return {"status": "idle"}
client = client_factory(cfg)
return await push_once(cfg, select_fn, client)
async def frame_push_loop(select_fn: SelectFn, client_factory=_default_client_factory) -> None:
"""Periodic pusher. Started once (leader worker only) from app.py's lifespan."""
logger.info("[Frame] push loop started")
while True:
delay = 900
try:
delay = get_frame_config(force=True)["interval_sec"]
res = await _run_one_cycle(select_fn, client_factory)
if res.get("status") not in ("idle", "unchanged"):
logger.info(f"[Frame] cycle: {res}")
except Exception:
logger.error(f"[Frame] loop error: {traceback.format_exc()}")
delay = max(60, delay)
await asyncio.sleep(delay)