"""server.py — Living Pixel Pet ("Pip the Hearthwick") server. One FastAPI app: GET / -> web/index.html (the three.js world) /web/* -> static frontend files GET /api/state -> {light, sound, live, mood, screen, thinking, diary} POST /api/sense -> {light, sound} (browser pushes real readings ~1Hz; -1 = not live) POST /api/poke -> {action: "write"} (force a diary entry + reaction now) /app -> Gradio control panel (status + diary feed + write-now button) Run locally: python server.py (or: uvicorn server:app --port 7860) HF Spaces: sdk: docker, app_port: 7860, CMD ["python", "server.py"] Verified against gradio 6.17.3. Keep ONE uvicorn worker (in-process Gradio queue). """ from __future__ import annotations import os import threading import time from pathlib import Path import gradio as gr import uvicorn from fastapi import FastAPI from fastapi.responses import FileResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from companion.runner import CompanionRunner ROOT = Path(__file__).parent WEB = ROOT / "web" # three.js bundle (never mount at /assets — gradio-internal name) MOODS = {"sleepy", "cozy", "alert", "annoyed", "happy", "idle"} DIARY_CAP = 12 # --------------------------------------------------------------------------- # Persona + runner # --------------------------------------------------------------------------- def _load_persona(): """Prefer the real pet persona (B3); fall back so the server always boots.""" try: from companion.pet_persona import PetPersona # documented name return PetPersona() except Exception: pass try: from companion.pet_persona import PipPersona # spec name return PipPersona() except Exception: pass # Last resort: a thin shim over the proven DeskPersona so /api/state, # diary generation and the Gradio panel all still work end-to-end. from companion.personas import DeskPersona class _FallbackPetPersona(DeskPersona): key = "pip-fallback" title = "pip (fallback)" produces = ("entry", "screen") cooldown_s = 45.0 def mood(self, r) -> str: if r is None: return "idle" if r.sound > 70: return "annoyed" if r.light < 20 and r.sound < 25: return "sleepy" if 20 <= r.light <= 60 and r.sound < 30: return "cozy" if r.light > 60 and 30 <= r.sound <= 60: return "happy" return "idle" return _FallbackPetPersona() persona = _load_persona() runner = CompanionRunner(persona, mock_mode="day", browser=True) def _clamp(v) -> int: try: return max(0, min(100, int(v))) except (TypeError, ValueError): return 0 def _current_mood() -> str: """Single source of truth is persona.engine.mood (B3's MoodEngine); fall back to a persona.mood(reading) method, then to "idle".""" mood = None engine = getattr(persona, "engine", None) if engine is not None: mood = getattr(engine, "mood", None) if mood not in MOODS: mood_fn = getattr(persona, "mood", None) if callable(mood_fn): try: mood = mood_fn(runner.latest()) except Exception: mood = None return mood if mood in MOODS else "idle" def _diary() -> list[dict]: """Newest-first diary, capped. Trigger tags ("dark"/"noise"/"return"/ "poke"/...) are stamped at the source by runner.generate(label=...).""" items = runner.feed() return [ { "clock": str(item.get("clock", "")), "entry": str(item.get("entry", "")), "trigger": str(item.get("trigger", "auto")), } for item in items[:DIARY_CAP] ] def _poke_now() -> None: """Force a diary entry + reaction: nudge the mood engine, then generate. The "poke" tag is stamped at the source; `thinking` is derived from the runner's generation lock, so there's nothing to toggle here.""" engine = getattr(persona, "engine", None) if engine is not None and hasattr(engine, "poke"): try: engine.poke(time.time()) except Exception: pass runner.generate(auto=False, label="poke") # --------------------------------------------------------------------------- # 1. Parent FastAPI app (custom routes registered BEFORE the gradio mount) # --------------------------------------------------------------------------- app = FastAPI(docs_url=None, redoc_url=None) class SenseBody(BaseModel): light: float = -1 sound: float = -1 class PokeBody(BaseModel): action: str = "write" @app.get("/api/state") def api_state() -> dict: r = runner.latest() return { "light": _clamp(r.light) if r is not None else 0, "sound": _clamp(r.sound) if r is not None else 0, "live": bool(getattr(runner.source, "live", False)), "mood": _current_mood(), "screen": runner.get("screen"), # True whenever ANY generation (auto, poke, panel) is in flight. "thinking": runner.gen_lock.locked(), "diary": _diary(), } @app.post("/api/sense") def api_sense(body: SenseBody) -> dict: # -1 means "not live"; BrowserSensorSource.ingest ignores negatives and # the runner keeps reading the mock, so we just pass everything through. runner.ingest_browser(int(body.light), int(body.sound)) return {"ok": True, "live": bool(getattr(runner.source, "live", False))} @app.post("/api/poke") def api_poke(body: PokeBody) -> dict: # Fire-and-forget: the browser polls /api/state and watches `thinking`. threading.Thread(target=_poke_now, daemon=True).start() return {"ok": True, "action": body.action} @app.get("/api/health") def api_health() -> dict: return {"ok": True} # --------------------------------------------------------------------------- # 2. three.js frontend at / and /web/* (defensive: boots even if web/ absent) # --------------------------------------------------------------------------- # check_dir=False -> clean 404s instead of a crash when web/ isn't built yet. app.mount("/web", StaticFiles(directory=str(WEB), check_dir=False), name="web") _PLACEHOLDER = """ pip the hearthwick

