Ben Stull
cdf36c9b57
Per docs/superpowers/plans/2026-06-04-experience-simulator.md Task 4. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
"""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()
|