"""Refresh partiel e-ink : calcule, par client, la zone du dashboard qui a changé depuis l'image précédemment servie, et décide d'un refresh partiel (crop) ou complet (full). Le diff se fait ici (côté serveur, PIL dispo) ; la Kobo ne reçoit qu'un PNG prêt à afficher (crop en partial, image pleine en full) + son offset, via les endpoints /frame.meta et /frame.png. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field from PIL import Image, ImageChops import render from config import config # Seuil de diff (0-255) sous lequel on considère deux pixels identiques. Le rendu PNG est # lossless et déterministe : les zones inchangées sont strictement identiques, ce seuil ne # sert qu'à absorber un éventuel bruit résiduel. _DIFF_THRESHOLD = 16 @dataclass class _ClientState: prev_image: Image.Image | None = None since_full: int = 0 # cycles écoulés depuis le dernier full refresh seq: int = 0 # compteur monotone, identifie le frame courant png: bytes = b"" # PNG à servir (crop ou image pleine) mode: str = "full" region: tuple[int, int, int, int] = (0, 0, 0, 0) # x, y, w, h lock: asyncio.Lock = field(default_factory=asyncio.Lock) _clients: dict[str, _ClientState] = {} def _changed_bbox(prev: Image.Image, cur: Image.Image) -> tuple[int, int, int, int] | None: """Rectangle (left, upper, right, lower) englobant les pixels modifiés, ou None si identiques.""" diff = ImageChops.difference(prev, cur) if _DIFF_THRESHOLD: diff = diff.point(lambda p: 255 if p > _DIFF_THRESHOLD else 0) return diff.getbbox() async def compute_frame(client: str, reset: bool = False) -> dict: """Rend le dashboard, calcule le diff vs l'image précédente du client, met à jour son état et renvoie {mode, x, y, w, h, seq}. Le PNG correspondant est stocké pour /frame.png. reset=True (envoyé par la Kobo au 1er cycle après un (re)démarrage) oublie l'image précédente : l'écran a été effacé par le reboot, un diff partiel se poserait sur une base erronée. On force alors un full refresh propre.""" state = _clients.setdefault(client, _ClientState()) async with state.lock: if reset: state.prev_image = None # -> force_full ci-dessous cur = await render.render_image() full_w, full_h = cur.size state.seq += 1 force_full = ( state.prev_image is None or state.since_full + 1 >= config.full_refresh_every ) bbox = None if force_full else _changed_bbox(state.prev_image, cur) if not force_full and bbox is not None: left, upper, right, lower = bbox w, h = right - left, lower - upper ratio = (w * h) / float(full_w * full_h) if ratio > config.partial_max_ratio: force_full = True if force_full: state.png = render.encode_png(cur) state.mode = "full" state.region = (0, 0, full_w, full_h) state.since_full = 0 elif bbox is None: # Aucune différence : on ne rafraîchit rien (la Kobo ignore ce cycle). On conserve # le PNG précédent et on incrémente quand même since_full pour garder la cadence du # full refresh horaire (basée sur le temps écoulé). state.mode = "noop" state.region = (0, 0, 0, 0) state.since_full += 1 else: left, upper, right, lower = bbox state.png = render.encode_png(cur.crop(bbox)) state.mode = "partial" state.region = (left, upper, right - left, lower - upper) state.since_full += 1 state.prev_image = cur x, y, w, h = state.region return {"mode": state.mode, "x": x, "y": y, "w": w, "h": h, "seq": state.seq} def get_png(client: str) -> bytes | None: """PNG stocké lors du dernier compute_frame() pour ce client (None si jamais calculé).""" state = _clients.get(client) return state.png if state and state.png else None