benstull/human-experience-filter-art
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(simulator): /api/alteration + /api/clips; retire selection endpoints

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-07 22:55:10 -07:00
parent c316309fc6
commit eb3aa0949d
2 changed files with 119 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
simulator/app.py
+65 -62
View File
@@ -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")
tests/test_simulator_api.py
+54 -66
View File
@@ -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