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

import hashlib
import os
import random
import time
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,
    pick_clip_id,
    scale_at,
)
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"

# Per-process boot token: changes on every server restart so the dev live-reload
# (below) also fires when Python code changes (which needs a restart), not just
# when a static asset's mtime changes.
_BOOT_ID = f"{os.getpid()}-{int(time.time())}"


def _asset_version() -> str:
    """A short digest of the renderer assets' mtimes + this process's boot token.
    Changes whenever app.js/style.css/index.html change on disk OR the server is
    restarted — the signal the browser polls to know it's running stale code."""
    parts = [_BOOT_ID]
    for name in ("app.js", "style.css", "index.html"):
        try:
            parts.append(f"{name}:{(STATIC_DIR / name).stat().st_mtime_ns}")
        except OSError:
            parts.append(f"{name}:?")
    return hashlib.sha1("|".join(parts).encode()).hexdigest()[:16]


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
    dream_gain: float = 1.0


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,
                dream_gain=req.calibration.dream_gain,
            )
            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("/dev/version")
    def dev_version():
        # Dev live-reload signal: the browser polls this and reloads when it
        # changes, so it never silently runs a stale renderer during iteration.
        return {"version": _asset_version()}

    @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,
        )
        # Rotating pool: pick a random member of the LANDED scale (content-pipeline
        # §11.1). A delta=0 advance is the initial / re-roll pick (a no-op move that
        # still yields a fresh random clip). The pure pick takes injected randomness.
        landed = scale_at(app.state.ring, move.to_index)
        chosen = pick_clip_id(landed, random.random())
        return ring_move_to_dict(move, app.state.ring, chosen)

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