"""
SepsisPilot — FastAPI HTTP Server
Exposes the OpenEnv API over HTTP so inference.py (and any agent) can interact
with the environment via standard REST calls.
Endpoints:
POST /reset → start new episode
POST /step → take one action
GET /state → current state
GET /grade → grade completed episode
GET /tasks → list available tasks
GET /health → liveness check
GET / → visual dashboard (HTML)
"""
from __future__ import annotations
import os
from contextlib import asynccontextmanager
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
from typing import Optional as Opt
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from environment import SepsisPilotEnv, AVAILABLE_TASKS
from environment.models import (
ActionRequest, GraderResult, PatientState, ResetRequest,
StepResult, TaskInfo,
)
# ── Session state (single-session server; suitable for hackathon eval) ──
_env: Optional[SepsisPilotEnv] = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global _env
_env = SepsisPilotEnv()
yield
app = FastAPI(
title="SepsisPilot OpenEnv",
description=(
"Reinforcement learning environment for optimal sepsis treatment sequencing. "
"Trains AI agents to learn antibiotic + vasopressor policies in simulated ICU patients."
),
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
# ──────────────────────────────────────────────
# OpenEnv API
# ──────────────────────────────────────────────
@app.post("/reset", response_model=PatientState, tags=["OpenEnv"])
async def reset(body: Opt[ResetRequest] = None):
"""Reset the environment and begin a new episode."""
if body is None:
body = ResetRequest()
if body.task not in AVAILABLE_TASKS:
raise HTTPException(
status_code=400,
detail=f"Unknown task '{body.task}'. Available: {AVAILABLE_TASKS}",
)
state = _env.reset(task=body.task, seed=body.seed)
return state
@app.post("/step", response_model=StepResult, tags=["OpenEnv"])
async def step(body: ActionRequest):
"""Apply an action. Returns next state, reward, done flag, and info."""
try:
result = _env.step(body.action)
except RuntimeError as e:
raise HTTPException(status_code=400, detail=str(e))
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
return result
@app.get("/state", response_model=PatientState, tags=["OpenEnv"])
async def state():
"""Return the current environment state without advancing the simulation."""
try:
return _env.state()
except RuntimeError as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/grade", response_model=GraderResult, tags=["OpenEnv"])
async def grade():
"""Grade a completed episode. Returns a score in [0.0, 1.0]."""
try:
return _env.grade()
except RuntimeError as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/tasks", response_model=list[TaskInfo], tags=["OpenEnv"])
async def tasks():
"""List all available tasks with descriptions."""
return SepsisPilotEnv.task_list()
@app.get("/health", tags=["System"])
async def health():
return {"status": "ok", "env": "SepsisPilot", "version": "1.0.0"}
# ──────────────────────────────────────────────
# Visual Dashboard
# ──────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def dashboard():
return HTMLResponse(content=_DASHBOARD_HTML)
_DASHBOARD_HTML = """
SepsisPilot — ICU AI Agent