feat(simulator): FastAPI app with /api/select and /api/catalog/meta

Per docs/superpowers/plans/2026-06-04-experience-simulator.md Task 4.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-05 03:42:13 -07:00
parent aadfccf32d
commit cdf36c9b57
2 changed files with 186 additions and 0 deletions
+96
View File
@@ -0,0 +1,96 @@
"""FastAPI service: dials -> real hef.selection -> X-ray (pick + ranked pool)."""
from __future__ import annotations
import os
from collections import Counter
from pathlib import Path
from typing import Literal, Optional
from fastapi import FastAPI
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 simulator.fixtures import generate_fixture_catalog
STATIC_DIR = Path(__file__).parent / "static"
class SelectRequest(BaseModel):
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
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()
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()
@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,
)
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)},
}
@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)),
}
if STATIC_DIR.exists():
app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")
return app
app = create_app()