app.py

"""Asterism Relay — a first-person explorable cosmos for the Build Small hackathon.

- Custom Three.js / WebGL frontend served from gr.Server (Off-Brand).
- Backend deterministically lays out a universe of real-phenomenon bodies.
- As the player approaches each body, MiniCPM4.1-8B (OpenBMB) running on Modal
  AUTHORS that sector: names the body, selects which curated facts to surface,
  and writes a fragment of a vanished civilization's final transmissions.
- The FACT layer is composed from a curated local knowledge base, so the science
  is always accurate; the FICTION layer is freely authored by the model.
"""

import json
import os
import random
from functools import lru_cache
from pathlib import Path
from typing import Any

import httpx
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from gradio import Server

ROOT = Path(__file__).parent
DATA = ROOT / "data"
STATIC = ROOT / "static"

# Deployed Modal endpoint (scripts/minicpm_modal.py). Set in the Space secrets.
MODAL_AUTHOR_URL = os.environ.get("MODAL_AUTHOR_URL", "").strip()

# Universe layout parameters.
NUM_BODIES = 26
UNIVERSE_SEED = 70707
SPAN_XZ = 7000
SPAN_Y = 1400
MIN_DISTANCE = 1500

app = Server()
app.mount("/static", StaticFiles(directory=STATIC), name="static")
app.mount("/data", StaticFiles(directory=DATA), name="data")


@lru_cache(maxsize=1)
def load_facts() -> list[dict[str, Any]]:
    return json.loads((DATA / "astro_facts.json").read_text(encoding="utf-8"))


@lru_cache(maxsize=1)
def build_universe() -> dict[str, Any]:
    """Deterministic body field. Each body is assigned a real phenomenon; appearance
    is drawn instantly in-browser from its visual cues, content is authored on approach."""
    facts = load_facts()
    rng = random.Random(UNIVERSE_SEED)

    # A shuffled bag so every phenomenon appears, roughly evenly.
    bag: list[dict[str, Any]] = []
    while len(bag) < NUM_BODIES:
        order = facts[:]
        rng.shuffle(order)
        bag.extend(order)
    bag = bag[:NUM_BODIES]

    bodies: list[dict[str, Any]] = []
    placed: list[tuple[float, float, float]] = []
    for i, concept in enumerate(bag):
        for _ in range(200):
            x = rng.uniform(-SPAN_XZ, SPAN_XZ)
            y = rng.uniform(-SPAN_Y, SPAN_Y)
            z = rng.uniform(-SPAN_XZ, SPAN_XZ)
            if all((x - px) ** 2 + (y - py) ** 2 + (z - pz) ** 2 > MIN_DISTANCE ** 2 for px, py, pz in placed):
                break
        placed.append((x, y, z))
        vc = concept["visual_cues"]
        bodies.append({
            "id": f"body-{i:02d}",
            "concept_id": concept["id"],
            "name": concept["name"],
            "body_type": concept["body_type"],
            "category": concept.get("category", "star"),
            "position": {"x": round(x, 1), "y": round(y, 1), "z": round(z, 1)},
            "visual": vc,
        })

    return {
        "seed": UNIVERSE_SEED,
        "intro": (
            "Long before you, the Last Cartographers mapped this whole sky — then fell silent. "
            "Their final transmissions are scattered across real stars and ruins. Fly out and read them."
        ),
        "bodies": bodies,
    }


def compose_fallback(body: dict[str, Any]) -> dict[str, Any]:
    """Deterministic authoring used when the model endpoint is cold or unavailable —
    the app never stalls. Fact layer from curated facts; fiction from curated fragments."""
    facts = {f["id"]: f for f in load_facts()}
    concept = facts[body["concept_id"]]
    rng = random.Random(f"{body['id']}:{concept['id']}")
    ids = sorted(rng.sample(range(len(concept["facts"])), k=min(3, len(concept["facts"]))))
    explanation = " ".join([concept["one_line"]] + [concept["facts"][i] for i in ids])
    vc = concept["visual_cues"]
    return {
        "sector_key": body["id"],
        "authored_by": "curated-fallback",
        "body": {
            "id": f"{concept['id']}-{rng.randrange(100, 999)}",
            "name": concept["name"],
            "type": concept["body_type"],
            "category": concept.get("category", "star"),
            "phenomenon": concept["name"],
            "position": body["position"],
        },
        "fact_layer": {
            "source": "curated_local_knowledge_base",
            "concept_id": concept["id"],
            "title": concept["name"],
            "fact_ids": ids,
            "explanation": explanation,
        },
        "fiction_layer": {
            "civilization": "the Last Cartographers",
            "fragment_id": f"transmission-{rng.randrange(1, 99):02d}",
            "transmission": rng.choice(concept["transmissions"]),
        },
        "shader": {
            "seed": rng.randrange(100000, 9999999),
            "render": vc["render"],
            "primary": vc["primary_color"],
            "secondary": vc["secondary_color"],
            "emissive": vc["emissive"],
            "radius": vc["radius"],
            "beam": vc.get("beam", False),
            "corona": vc.get("corona", False),
            "rings": vc.get("rings", False),
            "jets": vc.get("jets", False),
            "particles": vc.get("particles", vc["secondary_color"]),
            "noise_scale": round(1.8 + rng.random() * 3.0, 3),
            "atmosphere": round(0.3 + rng.random() * 0.5, 3),
        },
    }


@lru_cache(maxsize=256)
def author_sector(body_id: str) -> str:
    """Author one sector via the Modal MiniCPM endpoint, falling back to curated
    composition. Cached so each body is authored once per process."""
    universe = build_universe()
    body = next((b for b in universe["bodies"] if b["id"] == body_id), None)
    if body is None:
        return json.dumps({"error": "unknown body"})

    if MODAL_AUTHOR_URL:
        pos = body["position"]
        params = {
            "sector_key": body_id,
            "position": f"{pos['x']},{pos['y']},{pos['z']}",
            "concept_id": body["concept_id"],
        }
        try:
            resp = httpx.get(MODAL_AUTHOR_URL, params=params, timeout=25.0, follow_redirects=True)
            resp.raise_for_status()
            return json.dumps(resp.json())
        except Exception:
            pass  # fall through to deterministic composition

    return json.dumps(compose_fallback(body))


@app.get("/api/universe")
async def universe_manifest():
    return JSONResponse(build_universe())


@app.get("/api/sector/{body_id}")
async def sector_endpoint(body_id: str):
    return JSONResponse(json.loads(author_sector(body_id)))


@app.get("/", response_class=HTMLResponse)
async def homepage():
    return HTMLResponse((STATIC / "index.html").read_text(encoding="utf-8"))


# Hugging Face Spaces (gradio SDK) imports this module and launches a top-level
# object named `demo`; it does not run the __main__ block below.
demo = app

if __name__ == "__main__":
    app.launch(server_name="0.0.0.0", server_port=7860)