pi_epaper_agent.py

"""
Runs on a Raspberry Pi wired to a real SPI e-paper HAT. Sends heartbeats
to the hub on an interval, and once registered, polls GET /render/{id} and
draws the result to the actual panel. This is the piece that turns "the
hub renders PNGs" into "a screen on the desk shows something" — everything
before this script only ever fed simulate_device.py or a browser preview.

Requires Waveshare's e-Paper Python library (the de facto driver for most
SPI e-paper HATs sold for the Pi): clone
https://github.com/waveshare/e-Paper and `pip install -e python/` from
their repo, or copy `python/lib/waveshare_epd` next to this script. Which
driver module to import depends on the *exact* panel model — there is no
generic driver, and no way to auto-detect it:

    epd2in13_V4    2.13" mono
    epd4in2b_V2    4.2" black/white/red
    epd7in5_V2     7.5" grayscale (gray4-capable)

Check the model printed on your HAT (or Waveshare's wiki for the product
page you bought from) and set --driver accordingly.

Not hardware-tested by this session (no physical e-paper panel attached)
beyond the parts that don't touch real SPI hardware — see
pi_epaper_agent_selftest.py, which exercises the hub-interaction and
bwr-plane-splitting logic against a fake EPD and a real running hub.

Usage:
    python pi_epaper_agent.py \\
        --hub http://mac-mini.local:8000 \\
        --driver epd7in5_V2 \\
        --profile epaper_7.5_gray4 --width 800 --height 480 --color-mode gray4
"""

import argparse
import hashlib
import importlib
import io
import time
import uuid

import requests
from PIL import Image


def get_device_id() -> str:
    """MAC-derived and stable across re-flashing/re-imaging the SD card.
    uuid.getnode() falls back to a random-but-persistent-per-boot value if
    no real MAC is available (rare on a Pi) — fine for identity purposes,
    just won't survive a fresh OS image in that fallback case."""
    return f"pi-{uuid.getnode():012x}"


def load_epd(driver_name: str):
    module = importlib.import_module(f"waveshare_epd.{driver_name}")
    epd = module.EPD()
    epd.init()
    return epd


def png_to_epd_buffers(png_bytes: bytes, color_mode: str, epd):
    """Turns the hub's already-quantized PNG into whatever buffer shape
    the Waveshare driver's display() expects for this color mode."""
    img = Image.open(io.BytesIO(png_bytes)).convert("RGB")

    if color_mode == "bwr":
        # Tri-color panels take two separate 1-bit planes (black, red).
        # The hub already quantized to the real bwr palette (render.py's
        # BWR_PALETTE), so this is just sorting pixels the hub already
        # decided were black/white/red back into the two planes the
        # driver wants — no color decisions made here.
        black = Image.new("1", img.size, 1)  # 1 = white in PIL's "1" mode
        red = Image.new("1", img.size, 1)
        bp, rp = black.load(), red.load()
        px = img.load()
        for y in range(img.height):
            for x in range(img.width):
                r, g, b = px[x, y]
                if r < 100 and g < 100 and b < 100:
                    bp[x, y] = 0
                elif r > 150 and g < 100 and b < 100:
                    rp[x, y] = 0
        return (epd.getbuffer(black), epd.getbuffer(red))

    if color_mode == "gray4" and hasattr(epd, "getbuffer_4Gray"):
        return (epd.getbuffer_4Gray(img),)

    # mono, and gray4-without-a-4Gray-buffer-helper: 1-bit is already what
    # the hub sent (render.apply_color_mode already dithered it).
    return (epd.getbuffer(img.convert("1")),)


def draw(epd, png_bytes: bytes, color_mode: str):
    buffers = png_to_epd_buffers(png_bytes, color_mode, epd)
    epd.display(*buffers)


def run(args):
    device_id = get_device_id()
    print(f"[{device_id}] starting — hub={args.hub} driver={args.driver}")

    epd = load_epd(args.driver)
    last_content_hash = None
    last_etag = None
    sleep_seconds = args.interval

    while True:
        try:
            hb = requests.post(
                f"{args.hub}/api/heartbeat",
                json={
                    "device_id": device_id,
                    "profile": args.profile,
                    "capabilities": {"width": args.width, "height": args.height, "color_mode": args.color_mode},
                },
                timeout=10,
            )
            hb.raise_for_status()
            status = hb.json()
            print(f"[{device_id}] heartbeat -> {status}")
            # Hub-side pacing (quiet hours: overnight + work blocks) —
            # the hub says how soon to check back, so the fleet's cadence
            # is tuned centrally instead of baked in at deploy time. An
            # older hub without the field just means args.interval.
            sleep_seconds = status.get("poll_seconds") or args.interval

            if status["status"] == "registered":
                # If-None-Match: the hub answers 304 (no body) when
                # content hasn't changed since the ETag we last saw —
                # which is most polls. The hash comparison below stays as
                # a second line of defense (and for hubs without ETags).
                headers = {"If-None-Match": last_etag} if last_etag else {}
                resp = requests.get(f"{args.hub}/render/{device_id}", headers=headers, timeout=15)
                if resp.status_code == 304:
                    print(f"[{device_id}] content unchanged (304), skipping redraw")
                else:
                    resp.raise_for_status()
                    last_etag = resp.headers.get("ETag")
                    content_hash = hashlib.sha256(resp.content).hexdigest()
                    if content_hash != last_content_hash:
                        print(f"[{device_id}] content changed, drawing")
                        draw(epd, resp.content, args.color_mode)
                        last_content_hash = content_hash
                    else:
                        print(f"[{device_id}] content unchanged, skipping redraw (e-paper wear/flicker isn't free)")
        except requests.RequestException as e:
            print(f"[{device_id}] request failed: {e}")
        except Exception as e:
            print(f"[{device_id}] draw failed: {e}")

        time.sleep(sleep_seconds)


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--hub", required=True, help="hub base URL, e.g. http://mac-mini.local:8000")
    parser.add_argument("--driver", required=True, help="waveshare_epd module name, e.g. epd7in5_V2")
    parser.add_argument("--profile", required=True, help="e.g. epaper_7.5_gray4")
    parser.add_argument("--width", type=int, required=True)
    parser.add_argument("--height", type=int, required=True)
    parser.add_argument("--color-mode", required=True, choices=["mono", "gray4", "bwr", "color"])
    parser.add_argument("--interval", type=float, default=60.0, help="seconds between heartbeat+render checks")
    args = parser.parse_args()
    run(args)


if __name__ == "__main__":
    main()