From eb3aa0949db790266c022f4c980a04d781e35ecb Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Sun, 7 Jun 2026 22:55:10 -0700 Subject: [PATCH] feat(simulator): /api/alteration + /api/clips; retire selection endpoints Co-Authored-By: Claude Opus 4.8 (1M context) --- simulator/app.py | 127 ++++++++++++++++++------------------ tests/test_simulator_api.py | 120 +++++++++++++++------------------- 2 files changed, 119 insertions(+), 128 deletions(-) diff --git a/simulator/app.py b/simulator/app.py index 10b6593..b7e9b67 100644 --- a/simulator/app.py +++ b/simulator/app.py @@ -1,92 +1,95 @@ -"""FastAPI service: dials -> real hef.selection -> X-ray (pick + ranked pool).""" +"""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 os -from collections import Counter from pathlib import Path -from typing import Literal, Optional +from typing import Optional -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field -from hef.catalog import load_catalog, record_to_dict -from hef.selection import ( - Coordinate, - Weights, - candidates_for_mode, - ranked_candidates, - select, +from hef.selection import Coordinate +from player.alteration import ( + DEFAULT_CALIBRATION, + Calibration, + plan_alteration, + render_plan_to_dict, ) -from simulator.fixtures import generate_fixture_catalog +from player.content import resolve_content +from player.controls import CONTENT_POSITIONS +from simulator.clips import load_manifest STATIC_DIR = Path(__file__).parent / "static" +MEDIA_DIR = Path(__file__).parent / "sample_media" +DEFAULT_MANIFEST = MEDIA_DIR / "manifest.json" -class SelectRequest(BaseModel): +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) - mode: Literal["none", "audio", "video", "av"] - pool_size: int = Field(default=4, ge=1, le=25) - brain_weight: float = Field(default=1.0, ge=0.0) - mood_weight: float = Field(default=1.0, ge=0.0) - approved_only: bool = False + volume: int = Field(ge=0, le=4) + brightness: int = Field(ge=0, le=4) -def load_catalog_or_fixtures() -> list: - """Use the real catalog if a non-empty one is configured/exists, else fixtures.""" - configured = os.environ.get("HEF_SIM_CATALOG") - path = Path(configured) if configured else Path("catalog/library.jsonl") - if path.exists() and path.stat().st_size > 0: - return load_catalog(path) - return generate_fixture_catalog() +class CalibrationModel(BaseModel): + mood_gain: float = 1.0 + overlay_gain: float = 1.0 + right_variant_map: list[int] = [0, 1, 2, 3, 4] -def create_app(records: Optional[list] = None) -> FastAPI: - app = FastAPI(title="HEF Experience Simulator") - app.state.catalog = records if records is not None else load_catalog_or_fixtures() +class AlterationRequest(BaseModel): + controls: ControlsModel + calibration: Optional[CalibrationModel] = None - @app.post("/api/select") - def api_select(req: SelectRequest): - catalog = app.state.catalog - coord = Coordinate(req.left, req.right, req.dark, req.light) - weights = Weights(brain=req.brain_weight, mood=req.mood_weight) - if req.mode == "none": - return {"pick": None, "pool": [], "coverage": {"candidates_in_mode": 0}} - pool = catalog - if req.approved_only: - pool = [r for r in pool if r.review_status == "approved"] - eligible = candidates_for_mode(pool, req.mode, req.pool_size) - ranked = ranked_candidates( - catalog, coord, req.mode, - pool_size=req.pool_size, weights=weights, approved_only=req.approved_only, - ) - pick = select( - catalog, coord, req.mode, - pool_size=req.pool_size, weights=weights, approved_only=req.approved_only, - rng=None, + +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 create_app(manifest_path: Optional[Path] = None) -> FastAPI: + app = FastAPI(title="HEF Alteration Simulator") + app.state.clips = _load_clips(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 { - "pick": record_to_dict(pick) if pick else None, - "pool": [ - {"record": record_to_dict(r), "distance": d, "rank": i + 1} - for i, (r, d) in enumerate(ranked) - ], - "coverage": {"candidates_in_mode": len(eligible)}, + "plan": render_plan_to_dict(plan), + "content": {"audio_source": content.audio_source, "video": content.video}, } - @app.get("/api/catalog/meta") - def api_meta(): - catalog = app.state.catalog - return { - "total": len(catalog), - "by_mode": dict(Counter(r.mode for r in catalog)), - "by_status": dict(Counter(r.review_status for r in catalog)), - } + @app.get("/api/clips") + def api_clips(): + return {"clips": [c.to_dict() for c in app.state.clips]} + 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") diff --git a/tests/test_simulator_api.py b/tests/test_simulator_api.py index 57e6b24..d5e32c2 100644 --- a/tests/test_simulator_api.py +++ b/tests/test_simulator_api.py @@ -1,102 +1,90 @@ +import json + import pytest from fastapi.testclient import TestClient -from hef.catalog import Record from simulator.app import create_app -def make_record(**overrides): - base = dict( - id="r", - title="t", - source_url="u", - source_archive="internet_archive", - license="public_domain", - mode="video", - left=0, - right=0, - dark=0, - light=0, - duration_s=600, - file_path="", - ) - base.update(overrides) - return Record(**base) +@pytest.fixture +def manifest_path(tmp_path): + p = tmp_path / "manifest.json" + p.write_text(json.dumps({ + "clips": [{ + "id": "forest", + "title": "neutral forest", + "base_file": "forest/base.mp4", + "license": "poc", "source": "hef-poc", + "right_variants": {"4": {"file": "forest/right4.mp4"}}, + "annotations": [{"key": "detected.water", "box": [0.1, 0.2, 0.3, 0.4], "min_level": 1}], + "strings": {"en": {"detected.water": "flowing water"}}, + }] + })) + return p @pytest.fixture -def client(): - records = [ - make_record(id="v-near", mode="video", left=0, right=0, dark=0, light=0), - make_record(id="v-far", mode="video", left=4, right=4, dark=4, light=4), - make_record(id="a-one", mode="audio", left=1, right=1, dark=1, light=1), - make_record(id="prop", mode="video", left=0, right=0, dark=0, light=1, - review_status="proposed"), - make_record(id="appr", mode="video", left=0, right=0, dark=0, light=1, - review_status="approved"), - ] - return TestClient(create_app(records=records)) +def client(manifest_path): + return TestClient(create_app(manifest_path=manifest_path)) -def _body(**overrides): - base = dict(left=0, right=0, dark=0, light=0, mode="video") - base.update(overrides) - return base +def _controls(content="video", left=0, right=0, dark=0, light=0, volume=2, brightness=2): + return dict(content=content, left=left, right=right, dark=dark, + light=light, volume=volume, brightness=brightness) -def test_select_returns_pick_and_ranked_pool(client): - resp = client.post("/api/select", json=_body(mode="video", pool_size=4)) +def test_alteration_returns_the_engine_plan(client): + resp = client.post("/api/alteration", json={"controls": _controls(left=4, right=2, dark=4)}) assert resp.status_code == 200 data = resp.json() - assert data["pick"]["id"] == "v-near" - ids = [c["record"]["id"] for c in data["pool"]] - assert ids[0] == "v-near" - assert all("distance" in c and "rank" in c for c in data["pool"]) - assert [c["rank"] for c in data["pool"]] == list(range(1, len(data["pool"]) + 1)) + assert data["plan"]["overlay"]["level"] == 4 + assert data["plan"]["restyle"]["variant"] == 2 + assert data["plan"]["grade"]["tone"] == -1.0 + assert data["content"]["video"] is True -def test_none_mode_is_the_void(client): - resp = client.post("/api/select", json=_body(mode="none")) - assert resp.status_code == 200 +def test_alteration_honors_off_as_black(client): + resp = client.post("/api/alteration", json={"controls": _controls(content="off")}) data = resp.json() - assert data["pick"] is None - assert data["pool"] == [] + assert data["content"]["video"] is False -def test_dial_out_of_range_is_rejected(client): - resp = client.post("/api/select", json=_body(left=7)) +def test_alteration_accepts_calibration(client): + body = {"controls": _controls(light=4), + "calibration": {"mood_gain": 0.5, "overlay_gain": 1.0, "right_variant_map": [0, 1, 2, 3, 4]}} + resp = client.post("/api/alteration", json=body) + assert resp.json()["plan"]["grade"]["tone"] == 0.5 + + +def test_alteration_rejects_out_of_range_knob(client): + resp = client.post("/api/alteration", json={"controls": _controls(left=7)}) assert resp.status_code == 422 -def test_bad_mode_is_rejected(client): - resp = client.post("/api/select", json=_body(mode="banana")) +def test_alteration_rejects_bad_content(client): + resp = client.post("/api/alteration", json={"controls": _controls(content="banana")}) assert resp.status_code == 422 -def test_approved_only_narrows_pool(client): - resp = client.post("/api/select", json=_body(left=0, right=0, dark=0, light=1, - mode="video", approved_only=True)) - data = resp.json() - assert all(c["record"]["review_status"] == "approved" for c in data["pool"]) - - -def test_catalog_meta_reports_counts(client): - resp = client.get("/api/catalog/meta") +def test_clips_returns_the_manifest(client): + resp = client.get("/api/clips") assert resp.status_code == 200 data = resp.json() - assert data["total"] == 5 - assert data["by_mode"]["video"] == 4 - assert data["by_mode"]["audio"] == 1 - assert set(data["by_status"]) == {"proposed", "approved"} + assert data["clips"][0]["id"] == "forest" + assert data["clips"][0]["right_variants"]["0"]["file"] == "forest/base.mp4" + assert data["clips"][0]["annotations"][0]["key"] == "detected.water" -from simulator.app import create_app as _create_app_for_static +def test_retired_selection_endpoints_are_gone(client): + # The route no longer exists; the static catch-all yields 404 on GET and + # 405 on the (now-unrouted) POST. Either proves the endpoint is gone. + assert client.post("/api/select", json={}).status_code in (404, 405) + assert client.get("/api/catalog/meta").status_code == 404 def test_index_is_served(): - # The default app mounts the real static dir. - client = TestClient(_create_app_for_static()) + client = TestClient(create_app()) resp = client.get("/") assert resp.status_code == 200 assert "text/html" in resp.headers["content-type"] - assert "X-ray" in resp.text + assert "Alteration" in resp.text