Spaces:
Paused
Paused
File size: 4,699 Bytes
9762da0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """FastAPI router for the sequential regression-detection engine.
Endpoints (prefix /api/seqeval):
POST /runs start a sequential regression evaluation
GET /runs list runs
GET /runs/{id} state: stage, samples vs budget, per-node
confidence sequences, decisions, and the
sequential-evidence trajectories (points +
stopping boundaries) for the dashboard chart
GET /runs/{id}/evidence machine-readable evidence ledger + lineage
GET /runs/{id}/attribution the regression map
GET /policy default statistical policy (documented)
"""
from __future__ import annotations
from typing import Dict, Optional
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from sensorflow.seqeval import hierarchy as hierarchy_mod
from sensorflow.seqeval import ledger as ledger_mod
from sensorflow.seqeval.controller import (DEFAULT_POLICY, get_seqeval_store,
merged_policy)
router = APIRouter(prefix="/api/seqeval")
class ModelSpec(BaseModel):
model_version: str
effects: Dict[str, float] = Field(
default_factory=dict,
description="Per-stratum detection deltas, e.g. {'pedestrian|night': -0.02};"
" '__global__' applies everywhere (synthetic model harness).")
class StartRunRequest(BaseModel):
population_id: str
baseline: ModelSpec
candidate: ModelSpec
policy: Dict = Field(default_factory=dict)
seed: Optional[int] = None
sync: bool = Field(True, description="Run to completion before returning "
"(seconds); false = background thread.")
@router.post("/runs")
def start_run(req: StartRunRequest):
store = get_seqeval_store()
try:
run = store.create_run(req.population_id, req.baseline.model_dump(),
req.candidate.model_dump(),
policy=req.policy or None, seed=req.seed)
except KeyError as e:
raise HTTPException(404, str(e))
if req.sync:
store.execute_sync(run)
else:
store.start_async(run)
return run.to_dict(include_trajectories=False)
@router.get("/runs")
def list_runs():
return {"runs": get_seqeval_store().list_states()}
def _state_or_404(run_id: str, include_trajectories: bool = True) -> Dict:
state = get_seqeval_store().get_state(run_id, include_trajectories=include_trajectories)
if state is None:
raise HTTPException(404, f"Unknown seqeval run {run_id}")
return state
@router.get("/runs/{run_id}")
def get_run(run_id: str):
return _state_or_404(run_id)
@router.get("/runs/{run_id}/evidence")
def get_evidence(run_id: str):
_state_or_404(run_id, include_trajectories=False)
ledger = ledger_mod.EvidenceLedger(run_id)
return {"run_id": run_id, "records": ledger.records(),
"lineage": ledger.lineage(),
"required_fields": ledger_mod.REQUIRED_FIELDS}
@router.get("/runs/{run_id}/attribution")
def get_attribution(run_id: str):
state = _state_or_404(run_id, include_trajectories=False)
store = get_seqeval_store()
run = store.runs.get(run_id)
if run is not None and run.attribution is not None:
att = run.attribution
else:
att = state.get("attribution")
if att is None:
raise HTTPException(409, f"Run {run_id} has no attribution yet "
f"(status={state.get('status')})")
return {"run_id": run_id, "decision": state.get("decision"),
"gate": state.get("gate"), **att}
@router.get("/policy")
def get_policy():
return {
"default_policy": merged_policy(),
"decision_semantics": {
"REGRESSION": "block: anytime-valid evidence of a drop beyond the "
"practical margin in at least one tested node",
"PASS": "allow: overall + all pre-registered safety primaries proven "
"within the margin (equivalence-style claim)",
"INSUFFICIENT_EVIDENCE": "expand budget or report: NOT proven "
"equivalent; never treated as a pass",
},
"test_method": ledger_mod.TEST_METHOD,
"multiple_testing_method": hierarchy_mod.MULTIPLE_TESTING_METHOD,
"notes": [
"sampling plans are frozen (hashed) before candidate outcomes are seen",
"inference units are container clusters, not frames/objects",
"baseline predictions are cached by dataset+model fingerprints",
],
}
|