pip is still moving in.

web/index.html not found — the three.js room hasn't been built yet.

meanwhile: control panel · /api/state

""" @app.get("/") def index(): page = WEB / "index.html" if page.is_file(): return FileResponse(page) return HTMLResponse(_PLACEHOLDER, status_code=200) # --------------------------------------------------------------------------- # 3. Gradio control panel (the hackathon-required functional Gradio app) # --------------------------------------------------------------------------- def _panel_status() -> str: r = runner.latest() sensors = (f"**light** {_clamp(r.light)}% · **sound** {_clamp(r.sound)}%" if r is not None else "warming up…") return (f"{runner.status_md()} · **mood** `{_current_mood()}` · {sensors}" f"{' · ✍️ scribbling…' if runner.gen_lock.locked() else ''}") def _panel_feed() -> list[list[str]]: return [[d["clock"], d["trigger"], d["entry"]] for d in _diary()] def _panel_refresh(): return _panel_status(), _panel_feed() def _panel_poke(): _poke_now() # synchronous in the Gradio handler thread — fine return _panel_refresh() with gr.Blocks(title="pip — control panel") as demo: gr.Markdown("# 🕯️ pip the hearthwick — control panel\n" "the room lives at [/](/) · this panel watches the engine.") status_md = gr.Markdown(_panel_status()) feed_df = gr.Dataframe( headers=["clock", "trigger", "entry"], value=_panel_feed(), label="diary feed (newest first)", interactive=False, ) poke_btn = gr.Button("write now (poke pip)", variant="primary") poke_btn.click(_panel_poke, None, [status_md, feed_df], api_name="poke") gr.Timer(2.0).tick(_panel_refresh, None, [status_md, feed_df]) demo.queue(max_size=32) # --------------------------------------------------------------------------- # 4. Mount Gradio LAST. Never call demo.launch(). # --------------------------------------------------------------------------- app = gr.mount_gradio_app( app, demo, path="/app", ssr_mode=False, # no Node in the container footer_links=[], # hide gradio footer # root_path stays unset: mount path is auto-detected; HF Spaces edge is # handled by uvicorn proxy_headers below. ) if __name__ == "__main__": uvicorn.run( app, host="0.0.0.0", port=int(os.environ.get("PORT", "7860")), proxy_headers=True, # honour X-Forwarded-Proto on HF Spaces forwarded_allow_ips="*", # workers stays 1 (default) — the gradio queue is in-process. )