feat: add nostr-driven renderer for tasks scene

inky_today.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python3
+"""Render today's events and trash/recycling chores from Nostr to the Inky.
+
+Runs on the Pi against the pimoroni venv:
+
+    ~/.virtualenvs/pimoroni/bin/python ~/InkyImpression/inky_today.py [out.png] [--no-push]
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import subprocess
+import sys
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+
+import websockets
+from PIL import Image, ImageChops, ImageDraw, ImageFont
+
+
+PUBKEYS = {
+    "24076f99c21dfed5201087e99d9e51a6c4927b7ec7101181bb0186718bdf6c4e": "Pat",
+    "589ba6abe1cc0a20a47481dfe18786b948883212e1cbc15cb14fcd813ec2f861": "Coco",
+    "5a737aafbe78be03d72904bbf79282205ac6efb91c2ffe182c1015d5522164a6": "Castle",
+}
+# RELAYS = ["wss://lnbits.ariege.io/nostrrelay/castle"]
+RELAYS = ["ws://127.0.0.1:5001/nostrrelay/test"]
+TRASH_KEYWORDS = ("trash", "recycling", "recycle")
+
+W, H = 1200, 1600  # portrait; rotated 90° CCW before being sent to the panel.
+WHITE = (255, 255, 255)
+BLACK = (0, 0, 0)
+RED = (220, 30, 30)
+
+FONT_REGULAR = "DejaVu Sans"
+FONT_BOLD = "DejaVu Sans:weight=bold"
+
+LOGO_PATH = "/home/padreug/Pictures/castle2.png"
+LOGO_SIZE = 220
+BG_PATH = "pictures/cosmicstag.png"
+BG_STRENGTH = 0.40  # 0.0 = invisible, 1.0 = full intensity
+
+WEEKDAYS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
+
+
+@dataclass
+class Task:
+    addr: str
+    kind: int
+    pubkey: str
+    created_at: int
+    title: str
+    content: str
+    start: str
+    location: str
+
+
+@dataclass
+class Claim:
+    event_id: str
+    pubkey: str
+    created_at: int
+    task_addr: str
+    status: str
+    occurrence: str
+
+
+def first_tag(tags, name):
+    for t in tags:
+        if t and t[0] == name:
+            return t[1] if len(t) > 1 else ""
+    return ""
+
+
+def all_tags(tags, name):
+    return [t[1] for t in tags if t and t[0] == name and len(t) > 1]
+
+
+def parse_task(ev):
+    d = first_tag(ev["tags"], "d")
+    if not d:
+        return None
+    return Task(
+        addr=f"{ev['kind']}:{ev['pubkey']}:{d}",
+        kind=ev["kind"],
+        pubkey=ev["pubkey"],
+        created_at=ev["created_at"],
+        title=first_tag(ev["tags"], "title"),
+        content=ev.get("content", ""),
+        start=first_tag(ev["tags"], "start"),
+        location=first_tag(ev["tags"], "location"),
+    )
+
+
+def parse_claim(ev):
+    a = first_tag(ev["tags"], "a")
+    if not a:
+        return None
+    return Claim(
+        event_id=ev["id"],
+        pubkey=ev["pubkey"],
+        created_at=ev["created_at"],
+        task_addr=a,
+        status=first_tag(ev["tags"], "task-status").lower(),
+        occurrence=first_tag(ev["tags"], "occurrence"),
+    )
+
+
+async def fetch_relay(url, filters, timeout):
+    events = []
+    sub_id = "today"
+    try:
+        async with websockets.connect(url, open_timeout=10, close_timeout=5) as ws:
+            await ws.send(json.dumps(["REQ", sub_id, *filters]))
+            try:
+                async with asyncio.timeout(timeout):
+                    async for msg in ws:
+                        m = json.loads(msg)
+                        if m[0] == "EVENT" and m[1] == sub_id:
+                            events.append(m[2])
+                        elif m[0] == "EOSE" and m[1] == sub_id:
+                            break
+            except asyncio.TimeoutError:
+                pass
+            try:
+                await ws.send(json.dumps(["CLOSE", sub_id]))
+            except Exception:
+                pass
+    except Exception as exc:
+        print(f"  {url}: {exc}", file=sys.stderr)
+    return events
+
+
+async def fetch_all(relays, filters, timeout=10.0):
+    results = await asyncio.gather(*(fetch_relay(r, filters, timeout) for r in relays))
+    seen = {}
+    for evs in results:
+        for ev in evs:
+            seen[ev["id"]] = ev
+    return list(seen.values())
+
+
+def assemble_state(events):
+    tasks_by_addr: dict[str, Task] = {}
+    claims: list[Claim] = []
+    deleted_addrs: set[str] = set()
+    deleted_ids: set[str] = set()
+
+    for ev in events:
+        kind = ev.get("kind")
+        if kind in (31922, 31923):
+            t = parse_task(ev)
+            if t and (t.addr not in tasks_by_addr or tasks_by_addr[t.addr].created_at < t.created_at):
+                tasks_by_addr[t.addr] = t
+        elif kind == 31925:
+            c = parse_claim(ev)
+            if c:
+                claims.append(c)
+        elif kind == 5:
+            for a in all_tags(ev["tags"], "a"):
+                deleted_addrs.add(a)
+            for e_id in all_tags(ev["tags"], "e"):
+                deleted_ids.add(e_id)
+
+    tasks_by_addr = {a: t for a, t in tasks_by_addr.items() if a not in deleted_addrs}
+    claims = [c for c in claims if c.event_id not in deleted_ids]
+    return tasks_by_addr, claims
+
+
+def start_date(start):
+    # kind 31923 `start` is a unix timestamp string per NIP-52; kind 31922
+    # is ISO date or datetime. Return YYYY-MM-DD in local time.
+    if not start:
+        return ""
+    if start.isdigit():
+        return datetime.fromtimestamp(int(start)).strftime("%Y-%m-%d")
+    return start[:10]
+
+
+def fires_today(task, today_iso):
+    return start_date(task.start) == today_iso
+
+
+def is_event(task):
+    # NIP-52 calendar event: kind 31923, OR kind 31922 with a date-only start.
+    if task.kind == 31923:
+        return True
+    return task.kind == 31922 and "T" not in task.start
+
+
+def is_chore(task):
+    # Household task instance: kind 31922 with a datetime start.
+    return task.kind == 31922 and "T" in task.start
+
+
+def matches_trash(task):
+    haystack = f"{task.title} {task.content}".lower()
+    return any(k in haystack for k in TRASH_KEYWORDS)
+
+
+def latest_claim_for(task, claims, today_iso):
+    relevant = []
+    for c in claims:
+        if c.task_addr != task.addr:
+            continue
+        if is_chore(task):
+            occ = c.occurrence[:10] if c.occurrence else ""
+            if occ and occ != today_iso:
+                continue
+        relevant.append(c)
+    if not relevant:
+        return None
+    return max(relevant, key=lambda c: c.created_at)
+
+
+def load_font(query, size):
+    try:
+        r = subprocess.run(
+            ["fc-match", "-f", "%{file}", query],
+            capture_output=True, text=True, timeout=3,
+        )
+        path = r.stdout.strip()
+        if r.returncode == 0 and path and Path(path).exists():
+            return ImageFont.truetype(path, size)
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        pass
+    return ImageFont.load_default()
+
+
+def truncate(text, font, max_w, d):
+    if d.textlength(text, font=font) <= max_w:
+        return text
+    while text and d.textlength(text + "…", font=font) > max_w:
+        text = text[:-1]
+    return text + "…"
+
+
+def format_time(start):
+    if not start:
+        return ""
+    if start.isdigit():
+        return datetime.fromtimestamp(int(start)).strftime("%H:%M")
+    if len(start) <= 10:
+        return ""
+    s = start.replace("Z", "+00:00")
+    try:
+        return datetime.fromisoformat(s).strftime("%H:%M")
+    except ValueError:
+        return ""
+
+
+def paste_logo(img, path, size, margin):
+    if not Path(path).exists():
+        return
+    logo = Image.open(path).convert("RGBA")
+    logo.thumbnail((size, size), Image.LANCZOS)
+    img.paste(logo, (W - margin - logo.width, margin-20), logo)
+
+
+def composite_background(img, path, strength):
+    if not Path(path).exists():
+        return img
+    src = Image.open(path).convert("RGBA")
+    # Flatten alpha against white so transparent pixels don't read as black.
+    flat = Image.new("RGB", src.size, WHITE)
+    flat.paste(src, mask=src.split()[3])
+    # Fit-inside the canvas, centered.
+    scale = min(W / flat.width, H / flat.height)
+    new_size = (int(flat.width * scale), int(flat.height * scale))
+    flat = flat.resize(new_size, Image.LANCZOS)
+    overlay = Image.new("RGB", (W, H), WHITE)
+    overlay.paste(flat, ((W - new_size[0]) // 2, (H - new_size[1]) // 2))
+    # Soften toward white, then multiply — preserves canvas whites cleanly.
+    white_full = Image.new("RGB", (W, H), WHITE)
+    softened = Image.blend(white_full, overlay, strength)
+    return ImageChops.multiply(img, softened)
+
+
+def render(events_today, chores_today, claims, today_iso, weekday):
+    img = Image.new("RGB", (W, H), WHITE)
+    img = composite_background(img, BG_PATH, BG_STRENGTH)
+    paste_logo(img, LOGO_PATH, LOGO_SIZE, 60)
+    d = ImageDraw.Draw(img)
+
+    f_title = load_font(FONT_BOLD, 96)
+    f_date = load_font(FONT_REGULAR, 56)
+    f_section = load_font(FONT_BOLD, 56)
+    f_item = load_font(FONT_BOLD, 52)
+    f_meta = load_font(FONT_REGULAR, 40)
+    f_badge = load_font(FONT_BOLD, 40)
+    f_footer = load_font(FONT_REGULAR, 28)
+
+    PAD = 60
+    INDENT = PAD + 20
+    y = PAD
+
+    d.text((PAD, y), weekday.upper(), font=f_title, fill=BLACK)
+    y += 110
+    d.text((PAD, y), datetime.now().strftime("%-d %B %Y"), font=f_date, fill=BLACK)
+    y += 90
+
+    def divider(yy):
+        d.line([(PAD, yy), (W - PAD, yy)], fill=BLACK, width=3)
+
+    divider(y); y += 30
+
+    d.text((PAD, y), "EVENTS TODAY", font=f_section, fill=BLACK)
+    y += 80
+    if not events_today:
+        d.text((INDENT, y), "—", font=f_meta, fill=BLACK)
+        y += 60
+    else:
+        for t in events_today[:3]:
+            title = truncate(t.title or "(untitled)", f_item, W - 2 * PAD - 20, d)
+            d.text((INDENT, y), title, font=f_item, fill=BLACK)
+            y += 60
+            time_str = format_time(t.start) or "all day"
+            d.text((INDENT, y), time_str, font=f_meta, fill=RED)
+            if t.location:
+                sep = "  ·  "
+                offset = d.textlength(time_str + sep, font=f_meta)
+                loc = truncate(t.location, f_meta, W - 2 * PAD - 20 - offset, d)
+                d.text((INDENT + d.textlength(time_str, font=f_meta), y), sep + loc, font=f_meta, fill=BLACK)
+            y += 70
+
+    y += 20
+    divider(y); y += 30
+
+    d.text((PAD, y), "TRASH & RECYCLING", font=f_section, fill=BLACK)
+    y += 80
+    if not chores_today:
+        d.text((INDENT, y), "—", font=f_meta, fill=BLACK)
+        y += 60
+    else:
+        for t in chores_today:
+            title = truncate(t.title or "(untitled)", f_item, W - 2 * PAD - 20, d)
+            d.text((INDENT, y), title, font=f_item, fill=BLACK)
+            y += 60
+            claim = latest_claim_for(t, claims, today_iso)
+            if claim and claim.status in ("claimed", "in-progress", "completed"):
+                name = PUBKEYS.get(claim.pubkey, claim.pubkey[:8])
+                tick = " ✓" if claim.status == "completed" else ""
+                d.text((INDENT + 20, y), f"→ {name}{tick}", font=f_meta, fill=BLACK)
+            else:
+                badge = "UNCLAIMED"
+                bbox = d.textbbox((0, 0), badge, font=f_badge)
+                bw = bbox[2] - bbox[0] + 28
+                bh = bbox[3] - bbox[1] + 18
+                d.rectangle([(INDENT + 20, y - 2), (INDENT + 20 + bw, y + bh)], fill=RED)
+                d.text((INDENT + 34, y + 4), badge, font=f_badge, fill=WHITE)
+            y += 80
+
+    y += 20
+    divider(y); y += 30
+
+    d.text((PAD, y), "OPEN TASKS", font=f_section, fill=BLACK)
+    y += 70
+    d.text((INDENT, y), "(coming soon)", font=f_meta, fill=BLACK)
+
+    footer = f"refreshed {datetime.now().strftime('%H:%M')}  ·  {len(claims)} claims known"
+    d.text((PAD, H - PAD - 30), footer, font=f_footer, fill=BLACK)
+
+    return img
+
+
+def push_to_inky(img):
+    panel_img = img.rotate(90, expand=True)
+    from inky.auto import auto
+    inky = auto()
+    if panel_img.size != tuple(inky.resolution):
+        panel_img = panel_img.resize(inky.resolution)
+    inky.set_image(panel_img)
+    inky.show()
+
+
+async def main():
+    today_iso = datetime.now().strftime("%Y-%m-%d")
+    weekday = WEEKDAYS[datetime.now().weekday()]
+    print(f"today: {today_iso} ({weekday})")
+
+    authors = list(PUBKEYS)
+    filters = [
+        # {"kinds": [31922, 31923], "authors": authors, "limit": 400},
+        {"kinds": [31922, 31923], "limit": 400},
+        {"kinds": [31925], "authors": authors, "limit": 500},
+        {"kinds": [5], "authors": authors, "limit": 200},
+    ]
+    print(f"fetching from {len(RELAYS)} relays…")
+    events = await fetch_all(RELAYS, filters)
+    print(f"  {len(events)} unique events")
+
+    tasks_by_addr, claims = assemble_state(events)
+    print(f"  {len(tasks_by_addr)} tasks, {len(claims)} claims after dedup/deletions")
+
+    todays = [t for t in tasks_by_addr.values() if fires_today(t, today_iso)]
+    events_today = [t for t in todays if is_event(t)]
+    chores_today = [t for t in todays if is_chore(t) and matches_trash(t)]
+    print(f"  events today: {len(events_today)}, trash/recycling chores: {len(chores_today)}")
+
+    img = render(events_today, chores_today, claims, today_iso, weekday)
+
+    args = sys.argv[1:]
+    no_push = "--no-push" in args
+    out = next((a for a in args if not a.startswith("--")), None)
+    if out:
+        img.save(out)
+        print(f"saved PNG: {out}")
+
+    if no_push:
+        print("skipping inky push (--no-push)")
+        return
+    print("pushing to inky…")
+    push_to_inky(img)
+    print("done.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())