Spaces:
Sleeping
Sleeping
File size: 3,175 Bytes
8dff220 | 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 | """
FastAPI JSON API for the bundled EWS model.
Run locally: uvicorn main:app --host 0.0.0.0 --port 7860
"""
from __future__ import annotations
import os
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import ews_inference_core as ews
BUNDLE_PATH = Path(os.environ.get("EWS_BUNDLE_PATH", "/app/ews_pipeline_bundle.joblib"))
if not BUNDLE_PATH.is_file():
BUNDLE_PATH = Path(__file__).resolve().parent / "ews_pipeline_bundle.joblib"
_bundle: dict | None = None
app = FastAPI(title="MP Dengue EWS Inference", version="1.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.on_event("startup")
def _load() -> None:
global _bundle
p = BUNDLE_PATH
if not p.is_file():
raise RuntimeError(
f"Bundle not found: {p}. Run bundle_ews_pipeline.py and copy the .joblib here, "
"or set EWS_BUNDLE_PATH."
)
_bundle = ews.load_bundle(str(p))
def _bundle_get() -> dict:
if _bundle is None:
raise HTTPException(status_code=503, detail="Model not loaded")
return _bundle
class PredictRequest(BaseModel):
"""Provide either `features` (name → value) or `values` (length = n_features in order)."""
features: dict[str, float] | None = None
values: list[float] | None = None
population: int | None = Field(
default=None,
ge=1,
description="District population for estimated_cases and risk_tier (optional)",
)
@app.get("/")
def root():
from fastapi.responses import RedirectResponse
return RedirectResponse(url="/docs")
@app.get("/health")
def health():
return {"status": "ok", "bundle": str(BUNDLE_PATH.resolve())}
@app.get("/schema")
def schema():
b = _bundle_get()
cfg = b["config"]
return {
"feature_columns": cfg["feature_columns"],
"n_features": cfg["n_features"],
"prediction_horizon_weeks": cfg.get("prediction_horizon"),
"cases_lags": cfg.get("cases_lags"),
"probability_threshold": cfg.get("decision_threshold"),
}
@app.post("/predict")
def predict(body: PredictRequest):
b = _bundle_get()
if body.features is not None and body.values is not None:
raise HTTPException(
status_code=400,
detail="Send only one of `features` or `values`, not both.",
)
if body.features is None and body.values is None:
raise HTTPException(
status_code=400,
detail="Provide `features` (object) or `values` (array).",
)
try:
if body.features is not None:
X = ews.vector_from_feature_dict(b, body.features)
else:
X = ews.vector_from_values(b, body.values or [])
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
try:
out = ews.predict_from_bundle(b, X, body.population)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Prediction failed: {e!s}") from e
return {"ok": True, "result": out}
|