diff --git a/inky_today.py b/inky_today.py new file mode 100644 index 0000000..d22c93b --- /dev/null +++ b/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()) diff --git a/pictures/cosmicstag.png b/pictures/cosmicstag.png new file mode 100644 index 0000000..392a805 Binary files /dev/null and b/pictures/cosmicstag.png differ