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 json
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


class AuthorTrackRequest(BaseModel):
    """Author mode: propagate a hand-seeded box on a clip into a keyframed track
    (content-pipeline §11.5). `seed_box` + `seed_t` are normalized."""

    clip_id: str
    key: str
    seed_box: list[float] = Field(min_length=4, max_length=4)
    seed_t: float = Field(default=0.0, ge=0.0, le=1.0)
    salience: int = Field(default=4, ge=1, le=4)
    n_keyframes: int = Field(default=5, ge=2, le=20)
    max_frames: Optional[int] = None


class AuthorClipRequest(BaseModel):
    """Author mode: persist the authored labels/affect/strings for an existing
    pool clip into the manifest (geometry from the tracker, semantics by hand)."""

    clip_id: str
    annotations: list = []
    affect: list = []
    strings: dict = {}


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.manifest_path = Path(manifest_path) if manifest_path else DEFAULT_MANIFEST
    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.post("/api/author/track")
    def api_author_track(req: AuthorTrackRequest):
        # Author mode (§11.5): run the classical optical-flow tracker on a clip's
        # base footage, returning the keyframed track geometry for the author to
        # accept/correct. Semantics (key/strings/tiers) stay the author's.
        clip = next((c for c in app.state.clips if c.id == req.clip_id), None)
        if clip is None:
            raise HTTPException(status_code=404, detail=f"unknown clip {req.clip_id!r}")
        base = MEDIA_DIR / clip.base_file
        if not base.exists():
            raise HTTPException(status_code=404, detail=f"media absent: {clip.base_file}")
        try:
            import cv2
        except ImportError:
            raise HTTPException(status_code=503, detail="opencv-python not installed")
        from tools.pipeline.track import track_seed

        cap = cv2.VideoCapture(str(base))
        total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
        cap.release()
        seed_frame = max(0, min(total - 1, round(req.seed_t * total))) if total else 0
        return track_seed(
            base, req.key, tuple(req.seed_box), seed_frame=seed_frame,
            salience=req.salience, n_keyframes=req.n_keyframes, max_frames=req.max_frames,
        )

    @app.post("/api/author/clip")
    def api_author_clip(req: AuthorClipRequest):
        # Author mode (§11.5 / stage 6): persist authored labels/affect/strings for
        # an EXISTING pool clip — idempotent upsert that keeps the clip's media +
        # provenance and replaces only the authored content. Reloads in place so
        # the preview reflects the edit without a restart.
        from tools.pipeline.manifest import upsert_clip

        path = app.state.manifest_path
        data = json.loads(path.read_text())
        existing = next((c for c in data.get("clips", []) if c["id"] == req.clip_id), None)
        if existing is None:
            raise HTTPException(status_code=404, detail=f"unknown clip {req.clip_id!r}")
        merged = dict(existing)
        merged["annotations"] = req.annotations
        merged["affect"] = req.affect
        if req.strings:
            merged["strings"] = req.strings
        data = upsert_clip(data, merged)
        path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
        app.state.clips = load_manifest(path)
        app.state.ring = load_ring(path)
        return {"ok": True, "clip": merged}

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