app.py

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

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

try:
    import spaces
except ImportError:  # Local development outside Hugging Face ZeroGPU.
    class _Spaces:
        @staticmethod
        def GPU(*args: Any, **kwargs: Any):
            if args and callable(args[0]) and not kwargs:
                return args[0]

            def decorator(fn):
                return fn

            return decorator

    spaces = _Spaces()


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

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=128)
def generate_sector_cached(sector_key: str, position_key: str) -> dict[str, Any]:
    facts = load_facts()
    rng = random.Random(f"{sector_key}:{position_key}:asterism-relay")
    concept = rng.choice(facts)
    serial = rng.randrange(100, 999)
    fragment_no = rng.randrange(2, 8)
    hue_shift = rng.random()

    selected_facts = rng.sample(concept["facts"], k=min(3, len(concept["facts"])))
    explanation = " ".join([concept["one_line"], *selected_facts])

    names = [
        "Lacuna",
        "Vesper",
        "Aster",
        "Noctil",
        "Mirrorglass",
        "Far Choir",
        "Horizon Loom",
    ]
    civilization_terms = [
        "the Archivists",
        "the Choir of Measures",
        "the Lantern Kin",
        "the Last Cartographers",
    ]

    return {
        "sector_key": sector_key,
        "body": {
            "id": f"{concept['id']}-{serial}",
            "name": f"{rng.choice(names)} {serial}",
            "type": concept["body_type"],
            "phenomenon": concept["name"],
            "position": parse_position(position_key),
        },
        "fact_layer": {
            "source": "curated_local_knowledge_base",
            "concept_id": concept["id"],
            "title": concept["name"],
            "explanation": explanation,
            "facts": selected_facts,
        },
        "fiction_layer": {
            "civilization": rng.choice(civilization_terms),
            "fragment_id": f"transmission-{fragment_no:02d}",
            "transmission": build_transmission(concept["id"], rng),
        },
        "shader": {
            "seed": rng.randrange(1_000_000, 9_999_999),
            "primary": concept["visual_cues"]["primary_color"],
            "secondary": concept["visual_cues"]["secondary_color"],
            "emissive": concept["visual_cues"]["emissive"],
            "beam": concept["visual_cues"]["beam"],
            "noise_scale": round(1.8 + hue_shift * 3.4, 3),
            "atmosphere": round(0.25 + rng.random() * 0.5, 3),
        },
    }


def parse_position(position_key: str) -> dict[str, float]:
    try:
        x, y, z = [float(part) for part in position_key.split(",")]
    except ValueError:
        x, y, z = 0.0, 20.0, -450.0
    return {"x": x, "y": y, "z": z}


def build_transmission(concept_id: str, rng: random.Random) -> str:
    fragments = {
        "pulsar": [
            "We counted the beacon until the count began answering back.",
            "Their clock survived the cities; follow the pulse, not the dust.",
        ],
        "red_giant": [
            "The old suns taught us expansion is not escape, only a slower goodbye.",
            "When the red light filled our windows, every archive learned to whisper.",
        ],
        "white_dwarf": [
            "A small ember can remember a whole star if the dark is patient enough.",
            "We hid the index in the cooling remnant, where haste cannot follow.",
        ],
        "accretion_disk": [
            "The dark center is silent. The falling light around it confesses everything.",
            "Do not cross the bright ring. Read it from the edge where time still behaves.",
        ],
        "emission_nebula": [
            "New stars were born in the wound, and we mistook their first light for forgiveness.",
            "The cloud glows because something young is demanding to be seen.",
        ],
    }
    return rng.choice(fragments.get(concept_id, fragments["pulsar"]))


@spaces.GPU(duration=20)
def author_sector_payload(sector_key: str, position_key: str) -> dict[str, Any]:
    # The model adapter belongs here. For the slice, physics text is assembled only
    # from curated facts while the fiction layer remains free to vary by seed.
    return generate_sector_cached(sector_key, position_key)


@app.api(name="generate_sector", concurrency_limit=1)
def generate_sector(sector_key: str, position_key: str) -> dict[str, Any]:
    return author_sector_payload(sector_key, position_key)


@app.get("/api/sector/{sector_key}")
async def sector_rest(sector_key: str, position: str = "0,20,-450"):
    return JSONResponse(generate_sector_cached(sector_key, position))


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


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