human-experience-filter-art/simulator/app.py

"""FastAPI service: controls -> the real alteration engine -> a RenderPlan.

The simulator's alteration surface (reconciled slice). It calls the canonical
player.alteration.plan_alteration; the browser only renders. The selection-era
endpoints (/api/select, /api/catalog/meta) and the X-ray are retired.
"""

from __future__ import annotations

from pathlib import Path
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from hef.selection import Coordinate
from player.alteration import (
    DEFAULT_CALIBRATION,
    Calibration,
    plan_alteration,
    render_plan_to_dict,
)
from player.content import resolve_content
from player.controls import CONTENT_POSITIONS
from player.ring import DEFAULT_FAST_SPIN_THRESHOLD, advance_ring
from simulator.clips import load_manifest, load_ring, ring_move_to_dict, ring_to_dict

STATIC_DIR = Path(__file__).parent / "static"
MEDIA_DIR = Path(__file__).parent / "sample_media"
DEFAULT_MANIFEST = MEDIA_DIR / "manifest.json"


class ControlsModel(BaseModel):
    content: str
    left: int = Field(ge=0, le=4)
    right: int = Field(ge=0, le=4)
    dark: int = Field(ge=0, le=4)
    light: int = Field(ge=0, le=4)
    volume: int = Field(ge=0, le=4)
    brightness: int = Field(ge=0, le=4)


class CalibrationModel(BaseModel):
    mood_gain: float = 1.0
    overlay_gain: float = 1.0
    right_variant_map: list[int] = [0, 1, 2, 3, 4]


class AlterationRequest(BaseModel):
    controls: ControlsModel
    calibration: Optional[CalibrationModel] = None


class RingAdvanceRequest(BaseModel):
    from_index: int = 0
    delta: int


def _load_clips(manifest_path: Optional[Path]):
    path = Path(manifest_path) if manifest_path else DEFAULT_MANIFEST
    if path.exists():
        return load_manifest(path)
    return []


def _load_ring(manifest_path: Optional[Path]):
    path = Path(manifest_path) if manifest_path else DEFAULT_MANIFEST
    if path.exists():
        return load_ring(path)
    return None


def create_app(manifest_path: Optional[Path] = None) -> FastAPI:
    app = FastAPI(title="HEF Alteration Simulator")
    app.state.clips = _load_clips(manifest_path)
    app.state.ring = _load_ring(manifest_path)

    @app.post("/api/alteration")
    def api_alteration(req: AlterationRequest):
        c = req.controls
        if c.content not in CONTENT_POSITIONS:
            raise HTTPException(status_code=422, detail=f"invalid content {c.content!r}")
        coord = Coordinate(c.left, c.right, c.dark, c.light)
        cal = (
            Calibration(
                mood_gain=req.calibration.mood_gain,
                overlay_gain=req.calibration.overlay_gain,
                right_variant_map=tuple(req.calibration.right_variant_map),
            )
            if req.calibration
            else DEFAULT_CALIBRATION
        )
        plan = plan_alteration(coord, cal)
        content = resolve_content(c.content)
        return {
            "plan": render_plan_to_dict(plan),
            "content": {"audio_source": content.audio_source, "video": content.video},
        }

    @app.get("/api/clips")
    def api_clips():
        return {"clips": [c.to_dict() for c in app.state.clips]}

    @app.get("/api/ring")
    def api_ring():
        if app.state.ring is None:
            raise HTTPException(status_code=404, detail="no scale ring in manifest")
        return ring_to_dict(app.state.ring, app.state.clips)

    @app.post("/api/ring/advance")
    def api_ring_advance(req: RingAdvanceRequest):
        if app.state.ring is None:
            raise HTTPException(status_code=404, detail="no scale ring in manifest")
        move = advance_ring(
            app.state.ring,
            req.from_index,
            req.delta,
            fast_spin_threshold=DEFAULT_FAST_SPIN_THRESHOLD,
        )
        return ring_move_to_dict(move, app.state.ring)

    @app.middleware("http")
    async def _no_cache(request, call_next):
        # Dev preview server: never let a browser serve a stale app.js/style.css.
        # The whole point is by-eye iteration, so a plain refresh must always get
        # the latest assets (the heuristic cache otherwise hides JS/CSS edits).
        response = await call_next(request)
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        return response

    if MEDIA_DIR.exists():
        app.mount("/media", StaticFiles(directory=MEDIA_DIR), name="media")
    if STATIC_DIR.exists():
        app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")

    return app


app = create_app()