""" FastAPI application for the OASIS Environment. Uses OpenEnv's create_app factory to generate standard endpoints: - POST /reset : Reset the environment (accepts task_id in body) - POST /step : Execute an insulin dosing action - GET /state : Get current environment state - GET /health : Health check - GET /schema : Action/observation JSON schemas - WS /ws : WebSocket for persistent sessions Additional custom endpoints: - GET /tasks : List all 3 tasks with descriptions - GET /healthz : Alias health check Usage: uvicorn server.app:app --host 0.0.0.0 --port 8000 """ import logging import sys import os from fastapi.responses import HTMLResponse # Ensure project root is on path for imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) # Disable OpenEnv's broken web interface — we serve our own at / os.environ["ENABLE_WEB_INTERFACE"] = "false" from openenv.core.env_server.http_server import create_app from models import GlucoAction, GlucoObservation from server.glucorl_environment import GlucoRLEnvironment from server.graders import grade, grade_detailed logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Create the OpenEnv-compliant FastAPI app # --------------------------------------------------------------------------- app = create_app( GlucoRLEnvironment, GlucoAction, GlucoObservation, env_name="oasis", max_concurrent_envs=1, ) # --------------------------------------------------------------------------- # Custom endpoints # --------------------------------------------------------------------------- @app.get("/tasks", tags=["Environment Info"]) async def list_tasks(): """Return descriptions of all 4 OASIS tasks.""" return [ { "id": 1, "name": "Basal Rate Control", "difficulty": "easy", "description": ( "Single stable adult patient, no meals. " "Optimise basal insulin rate to keep glucose in " "70-180 mg/dL for a full 24-hour simulated day." ), }, { "id": 2, "name": "Meal Bolus Timing", "difficulty": "medium", "description": ( "Same adult patient with 3 announced daily meals " "(breakfast 50g, lunch 70g, dinner 80g). " "Deliver correct bolus doses at the right time to " "prevent post-meal spikes while avoiding hypoglycemia." ), }, { "id": 3, "name": "Cross-Patient Generalisation", "difficulty": "hard", "description": ( "Random patient sampled from 30 profiles " "(adolescent, adult, child). Meals are NOT announced. " "Develop a policy that generalises across varied " "patient physiology without knowing which patient " "is being treated." ), }, { "id": 4, "name": "Sick Day Management", "difficulty": "expert", "description": ( "Random patient with simulated illness causing 1.5-2.5x " "insulin resistance starting at an unknown time. " "Meals and exercise are unannounced. The agent must " "detect rising glucose from resistance and adapt its " "dosing strategy without being told illness is occurring." ), }, ] @app.get("/healthz", tags=["Health"]) async def healthz(): """Quick health check — verifies the environment can be instantiated.""" try: env = GlucoRLEnvironment() return {"status": "ok"} except Exception as e: logger.error("Health check failed: %s", e, exc_info=True) return {"status": "error", "error": str(e)} @app.post("/grade", tags=["Evaluation"]) async def grade_episode(task_id: int = 1): """ Grade the current completed episode and return detailed score breakdown. Instantiates an environment to access state. Note: for stateful grading, use the WebSocket client to run an episode, then call state() and grade client-side. This endpoint is provided for convenience and testing. Returns 400 if no episode data is available. """ try: env = GlucoRLEnvironment() state = env.state if not state.glucose_history: return {"error": "No episode data available. Run an episode first via WebSocket."} result = grade_detailed(task_id, state) return result except ValueError as e: return {"error": str(e)} except Exception as e: logger.error("Grade endpoint failed: %s", e, exc_info=True) return {"error": str(e)} # --------------------------------------------------------------------------- # OASIS Web Interface — served at / for HuggingFace Spaces # --------------------------------------------------------------------------- APP_HTML = """ OASIS — Insulin Dosing RL Environment

đŸĨ OASIS Optimized Adaptive System for Insulin Scheduling

OPENENV 1.0 LIVE
Glucose (mg/dL)
—
True Glucose
—
Time in Range
—
Current Step
—
Hypo Events
—
Total Reward
—
Episode: Not started
📈 Glucose Trace
Welcome to OASIS. Click Reset to start an episode.
⚡ Manual Control
Basal Rate (U/hr)
1.0
Bolus Dose (units)
0.0
📋 Current Observation
Click Reset to begin
đŸŽ¯ Task Scores

TASK 1

Basal Rate Control
—

TASK 2

Meal Bolus Timing
—

TASK 3

Cross-Patient
—

TASK 4

Sick Day
—
""" @app.get("/", response_class=HTMLResponse, tags=["UI"], include_in_schema=False) async def root_ui(): """OASIS web interface — served at root for HuggingFace Spaces.""" return HTMLResponse(content=APP_HTML) @app.get("/dashboard", response_class=HTMLResponse, tags=["UI"]) async def dashboard(): """Alias for the main web interface.""" return HTMLResponse(content=APP_HTML) # --------------------------------------------------------------------------- # Entry point for direct execution # --------------------------------------------------------------------------- def main(host: str = "0.0.0.0", port: int = 8000): """Run the OASIS server.""" import uvicorn logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s", ) # Suppress noisy loggers logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) logging.getLogger("websockets").setLevel(logging.WARNING) logging.getLogger("uvicorn.access").setLevel(logging.WARNING) uvicorn.run(app, host=host, port=port) if __name__ == "__main__": main()