"""
Simple Flask server for the PARA experiment dashboard.
Serves static HTML reports + media with directory listing fallback.
Usage:
python serve.py # localhost:8080
python serve.py --public # + cloudflared tunnel for public URL
python serve.py --port 9090 # custom port
"""
import argparse
import subprocess
import sys
import threading
from pathlib import Path
from flask import Flask, send_from_directory, send_file, redirect, abort, request, Response, make_response, jsonify
import mimetypes
import os
import uuid
from datetime import datetime
REPORTS_DIR = Path(__file__).parent
BROWSE_ROOT = Path("/data/cameron")
app = Flask(__name__)
@app.after_request
def add_range_header(response):
"""Add Accept-Ranges to all responses so browsers know they can seek videos."""
response.headers.setdefault("Accept-Ranges", "bytes")
# CORS: lets the packaged glasses app (which runs from a local origin) call
# the feeds + /api cross-origin. Wildcard is safe here — auth is via a
# password header/param, not cookies.
response.headers.setdefault("Access-Control-Allow-Origin", "*")
response.headers.setdefault("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
response.headers.setdefault("Access-Control-Allow-Headers", "Content-Type, X-Agents-Password")
return response
@app.before_request
def _cors_preflight():
"""Short-circuit CORS preflight requests with the headers above."""
if request.method == "OPTIONS":
return ("", 204)
@app.route("/")
def index():
index_file = REPORTS_DIR / "index.html"
if index_file.exists():
return send_from_directory(REPORTS_DIR, "index.html")
return "
No reports yet
Agents haven't generated any reports.
"
@app.route("/style.css")
def style():
return send_from_directory(REPORTS_DIR, "style.css")
# ---------------------------------------------------------------------------
# Life Dashboard
# ---------------------------------------------------------------------------
@app.route("/life")
def life_dashboard():
life_dir = Path("/data/cameron/life")
if (life_dir / "dashboard.html").exists():
return send_from_directory(life_dir, "dashboard.html")
return "Life dashboard not found.", 404
@app.route("/life/vision/")
def life_vision(filename):
return send_from_directory(Path("/data/cameron/life/vision"), filename)
# ---------------------------------------------------------------------------
# Notes
# ---------------------------------------------------------------------------
_NOTES_PASSWORD = "fabregas"
_NOTES_FILE = Path("/data/cameron/life/notes.md")
def _check_notes_auth():
from flask import request, abort
if request.headers.get("X-Notes-Auth") != _NOTES_PASSWORD:
abort(401)
@app.route("/notes")
def notes_page():
return send_from_directory(Path("/data/cameron/life"), "notes.html")
@app.route("/api/notes", methods=["GET"])
def get_notes():
_check_notes_auth()
if _NOTES_FILE.exists():
return _NOTES_FILE.read_text(), 200, {"Content-Type": "text/plain; charset=utf-8"}
return "", 200
@app.route("/api/notes", methods=["POST"])
def save_notes():
_check_notes_auth()
from flask import request
_NOTES_FILE.write_text(request.get_data(as_text=True))
return "ok", 200
@app.route("/api/annotate", methods=["POST"])
def annotate_rollout():
"""Save one rollout's annotation to `/_annotations.json`.
Body: {"eval_dir": "",
"rollout_id": "rollout_NNN",
"fields": {"success": "yes|partial|no",
"touched_obj": bool,
"failure_mode": "..."|null,
"notes": "..."}}
"""
import json
body = request.get_json(force=True, silent=True) or {}
eval_rel = body.get("eval_dir", "")
roll = body.get("rollout_id", "")
fields = body.get("fields", {}) or {}
if not eval_rel or not roll:
return jsonify({"ok": False, "error": "missing eval_dir / rollout_id"}), 400
eval_path = _check_browse_path(eval_rel)
if eval_path is None or not eval_path.is_dir():
return jsonify({"ok": False, "error": f"eval_dir not allowed/found: {eval_rel}"}), 403
ann_path = eval_path / "_annotations.json"
state: dict = {}
if ann_path.exists():
try:
state = json.loads(ann_path.read_text())
except Exception:
state = {}
state[roll] = fields
try:
ann_path.write_text(json.dumps(state, indent=2, sort_keys=True))
except OSError as e:
return jsonify({"ok": False, "error": str(e)}), 500
return jsonify({"ok": True, "saved": str(ann_path)}), 200
@app.route("/api/annotate", methods=["GET"])
def annotate_get():
"""Return the existing `_annotations.json` for a given eval_dir (browse-rel)."""
import json
eval_rel = request.args.get("eval_dir", "")
if not eval_rel:
return jsonify({}), 200
eval_path = _check_browse_path(eval_rel)
if eval_path is None or not eval_path.is_dir():
return jsonify({}), 200
ann_path = eval_path / "_annotations.json"
if not ann_path.exists():
return jsonify({}), 200
try:
return jsonify(json.loads(ann_path.read_text())), 200
except Exception:
return jsonify({}), 200
# ---------------------------------------------------------------------------
# Chat interface
# ---------------------------------------------------------------------------
@app.route("/567_project")
@app.route("/567_project/")
def school_project(filename="index.html"):
site_dir = Path("/data/cameron/567_augmentation_viewpoint_project/website")
if site_dir.is_dir():
return send_from_directory(site_dir, filename)
return "Not found.", 404
# ---------------------------------------------------------------------------
# Meeting Notes
# ---------------------------------------------------------------------------
NOTES_DIR = REPORTS_DIR / "meeting_notes"
@app.route("/meeting_notes")
@app.route("/meeting_notes/")
def meeting_notes_index():
"""List all meeting notes, newest first."""
notes = []
if NOTES_DIR.is_dir():
for f in sorted(NOTES_DIR.iterdir(), reverse=True):
if f.is_dir() and (f / "notes.md").exists():
md = (f / "notes.md").read_text()
# Extract title from first # heading
title = f.name.replace("-", " ").replace("_", " ")
for line in md.split("\n"):
if line.startswith("# "):
title = line[2:].strip()
break
mtime = datetime.fromtimestamp((f / "notes.md").stat().st_mtime).strftime("%Y-%m-%d %H:%M")
notes.append((f.name, title, mtime))
items = "".join(
f'
"""
@app.route("/upload", methods=["POST"])
def upload_file():
if "file" not in request.files:
return jsonify({"error": "No file provided"}), 400
file = request.files["file"]
name = request.form.get("name", "").strip() or file.filename
# Sanitize filename
import re
safe_name = re.sub(r'[^\w\-\. ]', '_', name)
dest = UPLOAD_DIR / safe_name
file.save(str(dest))
rel = str(dest).replace("/data/cameron/", "")
return jsonify({"path": str(dest), "browse_path": rel})
@app.route("/data_viewer")
@app.route("/data_viewer/")
@app.route("/data_viewer/")
def data_viewer(filename="index.html"):
viewer_dir = REPORTS_DIR / "data_viewer"
if viewer_dir.is_dir():
return send_from_directory(viewer_dir, filename)
return "Data viewer not built yet.", 404
# ---------------------------------------------------------------------------
# Data Viewer JSON API — lists directory contents under BROWSE_ROOT
# ---------------------------------------------------------------------------
_NATURAL_RE = __import__("re").compile(r"(\d+)")
def _natural_key(s):
return [int(t) if t.isdigit() else t.lower() for t in _NATURAL_RE.split(s)]
@app.route("/api/data_viewer/datasets")
def data_viewer_datasets():
cfg = REPORTS_DIR / "data_viewer" / "datasets.json"
if not cfg.exists():
return jsonify({"datasets": []})
import json as _json
return jsonify(_json.loads(cfg.read_text()))
@app.route("/api/data_viewer/annotations", methods=["GET"])
def data_viewer_get_annotations():
"""Load episodes.json (if present) from a frames directory, else return an empty skeleton with frame count."""
import json as _json
rel = request.args.get("path", "").lstrip("/")
full = _check_browse_path(rel)
if full is None or not full.is_dir():
return jsonify({"error": "path not a directory"}), 400
meta_file = full / "episodes.json"
if meta_file.exists():
try:
return jsonify(_json.loads(meta_file.read_text()))
except Exception as e:
return jsonify({"error": f"invalid episodes.json: {e}"}), 500
IMG_EXTS = {".png", ".jpg", ".jpeg", ".webp"}
try:
frames = [p for p in full.iterdir() if p.is_file() and p.suffix.lower() in IMG_EXTS]
except OSError:
frames = []
return jsonify({
"version": 1,
"source_path": rel,
"total_frames": len(frames),
"fps": 4,
"episodes": [],
})
@app.route("/api/data_viewer/annotations", methods=["POST"])
def data_viewer_put_annotations():
"""Persist episodes.json into the frames directory. Body: {path, total_frames, fps, episodes}."""
import json as _json
from datetime import timezone as _tz
data = request.get_json(silent=True) or {}
rel = (data.get("path") or "").lstrip("/")
full = _check_browse_path(rel)
if full is None or not full.is_dir():
return jsonify({"error": "path not a directory"}), 400
eps = data.get("episodes", [])
if not isinstance(eps, list):
return jsonify({"error": "episodes must be a list"}), 400
total_frames = int(data.get("total_frames") or 0)
normalized = []
for i, ep in enumerate(eps):
if not isinstance(ep, dict):
return jsonify({"error": f"episode {i} must be an object"}), 400
try:
start = int(ep["start"]); end = int(ep["end"])
except (KeyError, TypeError, ValueError):
return jsonify({"error": f"episode {i} missing/invalid start/end"}), 400
if start < 0 or end < start:
return jsonify({"error": f"episode {i} invalid range {start}-{end}"}), 400
if total_frames and end >= total_frames:
return jsonify({"error": f"episode {i} end {end} >= total_frames {total_frames}"}), 400
kfs_in = ep.get("keyframes") or []
if not isinstance(kfs_in, list):
return jsonify({"error": f"episode {i} keyframes must be a list"}), 400
kfs_out = []
seen_ids = set()
for j, kf in enumerate(kfs_in):
if not isinstance(kf, dict):
return jsonify({"error": f"episode {i} keyframe {j} must be an object"}), 400
try:
frame = int(kf["frame"])
except (KeyError, TypeError, ValueError):
return jsonify({"error": f"episode {i} keyframe {j} missing/invalid frame"}), 400
if frame < start or frame > end:
return jsonify({"error": f"episode {i} keyframe {j} frame {frame} outside ep range {start}-{end}"}), 400
kf_id = str(kf.get("id") or f"kf_{i}_{j}")
if kf_id in seen_ids:
kf_id = f"{kf_id}_{j}"
seen_ids.add(kf_id)
kfs_out.append({
"id": kf_id,
"frame": frame,
"label": str(kf.get("label") or ""),
})
kfs_out.sort(key=lambda k: k["frame"])
normalized.append({
"id": str(ep.get("id") or f"ep_{i}"),
"start": start,
"end": end,
"label": str(ep.get("label") or ""),
"notes": str(ep.get("notes") or ""),
"keyframes": kfs_out,
})
now = datetime.now(_tz.utc).isoformat(timespec="seconds")
payload = {
"version": 1,
"source_path": rel,
"total_frames": total_frames,
"fps": float(data.get("fps") or 4),
"episodes": normalized,
"updated_at": now,
}
dest = full / "episodes.json"
tmp = full / f".episodes.json.tmp.{os.getpid()}"
try:
tmp.write_text(_json.dumps(payload, indent=2))
tmp.replace(dest)
except OSError as e:
return jsonify({"error": f"write failed: {e}"}), 500
return jsonify({"status": "ok", "path": str(dest), "count": len(normalized), "updated_at": now})
@app.route("/api/data_viewer/list")
def data_viewer_list():
"""List directory contents under BROWSE_ROOT as JSON with natural sort."""
rel = request.args.get("path", "").lstrip("/")
limit = int(request.args.get("limit", "2000"))
offset = int(request.args.get("offset", "0"))
only = request.args.get("only", "") # "dirs", "files", or ""
full = _check_browse_path(rel) if rel else BROWSE_ROOT
if full is None:
return jsonify({"error": "forbidden"}), 403
if not full.exists() or not full.is_dir():
return jsonify({"error": "not a directory", "path": str(full)}), 404
dirs, files = [], []
try:
for item in full.iterdir():
name = item.name
if name.startswith("."):
continue
try:
if item.is_dir():
if only != "files":
dirs.append({"name": name, "type": "dir"})
else:
if only != "dirs":
ext = item.suffix.lower()
files.append({
"name": name,
"type": "file",
"ext": ext,
"size": item.stat().st_size,
})
except OSError:
continue
except PermissionError:
return jsonify({"error": "permission denied"}), 403
dirs.sort(key=lambda e: _natural_key(e["name"]))
files.sort(key=lambda e: _natural_key(e["name"]))
total_dirs = len(dirs)
total_files = len(files)
combined = dirs + files
paged = combined[offset: offset + limit]
return jsonify({
"path": rel,
"total_dirs": total_dirs,
"total_files": total_files,
"offset": offset,
"limit": limit,
"returned": len(paged),
"entries": paged,
})
# ---------------------------------------------------------------------------
# Mock-wandb run viewer (our_wandb agent) — filesystem scan of wandb run dirs.
# Heavy scan/parse logic lives in wandb_runs.py; these are thin JSON+static
# routes mirroring the data_viewer pattern above.
# ---------------------------------------------------------------------------
import wandb_runs as _wandb_runs
@app.route("/runs")
@app.route("/runs/")
@app.route("/runs/")
def wandb_viewer(filename="index.html"):
viewer_dir = REPORTS_DIR / "runs"
if viewer_dir.is_dir():
return send_from_directory(viewer_dir, filename)
return "Runs viewer not built yet.", 404
@app.route("/glasses")
@app.route("/glasses/")
@app.route("/glasses/")
def glasses_app(filename="index.html"):
# Even Realities G2 fleet-hub app (built Vite bundle). Served same-origin
# with /api/agents/* so the glasses WebView needs no CORS/proxy.
app_dir = REPORTS_DIR / "glasses_app"
if app_dir.is_dir():
return send_from_directory(app_dir, filename)
return "Glasses app not built yet.", 404
@app.route("/api/runs/list")
def api_runs_list():
return jsonify(_wandb_runs.list_runs())
@app.route("/api/runs/")
def api_runs_detail(run_id):
detail = _wandb_runs.run_detail(run_id)
if detail is None:
return jsonify({"error": "run not found"}), 404
return jsonify(detail)
@app.route("/api/runs//media/")
def api_runs_media(run_id, filename):
# Resolve the run dir via the same scan used everywhere else, then confirm it
# sits under a configured WANDB_ROOTS entry before serving. send_from_directory
# rejects `..`/absolute paths within `filename`; the root check blocks escaping
# `run_id` itself out of the FUSE mounts (see website_builder's note + _check_browse_path).
run_dir = _wandb_runs.find_run(run_id)
if run_dir is None:
abort(404)
resolved = run_dir.resolve()
roots = [Path(r).resolve() for r in _wandb_runs.get_roots()]
if not any(str(resolved).startswith(str(r) + "/") for r in roots):
abort(403)
return send_from_directory(run_dir / "files", filename)
@app.route("/figma")
@app.route("/figma/")
@app.route("/figma/")
def figma_exports(filename=None):
figma_dir = Path("/data/cameron/para/paper/figs/figma")
if not figma_dir.is_dir():
return "
No Figma exports yet
"
if filename:
return send_from_directory(figma_dir, filename)
files = sorted(figma_dir.iterdir())
items = "".join(
f'
"""
@app.route("/paper/")
def paper_file(filename):
paper_dir = Path("/data/cameron/para/paper")
if paper_dir.is_dir():
return send_from_directory(paper_dir, filename)
return "Not found.", 404
@app.route("/old_para_website")
def old_para_website_redirect():
return redirect("/old_para_website/")
@app.route("/old_para_website/")
@app.route("/old_para_website/")
def old_para_website(filename="index_goal.html"):
site_dir = Path("/data/cameron/para/old_para_website")
if site_dir.is_dir():
return send_from_directory(site_dir, filename)
return "Not found.", 404
@app.route("/para_website")
@app.route("/para_website/")
def para_website(filename="index.html"):
site_dir = REPORTS_DIR / "project_site"
if site_dir.is_dir():
return send_from_directory(site_dir, filename)
return "Project website not built yet.", 404
@app.route("/para_website_v1")
@app.route("/para_website_v1/")
def para_website_v1(filename="index.html"):
site_dir = REPORTS_DIR / "project_site_v1"
if site_dir.is_dir():
return send_from_directory(site_dir, filename)
return "Archived v1 site not found.", 404
@app.route("/chat")
def chat_page():
return MOBILE_CHATTER_HTML
@app.route("/api/mobile_chatter/iframe_url")
def mobile_chatter_iframe_url():
url_file = Path("/data/cameron/para/.agents/mobile_chatter/iframe_url.txt")
if url_file.exists():
return url_file.read_text().strip()
return "/"
@app.route("/api/chat", methods=["POST"])
def api_chat():
from chat import chat as do_chat
data = request.get_json()
user_msg = data.get("message", "")
session_id = data.get("session_id", str(uuid.uuid4()))
if not user_msg:
return jsonify({"error": "No message provided"}), 400
model = data.get("model", "claude-sonnet-4-20250514")
try:
response_text = do_chat(session_id, user_msg, model=model)
return jsonify({"response": response_text, "session_id": session_id})
except Exception as e:
return jsonify({"error": str(e)}), 500
# ---------------------------------------------------------------------------
# Agent terminal interface
# ---------------------------------------------------------------------------
AGENTS_PASSWORD = "fabregas"
def _check_agents_auth():
"""Check password from header or query param."""
pw = request.headers.get("X-Agents-Password") or request.args.get("pw")
return pw == AGENTS_PASSWORD
@app.route("/agents")
@app.route("/agents/")
def agents_page(agent_name=None):
return AGENTS_HTML
@app.route("/api/agents/list")
def api_agents_list():
"""List all agents with their tmux window status."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
import yaml
config_path = REPORTS_DIR.parent / "config.yaml"
if not config_path.exists():
return jsonify([])
with open(config_path) as f:
config = yaml.safe_load(f)
agents = []
for name, info in config.get("agents", {}).items():
status_file = REPORTS_DIR.parent / name / "status.md"
status = status_file.read_text().strip() if status_file.exists() else "unknown"
agents.append({"name": name, "description": info.get("description", ""), "status": status})
return jsonify(agents)
@app.route("/api/agents//pane")
def api_agent_pane(agent_name):
"""Capture tmux pane output for an agent."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
lines = request.args.get("lines", 200, type=int)
target = _find_tmux_window(agent_name)
if not target:
return jsonify({"error": f"Window '{agent_name}' not found", "output": ""})
result = subprocess.run(
["tmux", "capture-pane", "-t", target, "-p", "-S", f"-{lines}"],
capture_output=True, text=True,
)
# Strip ANSI escape codes
import re
clean = re.sub(r'\x1b\[[0-9;]*[a-zA-Z]', '', result.stdout)
clean = re.sub(r'\x1b\].*?\x07', '', clean) # OSC sequences
return jsonify({"output": clean})
@app.route("/api/agents//interrupt", methods=["POST"])
def api_agent_interrupt(agent_name):
"""Send Escape key to interrupt Claude Code."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
target = _find_tmux_window(agent_name)
if not target:
return jsonify({"error": f"Window '{agent_name}' not found"}), 404
subprocess.run(["tmux", "send-keys", "-t", target, "Escape"])
# Small delay then press 'i' to return to INSERT mode (in case vim mode is on)
import time as _t; _t.sleep(0.3)
subprocess.run(["tmux", "send-keys", "-t", target, "i"])
return jsonify({"status": "interrupted"})
@app.route("/api/agents//send", methods=["POST"])
def api_agent_send(agent_name):
"""Send keystrokes to an agent's tmux pane."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json()
text = data.get("text", "")
if not text:
return jsonify({"error": "No text provided"}), 400
target = _find_tmux_window(agent_name)
if not target:
return jsonify({"error": f"Window '{agent_name}' not found"}), 404
subprocess.run(["tmux", "send-keys", "-t", target, text, "Enter"])
import time as _time; _time.sleep(1)
subprocess.run(["tmux", "send-keys", "-t", target, "Enter"])
return jsonify({"status": "sent"})
def _spotify_lib():
import sys as _sys
gl = "/data/cameron/agents_stuff/agents/glasses"
if gl not in _sys.path:
_sys.path.insert(0, gl)
import spotify_lib
return spotify_lib
def _sp_force_refresh():
"""Signal the renderer to refresh now-playing immediately (glasses-initiated change)."""
try:
import pathlib, time as _t
pathlib.Path("/tmp/glasses_spotify_force").write_text(str(_t.time()))
except Exception:
pass
@app.route("/api/glasses/spotify/playpause", methods=["POST"])
def api_glasses_spotify_playpause():
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
code = _spotify_lib().toggle_playback()
_sp_force_refresh()
return jsonify({"code": code})
@app.route("/api/glasses/spotify/next", methods=["POST"])
def api_glasses_spotify_next():
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
code = _spotify_lib().next_track()
_sp_force_refresh()
return jsonify({"code": code})
def _sp_find_tool(name, inp, lib):
import json as _json
try:
if name == "search_spotify":
return _json.dumps(lib.search_tracks(inp.get("query", "")))
if name == "artist_albums":
return _json.dumps(lib.artist_albums(inp.get("artist", "")))
if name == "album_tracks":
return _json.dumps(lib.album_tracks(inp.get("album_id", "")))
except Exception as e:
return "tool error: " + str(e)[:120]
return "unknown tool"
@app.route("/api/glasses/spotify/find", methods=["POST"])
def api_glasses_spotify_find():
"""Pick the track to play for a spoken request, grounded in Spotify's LIVE
catalog via tools (so 'newest album' is current, not guessed). Returns the
candidate WITHOUT playing it — the glasses confirm first."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
text = (request.get_json() or {}).get("text", "").strip()
if not text:
return jsonify({"error": "No text provided"}), 400
lib = _spotify_lib()
import json as _json
import re as _re
tools = [
{"name": "search_spotify", "description": "Search Spotify for tracks. Returns up to 5 {name,artist,album,uri}.",
"input_schema": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}},
{"name": "artist_albums", "description": "List an artist's albums/singles NEWEST-FIRST with release_date. Use this for 'newest/latest album' requests — it reflects Spotify's live catalog.",
"input_schema": {"type": "object", "properties": {"artist": {"type": "string"}}, "required": ["artist"]}},
{"name": "album_tracks", "description": "List an album's tracks {name,uri} given an album id from artist_albums.",
"input_schema": {"type": "object", "properties": {"album_id": {"type": "string"}}, "required": ["album_id"]}},
]
system = (
"Pick the single Spotify track to play for the spoken request. ALWAYS ground your answer in the tool "
"results (Spotify's live catalog) — never rely on your own memory of what albums/songs exist or are newest. "
"For 'newest/latest album by X' call artist_albums (newest-first) and use the top album, then album_tracks. "
"For a named song use search_spotify. When you've decided, reply with ONLY JSON: "
'{"uri":"spotify:track:..","name":"..","artist":".."} — or {"found":false} if nothing fits.')
messages = [{"role": "user", "content": text}]
try:
import anthropic
client = anthropic.Anthropic()
for _ in range(8):
r = client.messages.create(model="claude-sonnet-4-6", max_tokens=800, system=system, tools=tools, messages=messages)
if r.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": r.content})
messages.append({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": b.id, "content": _sp_find_tool(b.name, b.input, lib)}
for b in r.content if b.type == "tool_use"]})
continue
txt = "".join(b.text for b in r.content if b.type == "text")
m = _re.search(r"\{.*\}", txt, _re.S)
d = _json.loads(m.group(0)) if m else {}
if d.get("uri"):
return jsonify({"found": True, "uri": d["uri"], "name": d.get("name", ""), "artist": d.get("artist", "")})
return jsonify({"found": False})
except Exception as e:
# fall back to a plain search so voice-search still works if the agentic path fails
t = lib.search(text)
return jsonify({"found": True, **t}) if t else jsonify({"found": False, "error": str(e)[:120]})
return jsonify({"found": False})
@app.route("/api/glasses/spotify/play_uri", methods=["POST"])
def api_glasses_spotify_play_uri():
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
uri = (request.get_json() or {}).get("uri", "").strip()
if not uri:
return jsonify({"error": "No uri provided"}), 400
code = _spotify_lib().play_uri(uri)
_sp_force_refresh()
return jsonify({"code": code})
@app.route("/browse/para/.agents/reports/glasses_app/feed/module_spotify.json")
def glasses_spotify_feed():
"""The client fetching this feed == it's on the Spotify tab. Touch a marker
so the renderer only polls Spotify while it's actually being viewed (avoids
24/7 polling -> rate-limit bans). Then serve the renderer-written file."""
try:
import pathlib, time as _t
pathlib.Path("/tmp/glasses_spotify_viewed").write_text(str(_t.time()))
except Exception:
pass
return send_from_directory("/data/cameron/para/.agents/reports/glasses_app/feed", "module_spotify.json")
GLASSES_CHAT_LOG = "/data/cameron/agents_stuff/agents/glasses/chat_log.md"
VOICE_AGENT = "voice_interface" # persistent voice-interface agent (tmux tab 0)
@app.route("/api/glasses/chat/send", methods=["POST"])
def api_glasses_chat_send():
"""Glasses Chat -> the persistent voice_interface agent. It observes/commands
other agents per its own ROLE, then appends its reply to the chat log, which
the glasses poll (so the reply shows up whenever it's done)."""
if not _check_agents_auth():
return jsonify({"error": "unauthorized"}), 401
text = (request.get_json() or {}).get("text", "").strip()
if not text:
return jsonify({"error": "No text provided"}), 400
with open(GLASSES_CHAT_LOG, "a") as f:
f.write("> you: " + text + "\n")
target = _find_tmux_window(VOICE_AGENT)
if not target:
with open(GLASSES_CHAT_LOG, "a") as f:
f.write(f"voice: (agent '{VOICE_AGENT}' window not found)\n")
return jsonify({"error": "voice agent not found"}), 404
# ONE line (embedded newlines put Claude Code into multi-line mode and it won't submit on Enter).
wrapped = (f"{text} [glasses chat from Cameron -- when done, append ONE line "
f"'voice: ' to {GLASSES_CHAT_LOG}; observe agents by reading "
f"their panes, only message one if Cameron explicitly asks]")
subprocess.run(["tmux", "send-keys", "-t", target, wrapped, "Enter"])
import time as _t; _t.sleep(1)
subprocess.run(["tmux", "send-keys", "-t", target, "Enter"])
return jsonify({"status": "dispatched"})
def _find_tmux_window(name):
"""Find tmux window target by name."""
result = subprocess.run(
["tmux", "list-windows", "-a", "-F", "#{session_name}:#{window_index} #{window_name}"],
capture_output=True, text=True,
)
for line in result.stdout.strip().split("\n"):
parts = line.split(" ", 1)
if len(parts) == 2 and parts[1] == name:
return parts[0]
return None
AGENTS_HTML = """
PARA Agents
"""
# Whisper model (lazy-loaded)
_whisper_model = None
def _get_whisper():
global _whisper_model
if _whisper_model is None:
from faster_whisper import WhisperModel
_whisper_model = WhisperModel("small", device="cpu", compute_type="int8")
return _whisper_model
@app.route("/api/transcribe", methods=["POST"])
def api_transcribe():
"""Transcribe audio using Whisper."""
import tempfile
if "audio" not in request.files:
return jsonify({"error": "No audio file"}), 400
audio = request.files["audio"]
tmp = tempfile.NamedTemporaryFile(suffix=".webm", delete=False)
audio.save(tmp.name)
tmp.close()
try:
if os.path.getsize(tmp.name) < 1024:
return jsonify({"text": "", "warning": "recording too short or empty"})
model = _get_whisper()
segments, _ = model.transcribe(tmp.name, language="en")
text = " ".join(seg.text.strip() for seg in segments)
return jsonify({"text": text})
except Exception as e:
return jsonify({"error": str(e)}), 500
finally:
os.unlink(tmp.name)
@app.route("/api/tts", methods=["POST"])
def api_tts():
"""Convert text to speech using Piper neural TTS."""
import tempfile
data = request.get_json()
text = data.get("text", "")
if not text:
return jsonify({"error": "No text provided"}), 400
# Truncate very long text to ~2000 chars to avoid timeouts
if len(text) > 2000:
text = text[:2000] + "... text truncated for speech."
try:
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp.close()
result = subprocess.run(
["python", "-m", "piper",
"--model", "/data/cameron/.piper_voices/en_US-ryan-high.onnx",
"--output_file", tmp.name],
input=text, capture_output=True, text=True, timeout=120,
)
if result.returncode != 0:
return jsonify({"error": result.stderr}), 500
return send_file(tmp.name, mimetype="audio/wav")
except subprocess.TimeoutExpired:
return jsonify({"error": "TTS timed out — text too long"}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/api/chat/clear", methods=["POST"])
def api_chat_clear():
from chat import clear_conversation
data = request.get_json()
session_id = data.get("session_id", "")
clear_conversation(session_id)
return jsonify({"status": "cleared"})
MOBILE_CHATTER_HTML = """
PARA Mobile
"""
def _human_size(nbytes: int) -> str:
for unit in ("B", "KB", "MB", "GB"):
if nbytes < 1024:
return f"{nbytes:.1f} {unit}" if unit != "B" else f"{nbytes} B"
nbytes /= 1024
return f"{nbytes:.1f} TB"
def _breadcrumb(rel_path: str) -> str:
parts = rel_path.strip("/").split("/") if rel_path.strip("/") else []
crumbs = ['/data/cameron']
for i, part in enumerate(parts):
link = "/browse/" + "/".join(parts[: i + 1]) + "/"
crumbs.append(f'{part}')
return '"
_BROWSE_ALLOWED_ROOTS = [
Path("/data/cameron").resolve(),
Path("/data/libero").resolve(),
# SSHFS mount of robot-lab via the school→dev (Tailscale) ProxyJump.
# Exposed at /browse_yam and via the /data/cameron/yam_remote symlink.
Path("/home/cameronsmith/mnt/robot-lab"),
# SSHFS mount of yukon (rig host on tailscale).
# Exposed via the /data/cameron/yukon_remote symlink → /browse/yukon_remote.
Path("/home/cameronsmith/mnt/yukon"),
]
def _check_browse_path(subpath):
"""Resolve `subpath` under BROWSE_ROOT, rejecting `..` traversal.
Symlinks that resolve into any allowlisted root (e.g. /data/cameron/libero -> /data/libero) are accepted."""
if ".." in Path(subpath).parts:
return None
candidate = BROWSE_ROOT / subpath
try:
resolved = candidate.resolve()
except Exception:
return None
for root in _BROWSE_ALLOWED_ROOTS:
if str(resolved) == str(root) or str(resolved).startswith(str(root) + "/"):
return resolved
return None
@app.route("/browse_yam")
@app.route("/browse_yam/")
def browse_yam():
"""Land on the YAM calibration QA folder served through the SSHFS mount.
The mount is school-server → dev (Tailscale ProxyJump) → robot-lab. If puget
is offline, the mount stalls. Triggers a stat() to surface that quickly.
"""
mount = Path("/home/cameronsmith/mnt/robot-lab")
try:
# Cheap probe: list the dir. If sshfs is dead this returns an OS error fast.
list(mount.iterdir())
except OSError as e:
return (f"
YAM mount unavailable
"
f"
SSHFS to robot-lab (via puget) returned: {e}
"
f"
On the school server run: "
f"fusermount -u /home/cameronsmith/mnt/robot-lab && "
f"sshfs robot-lab:/home/robot-lab /home/cameronsmith/mnt/robot-lab "
f"-o reconnect,ServerAliveInterval=30
"), 503
return redirect("/browse/yam_remote/cameron/yam_overlay/calibration_qa/")
@app.route("/browse/")
@app.route("/browse/")
def browse(subpath=""):
full_path = _check_browse_path(subpath)
if full_path is None:
abort(403)
if not full_path.exists():
abort(404)
# Serve a file directly
if full_path.is_file():
ext = full_path.suffix.lower()
# For video/image files accessed directly in browser, show a player page
# Use ?raw to get the actual bytes (used by