Ruben Hensen
145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
"""E-paper renderer loop.
|
|
|
|
Pipeline:
|
|
Playwright screenshot of $DASHBOARD_URL at 1304x984
|
|
-> Pillow 1-bit dither (Floyd-Steinberg)
|
|
-> Waveshare 12.48" driver push
|
|
|
|
Tunables come from env: DASHBOARD_URL, REFRESH_SECONDS, PANEL_VARIANT.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
log = logging.getLogger("eink.renderer")
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
|
|
WIDTH, HEIGHT = 1304, 984
|
|
DASHBOARD_URL = os.environ.get("DASHBOARD_URL", "http://127.0.0.1:3000")
|
|
REFRESH_SECONDS = int(os.environ.get("REFRESH_SECONDS", "300"))
|
|
PANEL_VARIANT = os.environ.get("PANEL_VARIANT", "bw") # bw | bwr
|
|
|
|
|
|
def screenshot(url: str) -> Image.Image:
|
|
"""Headless Chromium screenshot at the panel's native resolution."""
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
|
|
ctx = browser.new_context(
|
|
viewport={"width": WIDTH, "height": HEIGHT},
|
|
device_scale_factor=1,
|
|
)
|
|
page = ctx.new_page()
|
|
page.goto(url, wait_until="networkidle", timeout=30_000)
|
|
png = page.screenshot(full_page=False, type="png", omit_background=False)
|
|
browser.close()
|
|
return Image.open(io.BytesIO(png)).convert("RGB")
|
|
|
|
|
|
def to_bw(img: Image.Image) -> Image.Image:
|
|
"""Floyd-Steinberg dither to 1-bit. Pillow's built-in is C-accelerated."""
|
|
return img.convert("L").convert("1", dither=Image.Dither.FLOYDSTEINBERG)
|
|
|
|
|
|
def to_bwr(img: Image.Image) -> tuple[Image.Image, Image.Image]:
|
|
"""Separate red plane from B/W plane for the Module (B) variant.
|
|
|
|
Naive thresholding: pixels with R > 180 and G,B < 100 -> red plane.
|
|
Everything else -> dithered B/W.
|
|
"""
|
|
import numpy as np
|
|
|
|
arr = np.array(img)
|
|
r, g, b = arr[..., 0], arr[..., 1], arr[..., 2]
|
|
red_mask = (r > 180) & (g < 100) & (b < 100)
|
|
|
|
bw_arr = arr.copy()
|
|
bw_arr[red_mask] = 255 # remove red from BW plane
|
|
bw_img = to_bw(Image.fromarray(bw_arr))
|
|
|
|
red_img = Image.fromarray((red_mask * 255).astype("uint8"), mode="L").convert("1")
|
|
return bw_img, red_img
|
|
|
|
|
|
def push_to_panel_bw(bw: Image.Image) -> None:
|
|
"""Push a 1-bit image to the B/W 12.48\" panel.
|
|
|
|
NOTE: requires the Waveshare 12.48\" Python driver vendored under
|
|
renderer/waveshare/. See waveshare/README.md for install steps.
|
|
"""
|
|
try:
|
|
from waveshare import epd12in48 # type: ignore
|
|
except ImportError:
|
|
log.warning("Waveshare driver not vendored — dry-run mode (no panel push).")
|
|
bw.save("/tmp/eink-last.png")
|
|
return
|
|
|
|
epd = epd12in48.EPD()
|
|
epd.Init()
|
|
epd.display(epd.getbuffer(bw))
|
|
epd.EPD_Sleep()
|
|
|
|
|
|
def push_to_panel_bwr(bw: Image.Image, red: Image.Image) -> None:
|
|
try:
|
|
from waveshare import epd12in48b # type: ignore
|
|
except ImportError:
|
|
log.warning("Waveshare (B) driver not vendored — dry-run mode (no panel push).")
|
|
bw.save("/tmp/eink-last-bw.png")
|
|
red.save("/tmp/eink-last-red.png")
|
|
return
|
|
|
|
epd = epd12in48b.EPD()
|
|
epd.Init()
|
|
epd.display(epd.getbuffer(bw), epd.getbuffer(red))
|
|
epd.EPD_Sleep()
|
|
|
|
|
|
def render_once() -> None:
|
|
log.info("Screenshot %s", DASHBOARD_URL)
|
|
img = screenshot(DASHBOARD_URL)
|
|
if PANEL_VARIANT == "bwr":
|
|
bw, red = to_bwr(img)
|
|
push_to_panel_bwr(bw, red)
|
|
else:
|
|
push_to_panel_bw(to_bw(img))
|
|
log.info("Frame pushed to panel.")
|
|
|
|
|
|
def main() -> None:
|
|
stop = False
|
|
|
|
def handle(*_):
|
|
nonlocal stop
|
|
stop = True
|
|
log.info("Shutdown signal — exiting after current frame.")
|
|
|
|
signal.signal(signal.SIGTERM, handle)
|
|
signal.signal(signal.SIGINT, handle)
|
|
|
|
while not stop:
|
|
started = time.monotonic()
|
|
try:
|
|
render_once()
|
|
except Exception:
|
|
log.exception("Render cycle failed; will retry next interval.")
|
|
elapsed = time.monotonic() - started
|
|
sleep_for = max(0, REFRESH_SECONDS - elapsed)
|
|
log.info("Sleeping %.1fs", sleep_for)
|
|
for _ in range(int(sleep_for)):
|
|
if stop:
|
|
return
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|