[FEAT] frame: e-ink wall display mirroring the live collage
This commit is contained in:
Teddy Warner
@@ -0,0 +1,362 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Frame-Pi client: turn a collage screenshot into Inky panel pixels.
|
||||
|
||||
Runs on the Pi Zero W on a systemd timer. Each run it decides whether a
|
||||
refresh is worth it (the species set or call-count brackets changed, and it
|
||||
is not quiet hours), then crops the title and collage from the screenshot,
|
||||
centres and mats them, and pushes the result to the Inky Impression 13.3".
|
||||
``--preview out.png`` writes an approximate 6-ink dither instead, so the
|
||||
look can be checked on any machine without the panel.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import hashlib
|
||||
import inspect
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import statistics
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
|
||||
from PIL import Image, ImageChops, ImageDraw
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ModuleNotFoundError: # Python < 3.11
|
||||
import tomli as tomllib
|
||||
|
||||
PANEL_W, PANEL_H = 1200, 1600 # portrait; the panel itself is 1600x1200
|
||||
|
||||
# Approximate Spectra-6 inks, used only for --preview. On hardware the Inky
|
||||
# library maps to the panel's real palette.
|
||||
SPECTRA6 = [(236, 234, 223), (26, 26, 28), (165, 60, 56),
|
||||
(198, 176, 74), (49, 71, 130), (58, 110, 72)]
|
||||
|
||||
DEFAULTS = {
|
||||
"base_url": "http://birdnet.local",
|
||||
"hours": 24,
|
||||
"image": "", # local PNG written by the shooter
|
||||
"image_url": "", # or a published screenshot URL
|
||||
"shoot": False, # or capture inline (needs a browser; not a Zero W)
|
||||
"shoot_title": None, "shoot_subtitle": None,
|
||||
"shoot_headline_px": 42, "shoot_eyebrow_px": 18, "shoot_lowercase": False,
|
||||
"shoot_mat": 0.04, "shoot_small_floor": 0.04,
|
||||
"mat": 0.0, # extra global shrink of the content inside the A5 opening
|
||||
"rotate": 90, # 90 or 270 if the frame hangs the other way up
|
||||
"saturation": 0.6,
|
||||
"panel": "", # "el133uf1" forces the 13.3" driver if auto() fails
|
||||
"quiet_start": 0, "quiet_end": 0, # 0/0 = no quiet hours
|
||||
"heal_hours": 24,
|
||||
"state": "~/.birdframe/state.json",
|
||||
"cache": "~/.birdframe",
|
||||
"timeout": 45,
|
||||
"basic_user": None, "basic_pass": None,
|
||||
}
|
||||
|
||||
|
||||
def _auth(cfg):
|
||||
if not cfg.get("basic_user"):
|
||||
return None
|
||||
raw = f"{cfg['basic_user']}:{cfg.get('basic_pass') or ''}".encode()
|
||||
return "Basic " + base64.b64encode(raw).decode()
|
||||
|
||||
|
||||
# --- change detection -------------------------------------------------------
|
||||
def slugify(sci):
|
||||
return re.sub(r"[^a-z0-9]+", "-", sci.lower()).strip("-")
|
||||
|
||||
|
||||
def _bucket(n):
|
||||
for i, edge in enumerate((1, 2, 5, 15, 40, 100, 300, 1000)):
|
||||
if n <= edge:
|
||||
return i
|
||||
return 8
|
||||
|
||||
|
||||
def fetch_recent(base, hours, timeout, auth=None):
|
||||
url = f"{base.rstrip('/')}/avian/api/birdnet-api.php?action=recent&hours={hours}"
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "AvianVisitors-frame/1.0"})
|
||||
if auth:
|
||||
req.add_header("Authorization", auth)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return json.loads(r.read(2_000_000)).get("species", [])
|
||||
|
||||
|
||||
def signature(species):
|
||||
items = sorted((slugify(s["sci"]), _bucket(int(s.get("n") or 1))) for s in species)
|
||||
return hashlib.sha256(json.dumps(items).encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
# --- image ------------------------------------------------------------------
|
||||
def get_image(src, timeout, auth=None):
|
||||
if re.match(r"^https?://", src):
|
||||
req = urllib.request.Request(src, headers={"User-Agent": "AvianVisitors-frame/1.0"})
|
||||
if auth:
|
||||
req.add_header("Authorization", auth)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||||
return Image.open(io.BytesIO(r.read(20_000_000))).convert("RGB")
|
||||
return Image.open(os.path.expanduser(src)).convert("RGB")
|
||||
|
||||
|
||||
def fit_panel(img):
|
||||
if img.size != (PANEL_W, PANEL_H):
|
||||
img = img.resize((PANEL_W, PANEL_H), Image.LANCZOS)
|
||||
return img
|
||||
|
||||
|
||||
def _paper(img):
|
||||
"""Median of the four corners, robust to a stray inked corner."""
|
||||
w, h = img.size
|
||||
px = (img.getpixel(p) for p in ((4, 4), (w - 5, 4), (4, h - 5), (w - 5, h - 5)))
|
||||
return tuple(int(statistics.median(c)) for c in zip(*px))
|
||||
|
||||
|
||||
# The mat opening is an A5 rectangle (1 : sqrt(2)) centred in the panel; the
|
||||
# content floats inside it with `mat` of inner whitespace.
|
||||
A5_H = PANEL_H * 0.7071 # A5 is 1/sqrt(2) of the panel height
|
||||
A5_W = A5_H / 1.41421 # A5 aspect 1 : sqrt(2)
|
||||
|
||||
|
||||
def _place(content, paper, mat):
|
||||
s = min(A5_W * (1 - mat) / content.width, A5_H * (1 - mat) / content.height)
|
||||
nw, nh = max(1, round(content.width * s)), max(1, round(content.height * s))
|
||||
content = content.resize((nw, nh), Image.LANCZOS)
|
||||
canvas = Image.new("RGB", (PANEL_W, PANEL_H), paper)
|
||||
canvas.paste(content, ((PANEL_W - nw) // 2, (PANEL_H - nh) // 2))
|
||||
return canvas
|
||||
|
||||
|
||||
def _region_bbox(img, paper, y0, y1):
|
||||
region = img.crop((0, y0, img.width, y1))
|
||||
diff = ImageChops.difference(region, Image.new("RGB", region.size, paper))
|
||||
bb = diff.convert("L").point(lambda p: 255 if p > 34 else 0).getbbox()
|
||||
return None if not bb else (bb[0], y0 + bb[1], bb[2], y0 + bb[3])
|
||||
|
||||
|
||||
def _scale_w(img, target_w):
|
||||
s = target_w / img.width
|
||||
return img.resize((max(1, round(img.width * s)), max(1, round(img.height * s))), Image.LANCZOS)
|
||||
|
||||
|
||||
def _centroid_x(img, paper):
|
||||
"""Horizontal centre of ink weight (what the eye reads as centred)."""
|
||||
m = ImageChops.difference(img, Image.new("RGB", img.size, paper)).convert("L")
|
||||
cols = list(m.resize((img.width, 1), Image.BOX).tobytes())
|
||||
total = sum(cols) or 1
|
||||
return sum(x * v for x, v in enumerate(cols)) / total
|
||||
|
||||
|
||||
# Content layout inside the A5 opening: the title and collage are sized
|
||||
# independently (as fractions of the opening width), so tuning one leaves the
|
||||
# other untouched. gap is a fraction of the opening height.
|
||||
TITLE_FRAC, COLLAGE_FRAC, GAP_FRAC = 0.55, 0.66, 0.1
|
||||
|
||||
|
||||
def mat_and_center(img, mat):
|
||||
"""Crop the title and collage, size each to a fraction of the A5 opening,
|
||||
stack with a gap, and centre on the panel."""
|
||||
img = img.convert("RGB")
|
||||
paper = _paper(img)
|
||||
mask = ImageChops.difference(img, Image.new("RGB", img.size, paper))
|
||||
mask = mask.convert("L").point(lambda p: 255 if p > 34 else 0)
|
||||
full = mask.getbbox()
|
||||
if not full:
|
||||
return img
|
||||
levels = list(mask.resize((1, img.height), Image.BOX).tobytes()) # per-row content
|
||||
top, bot = full[1], full[3]
|
||||
split, run = None, 0
|
||||
for y in range(top, bot):
|
||||
if levels[y] <= 2:
|
||||
run += 1
|
||||
if run >= 30: # first empty band of 30px splits title from collage
|
||||
cy = y
|
||||
while cy < bot and levels[cy] <= 2:
|
||||
cy += 1
|
||||
split = (y - run + 1, cy)
|
||||
break
|
||||
else:
|
||||
run = 0
|
||||
tb = _region_bbox(img, paper, top, split[0]) if split else None
|
||||
cb = _region_bbox(img, paper, split[1], bot + 1) if split else None
|
||||
box_w, box_h = A5_W * (1 - mat), A5_H * (1 - mat)
|
||||
if not (tb and cb):
|
||||
return _place(img.crop(full), paper, mat)
|
||||
title = _scale_w(img.crop(tb), box_w * TITLE_FRAC)
|
||||
collage = _scale_w(img.crop(cb), box_w * COLLAGE_FRAC)
|
||||
gap = round(box_h * GAP_FRAC)
|
||||
ccx = _centroid_x(collage, paper) # centre the collage by ink weight, not bbox
|
||||
half = max(ccx, collage.width - ccx)
|
||||
cw = round(max(title.width, 2 * half))
|
||||
comp = Image.new("RGB", (cw, title.height + gap + collage.height), paper)
|
||||
comp.paste(title, ((cw - title.width) // 2, 0))
|
||||
comp.paste(collage, (round(cw / 2 - ccx), title.height + gap))
|
||||
fit = min(box_w / comp.width, box_h / comp.height, 1.0) # shrink to the opening, never enlarge
|
||||
if fit < 1.0:
|
||||
comp = comp.resize((max(1, round(comp.width * fit)), max(1, round(comp.height * fit))), Image.LANCZOS)
|
||||
canvas = Image.new("RGB", (PANEL_W, PANEL_H), paper)
|
||||
canvas.paste(comp, ((PANEL_W - comp.width) // 2, (PANEL_H - comp.height) // 2))
|
||||
return canvas
|
||||
|
||||
|
||||
def quantize_spectra6(img):
|
||||
pal = Image.new("P", (1, 1))
|
||||
flat = [c for ink in SPECTRA6 for c in ink]
|
||||
flat += list(SPECTRA6[0]) * ((768 - len(flat)) // 3) # pad the 256-entry palette with paper
|
||||
pal.putpalette(flat[:768])
|
||||
return img.convert("RGB").quantize(palette=pal, dither=Image.Dither.FLOYDSTEINBERG).convert("RGB")
|
||||
|
||||
|
||||
def _draw_mat_box(img):
|
||||
"""Dev aid: outline the A5 mat opening so the matte and centring show."""
|
||||
x0, y0 = round((PANEL_W - A5_W) / 2), round((PANEL_H - A5_H) / 2)
|
||||
ImageDraw.Draw(img).rectangle((x0, y0, PANEL_W - x0 - 1, PANEL_H - y0 - 1),
|
||||
outline=(170, 60, 56), width=2)
|
||||
|
||||
|
||||
# --- hardware ---------------------------------------------------------------
|
||||
def push_panel(img, rotate, saturation, panel=""):
|
||||
"""Rotate to the panel's landscape buffer and push. Lazy import so this
|
||||
module still loads on a machine without the Inky library."""
|
||||
if rotate not in (90, 270):
|
||||
print(f"rotate must be 90 or 270, not {rotate}; using 90", file=sys.stderr)
|
||||
rotate = 90
|
||||
if panel == "el133uf1":
|
||||
from inky.inky_el133uf1 import Inky
|
||||
dev = Inky(resolution=(1600, 1200))
|
||||
else:
|
||||
from inky.auto import auto
|
||||
dev = auto()
|
||||
buf = img.rotate(rotate, expand=True)
|
||||
if buf.size != (dev.width, dev.height):
|
||||
buf = buf.resize((dev.width, dev.height), Image.LANCZOS)
|
||||
kw = {"saturation": saturation} if "saturation" in inspect.signature(dev.set_image).parameters else {}
|
||||
dev.set_image(buf, **kw)
|
||||
dev.show()
|
||||
|
||||
|
||||
# --- state ------------------------------------------------------------------
|
||||
def load_state(path):
|
||||
try:
|
||||
with open(os.path.expanduser(path)) as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return {"signature": None, "last_refresh": 0}
|
||||
|
||||
|
||||
def save_state(path, sig, when):
|
||||
path = os.path.expanduser(path)
|
||||
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
||||
tmp = path + ".tmp"
|
||||
with open(tmp, "w") as f:
|
||||
json.dump({"signature": sig, "last_refresh": when}, f)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp, path) # atomic: a power cut can't leave a half-written file
|
||||
|
||||
|
||||
def in_quiet_hours(cfg, hour):
|
||||
s, e = cfg["quiet_start"], cfg["quiet_end"]
|
||||
if s == e:
|
||||
return False
|
||||
return s <= hour < e if s < e else hour >= s or hour < e
|
||||
|
||||
|
||||
# --- run --------------------------------------------------------------------
|
||||
def obtain_image(cfg):
|
||||
if cfg["shoot"]:
|
||||
from shoot import shoot
|
||||
out = os.path.join(os.path.expanduser(cfg["cache"]), "shot.png")
|
||||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||||
shoot(cfg["base_url"], out, title=cfg["shoot_title"], subtitle=cfg["shoot_subtitle"],
|
||||
headline_px=cfg["shoot_headline_px"], eyebrow_px=cfg["shoot_eyebrow_px"],
|
||||
lowercase=cfg["shoot_lowercase"], mat=cfg["shoot_mat"],
|
||||
small_floor=cfg["shoot_small_floor"], timeout_ms=cfg["timeout"] * 1000,
|
||||
user=cfg["basic_user"], password=cfg["basic_pass"])
|
||||
return Image.open(out).convert("RGB")
|
||||
src = cfg["image_url"] or cfg["image"]
|
||||
if not src:
|
||||
raise ValueError("set image, image_url, or shoot in config")
|
||||
return get_image(src, cfg["timeout"], _auth(cfg))
|
||||
|
||||
|
||||
def run(cfg, preview=None, force=False, use_signature=True, mat_box=False):
|
||||
now = time.time()
|
||||
state = load_state(cfg["state"])
|
||||
sig = None
|
||||
if use_signature:
|
||||
try:
|
||||
sig = signature(fetch_recent(cfg["base_url"], cfg["hours"], cfg["timeout"], _auth(cfg)))
|
||||
except Exception as e:
|
||||
print(f"signature fetch failed: {e}", file=sys.stderr) # treat as no change
|
||||
heal_due = now - state.get("last_refresh", 0) >= cfg["heal_hours"] * 3600
|
||||
changed = (not use_signature) or (sig is not None and sig != state.get("signature"))
|
||||
if not force and not preview:
|
||||
if in_quiet_hours(cfg, datetime.now().hour):
|
||||
print("quiet hours; skip")
|
||||
return
|
||||
if not changed and not heal_due:
|
||||
print("no change; skip")
|
||||
return
|
||||
print("refresh:", "changed" if changed else "heal")
|
||||
|
||||
try:
|
||||
img = fit_panel(obtain_image(cfg))
|
||||
except Exception as e:
|
||||
print(f"could not get image: {e}", file=sys.stderr) # keep last panel image
|
||||
return
|
||||
img = mat_and_center(img, cfg["mat"])
|
||||
if preview:
|
||||
out = quantize_spectra6(img)
|
||||
if mat_box:
|
||||
_draw_mat_box(out)
|
||||
out.save(preview)
|
||||
print(f"wrote preview {preview}")
|
||||
return
|
||||
try:
|
||||
push_panel(img, cfg["rotate"], cfg["saturation"], cfg.get("panel", ""))
|
||||
except Exception as e:
|
||||
print(f"panel push failed: {e}", file=sys.stderr)
|
||||
return
|
||||
save_state(cfg["state"], sig if sig is not None else state.get("signature"), now)
|
||||
print("panel updated")
|
||||
|
||||
|
||||
def load_config(path):
|
||||
cfg = dict(DEFAULTS)
|
||||
if path:
|
||||
with open(os.path.expanduser(path), "rb") as f:
|
||||
cfg.update(tomllib.load(f))
|
||||
return cfg
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="Push the collage screenshot to the Inky panel.")
|
||||
ap.add_argument("--config")
|
||||
ap.add_argument("--base-url")
|
||||
ap.add_argument("--image")
|
||||
ap.add_argument("--image-url")
|
||||
ap.add_argument("--preview", help="write a 6-ink preview PNG instead of pushing")
|
||||
ap.add_argument("--rotate", type=int)
|
||||
ap.add_argument("--force", action="store_true", help="refresh even if unchanged")
|
||||
ap.add_argument("--no-signature", action="store_true", help="skip change detection")
|
||||
ap.add_argument("--mat-box", action="store_true", help="dev: outline the mat window on the preview")
|
||||
args = ap.parse_args()
|
||||
|
||||
cfg = load_config(args.config)
|
||||
for key in ("base_url", "image", "image_url"):
|
||||
val = getattr(args, key)
|
||||
if val:
|
||||
cfg[key] = val
|
||||
if args.rotate is not None:
|
||||
cfg["rotate"] = args.rotate
|
||||
run(cfg, preview=args.preview, force=args.force, use_signature=not args.no_signature, mat_box=args.mat_box)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user