Spaces:

build-small-hackathon
/

the-committee

Sleeping

The Committee — two cats judge your webcam (local MiniCPM + Llama via llama.cpp)

3481136 verified 5 months ago

9.41 kB

"""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 = """<!doctype html>
<html><head><meta charset="utf-8"><title>pip the hearthwick</title>
<style>body{background:#1A1424;color:#F2E3C6;font-family:monospace;
display:flex;align-items:center;justify-content:center;height:100vh;margin:0}
a{color:#FFD98A}</style></head>
<body><div><h1>pip is still moving in.</h1>
<p>web/index.html not found — the three.js room hasn't been built yet.</p>
<p>meanwhile: <a href="/app">control panel</a> · <a href="/api/state">/api/state</a></p>
</div></body></html>"""


@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.
    )

	"""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 = """<!doctype html>
	<html><head><meta charset="utf-8"><title>pip the hearthwick</title>
	<style>body{background:#1A1424;color:#F2E3C6;font-family:monospace;
	display:flex;align-items:center;justify-content:center;height:100vh;margin:0}
	a{color:#FFD98A}</style></head>
	<body><div><h1>pip is still moving in.</h1>
	<p>web/index.html not found — the three.js room hasn't been built yet.</p>
	<p>meanwhile: <a href="/app">control panel</a> · <a href="/api/state">/api/state</a></p>
	</div></body></html>"""


	@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.
	)