eink_client.py
Rama main · réplica de lectura del repositorio original.
"""
E-ink device client (Track B) — the poll-and-sleep bridge that pulls the current image from
`GET /display/{id}/current.{ext}` and blits it to a Pimoroni Inky Impression 13.3" Spectra 6 panel.
Runs HOST-SIDE (not in the container) via `deploy/appliance/bin/sd-eink` — GPIO/SPI aren't reachable
from the non-root app container (ADR-037), and this client must also run standalone in "satellite" mode
with no local Pieria container at all. So this module is deliberately standalone: only stdlib +
Pillow + httpx + (lazily) `inky`. No `database`/app imports — see `.ai/spec_eink_spectra6.md` §3.2/3.4.
Design note (no hardware here): mirrors `frame_push.py`'s shape — an `EinkClient` ABC + a real
`InkyClient` + a `FakeInkyClient` + a testable single-tick (`push_once` / `run_tick`) + a loop
(`run_loop`). Same "test the platform without hardware" posture as the Frame TV integration
(memory `no-frame-tv-for-testing`), applied to e-ink.
"""
from __future__ import annotations
import hashlib
import io
import logging
import os
import re
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Optional, Tuple
import httpx
from PIL import Image
logger = logging.getLogger("sd-eink")
DEFAULT_HEARTBEAT_FILE = "/run/sd-eink.state"
# fetch_fn(url) -> (content_bytes, headers_dict). Injected so push_once never calls httpx directly.
#: First retry delay after a failed tick, in seconds; doubles per consecutive failure up to
#: EINK_MIN_INTERVAL. Small enough that a device which outran its own server recovers in seconds.
ERROR_BACKOFF_BASE = 15
FetchFn = Callable[[str], Tuple[bytes, dict]]
# schedule_fn() -> schedule-state JSON dict (has a "quiet" bool).
ScheduleFn = Callable[[], dict]
# --------------------------------------------------------------------------- config
def split_conf_value(rest: str):
"""Split the text after `KEY=` into (value, trailing), matching what bash `.`-sourcing yields for
the forms pieria.conf actually uses (ADR-137 conf comments). Only space/tab separate words, as in bash:
KEY=value # note -> value (an unquoted `#` starts a comment only when space/tab precedes it)
KEY="a # b" # note -> a # b (quotes protect a `#`)
KEY='a # b' -> a # b
KEY="a"b -> ab (text glued to a closing quote joins the word, so KEY="90"#x -> 90#x)
KEY= # note -> ""
`trailing` is the raw suffix after the value word (it starts with space/tab, or is empty), kept so a
writer can preserve a line's comment; writers keep it ONLY if it matches `^[ \t]+#`. An unterminated
quote is treated as an unquoted value. Copied verbatim into eink_client.py, sd-conf and sd_setup.py
(standalone scripts); a test pins that the source text is identical."""
body = rest.lstrip(" \t")
if len(body) != len(rest) and body.startswith("#"):
return "", rest
if body[:1] in ("'", '"'):
end = body.find(body[0], 1)
if end != -1:
glued = re.match(r"[^ \t]*", body[end + 1:]).group(0)
return body[1:end] + glued, body[end + 1 + len(glued):]
m = re.search(r"[ \t]+#", body)
cut = m.start() if m else len(body)
return body[:cut].strip(" \t"), body[cut:]
def parse_conf_text(text: str) -> dict:
"""KEY -> value for every assignment line in a pieria.conf, bash-sourcing semantics (last wins)."""
out = {}
for raw in (text or "").split("\n"): # bash breaks lines on \n ONLY (not \v \f \x85 ...)
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, rest = line.partition("=")
key = key.strip()
if key:
out[key] = split_conf_value(rest)[0]
return out
@dataclass
class EinkConfig:
server_url: str
display_id: str
min_interval: int = 900
saturation: float = 0.5
orientation: str = "" # "" = landscape (1600x1200) | "portrait" = rotate 90 (1200x1600)
w: int = 1600
h: int = 1200
palette: str = "spectra6"
fit: str = "cover"
@classmethod
def from_env(cls) -> EinkConfig:
return cls(
server_url=os.environ.get("SERVER_URL", "http://localhost:8000").rstrip("/"),
display_id=os.environ.get("DISPLAY_ID", "default"),
min_interval=_int_env("EINK_MIN_INTERVAL", 900),
orientation=os.environ.get("EINK_ORIENTATION", "").strip(),
)
@property
def render_size(self) -> tuple[int, int]:
"""The w,h to ASK the server for — swapped when the panel hangs portrait.
The panel's buffer is always landscape (w x h native); `orientation=portrait` means it is
physically rotated 90 degrees, so what the viewer sees is a h x w portrait canvas. The art must
therefore be COMPOSED at h x w and rotated back to the native buffer at paint time — not
composed landscape and rotated, which is what happened before this: the server framed for a
1600x1200 landscape window and the client handed the 1200x1600 result to a 1600x1200 panel,
so a portrait frame got both the wrong composition and a buffer the panel couldn't take.
"""
return (self.h, self.w) if self.orientation == "portrait" else (self.w, self.h)
@property
def pull_url(self) -> str:
w, h = self.render_size
return (
f"{self.server_url}/display/{self.display_id}/current.png"
f"?w={w}&h={h}&palette={self.palette}&fit={self.fit}"
)
@property
def schedule_url(self) -> str:
return f"{self.server_url}/api/displays/{self.display_id}/schedule-state"
def _int_env(key: str, fallback: int) -> int:
try:
return int(str(os.environ.get(key, fallback)).strip())
except (TypeError, ValueError):
return fallback
def _float_env(key: str, fallback: float) -> float:
try:
return float(str(os.environ.get(key, fallback)).strip())
except (TypeError, ValueError):
return fallback
# --------------------------------------------------------------------------- panel abstraction
class EinkClient(ABC):
"""Minimal interface over an e-ink panel, so the pusher is hardware-agnostic + testable."""
resolution: Optional[Tuple[int, int]] = None
@abstractmethod
def show(self, image: Image.Image, saturation: float) -> None:
"""Blit an image to the panel (blocking; a full refresh is ~20-35s)."""
class InkyClient(EinkClient):
"""Thin wrapper over the Pimoroni `inky` library. THE ONLY HARDWARE-DEPENDENT CODE.
`inky_factory` is injectable purely so tests can verify the mapping without the real library or a
panel; the default factory does the `inky` import INSIDE itself so importing this module on a
laptop with no `inky` installed never fails."""
def __init__(self, inky_factory=None):
self._inky_factory = inky_factory or self._default_factory
self._inky = None
@staticmethod
def _default_factory():
from inky.auto import auto # lazy: optional dep, only needed on real hardware
return auto()
def _panel(self):
if self._inky is None:
self._inky = self._inky_factory()
return self._inky
@property
def resolution(self):
return self._panel().resolution
def show(self, image: Image.Image, saturation: float) -> None:
panel = self._panel()
panel.set_image(image, saturation=saturation)
panel.show()
class FakeInkyClient(EinkClient):
"""In-memory EinkClient for tests: records calls, optionally raises to exercise error paths."""
def __init__(self, fail_on: Optional[str] = None, resolution: Tuple[int, int] = (1600, 1200)):
self.calls: list = []
self.shown: list = []
self.resolution = resolution
self._fail_on = fail_on
def show(self, image: Image.Image, saturation: float) -> None:
if self._fail_on == "show":
raise RuntimeError("simulated failure in show")
self.calls.append(("show", image.size, saturation))
self.shown.append((image.size, saturation))
# --------------------------------------------------------------------------- HTTP (real fetch/schedule)
def make_http_fetch_fn(timeout: float = 15.0) -> FetchFn:
"""Real FetchFn: GET the pull URL, return (body bytes, headers dict)."""
def _fetch(url: str) -> Tuple[bytes, dict]:
r = httpx.get(url, timeout=timeout)
r.raise_for_status()
return r.content, dict(r.headers)
return _fetch
def make_http_schedule_fn(cfg: EinkConfig, timeout: float = 8.0) -> ScheduleFn:
"""Real ScheduleFn: GET the display's schedule-state JSON."""
def _schedule() -> dict:
r = httpx.get(cfg.schedule_url, timeout=timeout)
r.raise_for_status()
return r.json()
return _schedule
# --------------------------------------------------------------------------- orchestration
def push_once(
cfg: EinkConfig,
fetch_fn: FetchFn,
client: EinkClient,
*,
last_etag: Optional[str] = None,
interval: Optional[int] = None,
) -> dict:
"""Pull the current image and paint it if it changed. Returns a status dict; never raises for
expected conditions (unchanged). A fetch or client.show error propagates to the caller
(run_tick handles it) — matches frame_push's contract.
`interval` (ADR-121) is this client's own actual pull cadence — the sleep it settled on after its
LAST tick, i.e. max(display_time, EINK_MIN_INTERVAL). Sent as `&interval=` so the server can widen
its KNOWN-display window to match a cadence slower than 2x display_time (a fixed EINK_MIN_INTERVAL
floor made a 30s playlist's panel disappear from /api/remote/displays for 14 of every 15 minutes —
the 2026-09-20 bench incident). run_tick supplies the previous tick's `sleep`; the very first tick
of a run has none yet, so it's omitted (the server's un-widened fallback covers that one request)."""
url = cfg.pull_url
if interval:
url = f"{url}&interval={int(interval)}"
content, headers = fetch_fn(url)
etag = headers.get("ETag") or headers.get("etag")
if not etag:
# Content-hash fallback so dedupe still works if the server ever omits the header.
etag = "sha256:" + hashlib.sha256(content).hexdigest()[:16]
if last_etag is not None and etag == last_etag:
return {"status": "unchanged", "etag": etag}
image = Image.open(io.BytesIO(content))
image.load()
if cfg.orientation == "portrait":
# The server composed this at (h, w) portrait; rotate it back onto the panel's native
# landscape buffer. Net effect: the panel gets its normal w x h, the viewer sees portrait art.
image = image.rotate(90, expand=True)
client.show(image, cfg.saturation)
refresh_after = headers.get("X-Refresh-After") or headers.get("x-refresh-after")
try:
refresh_after = int(refresh_after)
except (TypeError, ValueError):
refresh_after = cfg.min_interval
return {"status": "painted", "etag": etag, "refresh_after": refresh_after}
def _write_heartbeat() -> None:
path = os.environ.get("EINK_HEARTBEAT_FILE", DEFAULT_HEARTBEAT_FILE)
try:
with open(path, "w") as f:
f.write(str(time.time()))
except OSError:
pass # e.g. no /run perms on a laptop/test box — non-fatal, watchdog-friendly only
def run_tick(
cfg: EinkConfig,
fetch_fn: FetchFn,
client: EinkClient,
schedule_fn: ScheduleFn,
state: dict,
) -> dict:
"""A single scheduler tick (factored out so it's testable without the infinite loop). Never
raises — errors are caught and reported in the returned status dict."""
try:
sched = schedule_fn()
except Exception as e:
logger.warning(f"[eink] schedule-state fetch failed, proceeding to paint: {e}")
sched = {}
if sched.get("quiet"):
# E-ink holds its image at zero power — night/quiet means STOP refreshing, not blank.
return {"status": "quiet", "sleep": cfg.min_interval}
try:
res = push_once(cfg, fetch_fn, client, last_etag=state.get("last_etag"),
interval=state.get("last_interval"))
except Exception as e:
# Escalating backoff, starting SHORT. An all-in-one Pi boots its e-ink client seconds before its
# own Docker container binds the port, so the very first tick after a fresh setup reliably gets
# connection-refused. Sleeping the full steady-state interval on that one failure left a
# just-configured panel holding a stale image for 15 minutes — indistinguishable from a broken
# device to the person who just set it up (found on hardware 2026-07-21).
fails = state["error_streak"] = state.get("error_streak", 0) + 1
backoff = min(cfg.min_interval, ERROR_BACKOFF_BASE * (2 ** (fails - 1)))
logger.error(f"[eink] tick failed (attempt {fails}): {e} — retrying in {backoff}s")
return {"status": "error", "reason": str(e) or e.__class__.__name__,
"attempt": fails, "sleep": backoff}
state["error_streak"] = 0
res = dict(res)
if res["status"] == "painted":
state["last_etag"] = res["etag"]
_write_heartbeat()
res["sleep"] = max(res.get("refresh_after", cfg.min_interval), cfg.min_interval)
else: # unchanged
res["sleep"] = cfg.min_interval
# Remembered for the NEXT tick's `&interval=` (see push_once) — one tick behind, which converges
# immediately and matches steady state since a playlist's display_time rarely changes tick to tick.
state["last_interval"] = res["sleep"]
return res
def run_loop(
cfg: EinkConfig,
fetch_fn: Optional[FetchFn] = None,
client: Optional[EinkClient] = None,
schedule_fn: Optional[ScheduleFn] = None,
) -> None:
"""Periodic pull-and-paint loop. Builds real httpx fetch/schedule fns and an InkyClient unless
injected (tests / --dry-run inject fakes)."""
fetch_fn = fetch_fn or make_http_fetch_fn()
schedule_fn = schedule_fn or make_http_schedule_fn(cfg)
client = client or InkyClient()
logger.info(f"[eink] loop started (display={cfg.display_id}, server={cfg.server_url})")
state: dict = {}
while True:
res = run_tick(cfg, fetch_fn, client, schedule_fn, state)
if res.get("status") not in ("unchanged", "quiet"):
logger.info(f"[eink] tick: {res}")
time.sleep(max(1, res.get("sleep", cfg.min_interval)))