""" PharmaAgent — Clinical Decision RL Environment app.py: FastAPI server exposing OpenMV-compatible endpoints """ import json import uuid import time from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse import os from .models import Action, Observation, State from .environment import PharmaAgentEnvironment app = FastAPI( title="PharmaAgent RL Environment", description="Clinical Decision Agent — Drug Intelligence RL Environment powered by DrugBank", version="1.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) env = PharmaAgentEnvironment() # Fix #10: session store now tracks last-access timestamps so stale sessions # can be evicted, preventing unbounded memory growth under sustained load. SESSION_TTL_SECONDS = 3600 # evict sessions idle for more than 1 hour sessions: dict = {} # session_id -> state dict session_touched: dict = {} # session_id -> last access timestamp def _evict_stale_sessions(): """Remove sessions that have been idle longer than SESSION_TTL_SECONDS.""" now = time.time() stale = [sid for sid, t in session_touched.items() if now - t > SESSION_TTL_SECONDS] for sid in stale: sessions.pop(sid, None) session_touched.pop(sid, None) def _touch(session_id: str): session_touched[session_id] = time.time() # ── OpenMV-compatible endpoints ──────────────────────────────────────── @app.post("/reset") def reset(session_id: str = None): """Reset the environment and return the initial observation.""" _evict_stale_sessions() if session_id is None: session_id = str(uuid.uuid4()) state, obs = env.reset() sessions[session_id] = state.model_dump() _touch(session_id) return { "session_id": session_id, "observation": obs.model_dump(), "state": state.model_dump(), } @app.post("/step") def step(action: Action, session_id: str = None): """Take one step in the environment.""" if session_id is None or session_id not in sessions: raise HTTPException(status_code=400, detail="Invalid or missing session_id. Call /reset first.") state = State(**sessions[session_id]) if state.done: raise HTTPException(status_code=400, detail="Episode is done. Call /reset to start a new episode.") new_state, obs, reward, done = env.step(state, action) sessions[session_id] = new_state.model_dump() if done: # Fix #10: eagerly evict completed sessions — no need to keep them around sessions.pop(session_id, None) session_touched.pop(session_id, None) else: _touch(session_id) return { "session_id": session_id, "observation": obs.model_dump(), "reward": reward, "done": done, "state": new_state.model_dump(), } @app.get("/state") def get_state(session_id: str): """Get the current state of a session.""" if session_id not in sessions: raise HTTPException(status_code=404, detail="Session not found.") _touch(session_id) return {"session_id": session_id, "state": sessions[session_id]} @app.get("/health") def health(): """Health check endpoint.""" return { "status": "ok", "environment": "PharmaAgent", "version": "1.0.0", "active_sessions": len(sessions), } @app.get("/") def root(): return { "name": "PharmaAgent RL Environment", "description": "Clinical Decision Agent powered by DrugBank", "endpoints": ["/reset", "/step", "/state", "/health", "/web"], } # ── Optional Web UI ──────────────────────────────────────────── ENABLE_WEB = os.environ.get("ENABLE_WEB_INTERFACE", "true").lower() == "true" if ENABLE_WEB: @app.get("/web", response_class=HTMLResponse) def web_ui(): """Simple browser UI for manually testing the environment.""" return """ PharmaAgent — RL Environment Tester

🏥 PharmaAgent

Clinical Decision RL Environment — Manual Tester
Session Control
"""