#!/usr/bin/env python3 """Glasses feed renderer. Captures each agent's tmux pane + status and writes pre-formatted view files that the glasses app (and its browser preview) poll. ALL glasses formatting lives here — edit this file and restart this process and the glasses pick it up on the next poll (~2s). No app rebuild, no serve.py restart, no re-scan. Run (lab box): nohup python3 glasses_renderer.py > /tmp/glasses_renderer.log 2>&1 & Writes atomically into the deployed app's feed/ dir, which is served (static) at omidlab.net/browse/para/.agents/reports/glasses_app/feed/. """ import json import os import re import subprocess import sys import time from datetime import datetime from pathlib import Path FEED = Path("/data/cameron/para/.agents/reports/glasses_app/feed") AGENTS_DIR = Path("/data/cameron/agents_stuff/agents") # --- formatting knobs (tune live) --- WIDTH = 48 # wrap each mirrored line at this many chars (display width) SCROLLBACK = 120 # lines of tmux history to expose for scrolling PERIOD = 2.0 # seconds between refreshes ANSI = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]|\x1b\][^\x07]*\x07") NONASCII = re.compile(r"[^\x20-\x7E]") SESSION = "agents" # mirror this tmux session's windows, in index order def session_windows(): """[(index, name), ...] for every window in SESSION, in tmux order.""" r = subprocess.run( ["tmux", "list-windows", "-t", SESSION, "-F", "#{window_index}\t#{window_name}"], capture_output=True, text=True, ) out = [] for line in r.stdout.strip().split("\n"): if "\t" in line: idx, name = line.split("\t", 1) out.append((idx, name)) return out def status_word(name): f = AGENTS_DIR / name / "status.md" if not f.exists(): return "" # service/utility windows have no status.md first = NONASCII.sub("", f.read_text().strip().split("\n")[0]).lstrip("#").strip() return (first.split() or [""])[0][:12] def wrap(line): if len(line) <= WIDTH: return [line] return [line[i:i + WIDTH] for i in range(0, len(line), WIDTH)] BULLET = "●" # ● — Claude Code prefixes each assistant turn/tool-step def last_response_start(raw): """Index in `raw` of the start of the latest assistant response burst (the line just before its first ● bullet — usually the user's message), so the view can open there and scroll down through the reply.""" bullets = [i for i, l in enumerate(raw) if l.lstrip().startswith(BULLET)] if not bullets: return None i = bullets[-1] bset = set(bullets) # walk up through the burst (bullets + their indented output + blanks) while i - 1 >= 0 and ((i - 1) in bset or not raw[i - 1].strip() or raw[i - 1].startswith(" ")): i -= 1 return max(0, i - 1) # include the line above the burst (the user message) def mirror(target): if target is None: return ["(no tmux window)"], None r = subprocess.run( ["tmux", "capture-pane", "-t", target, "-p", "-S", f"-{SCROLLBACK}"], capture_output=True, text=True, ) if r.returncode != 0: return ["(capture failed)"], None raw = ANSI.sub("", r.stdout).split("\n") # keep ● for burst detection start_raw = last_response_start(raw) lines, raw_to_wrapped = [], [] for l in raw: raw_to_wrapped.append(len(lines)) lines.extend(wrap(NONASCII.sub("", l.rstrip()))) while lines and not lines[-1].strip(): lines.pop() start = raw_to_wrapped[start_raw] if start_raw is not None and start_raw < len(raw_to_wrapped) else None return (lines or ["(empty)"]), start def write_atomic(path, data): tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data)) os.replace(tmp, path) # --- generic view builders (the client renders these verbatim) --- def text_view(lines, on_press=None, start=None, tts=None): v = {"mode": "text", "lines": lines} if on_press: v["onPress"] = on_press if start is not None: v["start"] = start if tts is not None: v["tts"] = tts return v def list_view(title, items): return {"mode": "list", "title": title, "items": items} def send_view(name, key): base = f"/api/agents/{name}" msg = lambda t: {"post": f"{base}/send", "body": {"text": t}, "then": "back"} return list_view(f"-> {key}", [ {"label": "speak (voice)", "do": {"record": {"to": f"{base}/send"}}}, {"label": "status?", "do": msg("status?")}, {"label": "whats blocking you?", "do": msg("whats blocking you?")}, {"label": "looks good, continue", "do": msg("looks good, continue")}, {"label": "STOP (interrupt)", "do": {"post": f"{base}/interrupt", "then": "back"}}, ]) # Hub launcher. (label, desc, action). action is what a press does — {go:feed} # for a feed-driven page, or a client mode like {notes:{...}}. Edit live. NOTES_WRAP = "Incoming note from audio, append this to bottom of notes app page and then respond: " MODULES = [ ("Chat", "voice -> agents", {"go": "chat_feed"}), ("Dashboard", "today at a glance", {"go": "dashboard"}), ("Persian", "tutor + audio", {"go": "persian"}), ("Spotify", "now playing", {"go": "module_spotify"}), ("Agents", "fleet status", {"go": "agents"}), ] CHAT_LOG = Path("/data/cameron/agents_stuff/agents/glasses/chat_log.md") def chat_view(): """Render the chat log (agents' concise replies + your commands), tail.""" try: text = CHAT_LOG.read_text() if CHAT_LOG.exists() else "" except Exception: text = "" raw = [NONASCII.sub("", x.rstrip()) for x in text.split("\n")] # voice synthesis DISABLED 2026-06-24 (ElevenLabs quota) — text-only now. lines = [] for l in raw: lines.extend(wrap(l) if l.strip() else [""]) while lines and not lines[-1].strip(): lines.pop() lines = lines[-80:] if not lines: lines = ["Chat", "-" * 40, "tap to speak. by default it", "CHECKS an agent and reports", "back (no interrupting).", "say \"tell X to ...\" to actually", "send a command."] return text_view(lines, {"record": {"to": "/api/glasses/chat/send", "after": "stay", "autosend": True}}) # --- Persian tutor: mirrors omidlab.net/persian. Tap to speak a question; # the response's transliteration + English show here (the G2 display strips # non-ASCII, so it can't render the Persian script — the translit IS the # read-along) and the Persian audio auto-plays on the phone via /api/tts. PERSIAN_STATE = Path("/data/cameron/agents_stuff/agents/glasses/persian_state.json") def persian_view(): try: turns = json.loads(PERSIAN_STATE.read_text()).get("turns", []) except Exception: turns = [] lines = [] for t in turns[-12:]: q = NONASCII.sub("", t.get("query", "")).strip() if q: lines.extend(wrap("> " + q)) # The G2 can't render Persian script — show underscores in its place # (one per glyph, spaces kept) so you can see script was there. fa = t.get("persian", "") if fa.strip(): lines.extend(wrap("".join(" " if c.isspace() else "_" for c in fa).strip())) tr = NONASCII.sub("", t.get("translit", "")).strip() if tr: lines.extend(wrap(tr)) en = NONASCII.sub("", t.get("english", "")).strip() if en: lines.extend(wrap("(" + en + ")")) nt = NONASCII.sub("", t.get("notes", "")).strip() if nt: lines.extend(wrap(nt)) lines.append("") while lines and not lines[-1].strip(): lines.pop() lines = lines[-80:] if not lines: lines = ["Persian", "-" * 40, "tap to speak a question, like", "\"how do you say good morning\".", "the transliteration + English", "show here; the Persian is read", "aloud."] # voice synthesis DISABLED 2026-06-24 (ElevenLabs quota) — translit is the # read-along; hear Persian audio on the website (omidlab.net/persian). return text_view(lines, {"record": {"to": "/api/glasses/persian/ask", "after": "stay", "autosend": True}}) def placeholder(label, desc): return text_view([label, "", f"not built yet — {desc}", "", "tell the glasses agent what", "to show here and it'll", "appear live."]) # --- Dashboard module: today at a glance. A list (title=clock) where each row # is one life domain — swipe between them, click to expand. Sources are # life_manager's KB files + WHOOP. The NOW row is tap-to-speak: it sets the # current task (start time stamped server-side, routed to life_manager). LIFE = Path("/data/cameron/life") NOW_STATE = Path("/data/cameron/agents_stuff/agents/glasses/now_state.json") EVENTS_FILE = Path("/data/cameron/agents_stuff/agents/glasses/events.json") def _now_task(): """(task, elapsed_str) from the state serve.py writes on tap-to-speak.""" try: s = json.loads(NOW_STATE.read_text()) started = datetime.fromisoformat(s["started_at"]) secs = (datetime.now(started.tzinfo) - started).total_seconds() h, m = int(secs // 3600), int((secs % 3600) // 60) return s.get("task", ""), (f"{h}:{m:02d}" if h else f"{m}m") except Exception: return None, None def _todo_summary(): """(done, total, next_undone) from life_manager's checklist.md.""" try: done = total = 0 nxt = None for l in (LIFE / "checklist.md").read_text().split("\n"): m = re.match(r"\s*-\s*\[( |x|X)\]\s*(.+)", l) if not m: continue total += 1 if m.group(1).lower() == "x": done += 1 elif nxt is None: nxt = NONASCII.sub("", re.sub(r"[*_`]|\([^)]*\)", "", m.group(2))).strip() return done, total, nxt except Exception: return None, None, None def _food_summary(): """(calories_in, target) from today's food log.""" try: t = (LIFE / "food" / (time.strftime("%Y-%m-%d") + ".md")).read_text() cin = re.search(r"[Rr]unning total:\**\s*~?([\d,]+)", t) tgt = re.search(r"[Tt]arget:\**\s*~?([\d,]+)", t) return (int(cin.group(1).replace(",", "")) if cin else None, int(tgt.group(1).replace(",", "")) if tgt else None) except Exception: return None, None _whoop_brief_cache = {"v": (None, None, None), "ts": 0.0} def _whoop_brief(): """(recovery%, hours_asleep, kcal_out) — cached 60s.""" now = time.time() if _whoop_brief_cache["ts"] and now - _whoop_brief_cache["ts"] < 60: return _whoop_brief_cache["v"] v = (None, None, None) try: if "/data/cameron/life/scripts" not in sys.path: sys.path.insert(0, "/data/cameron/life/scripts") from whoop_lib import get_latest_summary s = get_latest_summary() kj = s.get("kilojoules_today") v = (round(s["recovery_score"]) if s.get("recovery_score") is not None else None, s.get("hours_asleep"), round(kj / 4.184) if kj else None) except Exception: v = (None, None, None) _whoop_brief_cache.update(v=v, ts=now) return v def _next_event(): """Next upcoming event from events.json ([{time:'HH:MM',title}]); None if none.""" try: evs = json.loads(EVENTS_FILE.read_text()) nowhm = time.strftime("%H:%M") for e in sorted((e for e in evs if e.get("time")), key=lambda e: e["time"].zfill(5)): if e["time"].zfill(5) >= nowhm: return e return None except Exception: return None def dashboard_view(): clock = time.strftime("%-I:%M%p").lower() date = time.strftime("%a %b %-d") ev = _next_event() next_lbl = f"NEXT {ev['time']} {ev.get('title','')}" if ev else "NEXT --" d, t, nxt = _todo_summary() todo_lbl = f"TODO {d}/{t} {(nxt or '')[:22]}".rstrip() if t is not None else "TODO --" cin, tgt = _food_summary() rec, hrs, cout = _whoop_brief() cal_in = (f"{cin}" + (f"/{tgt}" if tgt else "")) if cin is not None else "--" cal_lbl = f"CAL {cal_in} in" + (f" {cout} out" if cout else "") body_lbl = (f"BODY {rec}% recov" if rec is not None else "BODY --") + (f" {hrs:.1f}h" if hrs else "") return list_view(f"{clock} - {date}", [ {"label": next_lbl, "do": {"go": "dash_cal"}}, {"label": todo_lbl, "do": {"go": "dash_todos"}}, {"label": cal_lbl, "do": {"go": "dash_food"}}, {"label": body_lbl, "do": {"go": "module_whoop"}}, ]) def _file_text_view(title, path, empty): try: raw = [NONASCII.sub("", x.rstrip()) for x in Path(path).read_text().split("\n")] except Exception: raw = [] lines = [] for l in raw: lines.extend(wrap(l) if l.strip() else [""]) while lines and not lines[-1].strip(): lines.pop() return text_view([title, "-" * 40] + (lines or [empty])) def dash_todos_view(): return _file_text_view("Today's Todos", LIFE / "checklist.md", "no checklist yet") def dash_food_view(): return _file_text_view("Food Log", LIFE / "food" / (time.strftime("%Y-%m-%d") + ".md"), "no food logged today") def dash_cal_view(): try: evs = sorted(json.loads(EVENTS_FILE.read_text()), key=lambda e: e.get("time", "").zfill(5)) except Exception: evs = [] if not evs: return text_view(["Calendar", "-" * 40, "no events today.", "", "tell life_manager your plans", "and they'll show up here."]) return text_view(["Calendar", "-" * 40] + [f"{e.get('time','')} {e.get('title','')}" for e in evs]) # --- Spotify module: now playing + live synced lyrics (LRCLIB) --- _lyrics_cache = {} SPOTIFY_PRESS = {"go": "spotify_menu"} # tap on now-playing -> options menu def spotify_menu_view(): return list_view("Spotify", [ {"label": "Start / Pause", "do": {"post": "/api/glasses/spotify/playpause", "then": "back"}}, {"label": "Next Song", "do": {"post": "/api/glasses/spotify/next", "then": "back"}}, {"label": "Search for a new song", "do": {"spotify_search": True}}, ]) LYRICS_NEG_TTL = 90 # re-try "no lyrics" after this long (don't poison on a transient miss) def _clean_title(name): """Drop remaster/version/feat suffixes so LRCLIB's clean titles still match.""" n = re.sub(r"\s*[-(].*?\b(remaster(ed)?|version|deluxe|edit|mono|stereo|featuring|feat|live|" r"acoustic|radio|anniversary|expanded|bonus|remix|single|album)\b.*$", "", name, flags=re.I) n = re.sub(r"\s*\([^)]*\)\s*$", "", n) # any leftover trailing parenthetical return n.strip() or name def _lrclib_synced(track_name, artist): import requests r = requests.get("https://lrclib.net/api/search", params={"track_name": track_name, "artist_name": artist}, headers={"User-Agent": "para-glasses/1.0"}, timeout=10) return [a for a in r.json() if a.get("syncedLyrics")] def _parse_lrc(lrc): out = [] for line in (lrc or "").split("\n"): m = re.match(r"\[(\d+):(\d+)[.:](\d+)\]\s*(.*)", line) if m: mm, ss, fr, text = m.groups() ms = (int(mm) * 60 + int(ss)) * 1000 + (int(fr) * 10 if len(fr) <= 2 else int(fr)) if text.strip(): out.append((ms, text.strip())) return out def _lyrics(track): # cache holds (lyrics, fetched_at). Good lyrics are kept forever; an empty # result is only trusted for LYRICS_NEG_TTL, so a transient LRCLIB timeout # doesn't permanently poison a song that really does have lyrics. key = (track["name"], track["artist"], track.get("album")) cached = _lyrics_cache.get(key) if cached and (cached[0] or (time.time() - cached[1]) < LYRICS_NEG_TTL): return cached[0] res = [] artist = (track["artist"] or "").split(",")[0] try: # fuzzy search (NOT /api/get — that needs exact album+duration and 404s # on remasters/deluxe versions). Fall back to a suffix-stripped title so # "Song - 2011 Remaster" still finds plain "Song". Pick the synced match # closest in length. synced = _lrclib_synced(track["name"], artist) if not synced: clean = _clean_title(track["name"]) if clean != track["name"]: synced = _lrclib_synced(clean, artist) if synced: dur = (track.get("duration_ms") or 0) / 1000 best = min(synced, key=lambda a: abs((a.get("duration") or 0) - dur)) res = _parse_lrc(best["syncedLyrics"]) except Exception: res = [] if len(_lyrics_cache) > 50: _lyrics_cache.clear() _lyrics_cache[key] = (res, time.time()) return res # Throttle Spotify API calls (dev-mode apps rate-limit/429 easily). Hit the API # at most every SP_TTL seconds; between calls, interpolate progress_ms from the # wall clock so lyrics stay synced without polling. _sp_cache = {"track": None, "ts": 0.0, "blocked_until": 0.0} SP_TTL = 5 # throttle between Spotify calls WHILE the tab is open SP_VIEW_MARKER = "/tmp/glasses_spotify_viewed" # serve.py touches when the client fetches the feed SP_FORCE_MARKER = "/tmp/glasses_spotify_force" # serve.py touches when YOU change playback via the glasses def _spotify_viewed_recently(): # only poll Spotify if the client fetched the feed in the last ~12s (i.e. is on the tab) try: return (time.time() - os.path.getmtime(SP_VIEW_MARKER)) < 12 except Exception: return False def _force_refresh_pending(): # a glasses-initiated play/pause/next happened since our last fetch -> refresh now try: return os.path.getmtime(SP_FORCE_MARKER) > _sp_cache["ts"] except Exception: return False def _spotify_track(): now = time.time() if now < _sp_cache["blocked_until"]: return None # in a 429 penalty window — do NOT call (calling extends the ban) if not _spotify_viewed_recently(): return _sp_cache["track"] # nobody's looking — don't call Spotify at all if not _force_refresh_pending() and _sp_cache["ts"] and now - _sp_cache["ts"] < SP_TTL: tr = _sp_cache["track"] if tr and tr.get("is_playing") and tr.get("progress_ms") is not None: tr = dict(tr) tr["progress_ms"] += int((now - _sp_cache["ts"]) * 1000) return tr _sp_cache["ts"] = now try: sys.path.insert(0, str(Path(__file__).parent)) if str(Path(__file__).parent) not in sys.path else None import spotify_lib _sp_cache["track"] = spotify_lib.get_current_track() except Exception as e: resp = getattr(e, "response", None) if resp is not None and getattr(resp, "status_code", None) == 429: try: ra = int(resp.headers.get("Retry-After", "300")) except Exception: ra = 300 _sp_cache["blocked_until"] = now + ra # honor Spotify's cool-off; stop calling return _sp_cache["track"] def spotify_view(): now = time.time() if now < _sp_cache["blocked_until"]: secs = int(_sp_cache["blocked_until"] - now) when = f"~{secs // 3600 + 1}h" if secs >= 3600 else f"~{max(1, secs // 60)} min" return text_view(["Spotify", "-" * 40, "Spotify rate-limited.", f"auto-resumes in {when}.", "(over-polled earlier; fixed)"], SPOTIFY_PRESS) tr = _spotify_track() if not tr: return text_view(["Spotify", "-" * 40, "nothing playing", "", "tap = speak a song to play", "2tap = back"], SPOTIFY_PRESS) lines = [(tr["name"] + " - " + tr["artist"])[:WIDTH], "-" * 40] lyr = _lyrics(tr) if lyr: idx = 0 for i, (ms, _) in enumerate(lyr): if ms <= tr["progress_ms"]: idx = i else: break for i in range(max(0, idx - 2), min(len(lyr), idx + 7)): lines.append(("> " if i == idx else " ") + lyr[i][1][:WIDTH - 2]) else: lines += ["", "(no synced lyrics found)", "", "tap = change song"] return text_view(lines, SPOTIFY_PRESS) # WHOOP module — reads life_manager's whoop_lib. Throttled to 60s; wrapped at # the API boundary so a WHOOP outage shows an error line, not a dead hub. _whoop = {"lines": ["WHOOP", "loading…"], "ts": 0.0} def whoop_lines(): now = time.time() if _whoop["ts"] and now - _whoop["ts"] < 60: return _whoop["lines"] try: sys.path.insert(0, "/data/cameron/life/scripts") if "/data/cameron/life/scripts" not in sys.path else None from whoop_lib import get_latest_summary s = get_latest_summary() f = lambda k, d=None: s.get(k, d) kcal = round(f("kilojoules_today") / 4.184) if f("kilojoules_today") else None _whoop["lines"] = [ "WHOOP", "-" * 40, f"Recovery {round(f('recovery_score'))}%" if f("recovery_score") is not None else "Recovery --", f"Sleep {round(f('sleep_performance_percentage'))}% {f('hours_asleep'):.1f}h" if f("sleep_performance_percentage") is not None else "Sleep --", f"Strain {f('strain_today'):.1f} / 21" if f("strain_today") is not None else "Strain --", f"Calories {kcal} kcal" if kcal else "Calories --", f"HRV {round(f('hrv_rmssd_milli') or 0)}ms RHR {round(f('resting_heart_rate') or 0)}bpm", f"REM {round(f('rem_minutes') or 0)}m Deep {round(f('deep_minutes') or 0)}m", "-" * 40, "steps: n/a workout: needs re-auth", ] except Exception as e: _whoop["lines"] = ["WHOOP", "", "unavailable:", str(e)[:60]] _whoop["ts"] = now return _whoop["lines"] def main(): FEED.mkdir(parents=True, exist_ok=True) while True: write_atomic(FEED / "home.json", list_view( "PARA HUB", [{"label": f"{lbl.ljust(8)} {desc}", "do": do} for lbl, desc, do in MODULES])) for lbl, desc, do in MODULES: feed = do.get("go", "") if feed == "dashboard": write_atomic(FEED / "dashboard.json", dashboard_view()) write_atomic(FEED / "dash_todos.json", dash_todos_view()) write_atomic(FEED / "dash_food.json", dash_food_view()) write_atomic(FEED / "dash_cal.json", dash_cal_view()) write_atomic(FEED / "module_whoop.json", text_view(whoop_lines())) elif feed == "module_whoop": write_atomic(FEED / f"{feed}.json", text_view(whoop_lines())) elif feed == "module_spotify": write_atomic(FEED / f"{feed}.json", spotify_view()) write_atomic(FEED / "spotify_menu.json", spotify_menu_view()) elif feed == "chat_feed": write_atomic(FEED / "chat_feed.json", chat_view()) elif feed == "persian": write_atomic(FEED / "persian.json", persian_view()) elif feed.startswith("module_"): write_atomic(FEED / f"{feed}.json", placeholder(lbl, desc)) roster_items = [] seen = {} for idx, name in session_windows(): seen[name] = seen.get(name, 0) + 1 key = name if seen[name] == 1 else f"{name}{seen[name]}" st = status_word(name) roster_items.append({"label": f"{key[:15].ljust(16)}{st}", "do": {"go": f"agent_{key}"}}) m_lines, _ = mirror(f"{SESSION}:{idx}") # ignore last-response start: tail to bottom by default write_atomic(FEED / f"agent_{key}.json", text_view(m_lines, {"go": f"send_{key}"})) write_atomic(FEED / f"send_{key}.json", send_view(name, key)) write_atomic(FEED / "agents.json", list_view("Agents", roster_items)) time.sleep(PERIOD) if __name__ == "__main__": main()