"""FastAPI application for the Pyre Environment. Uses a factory function so each WebSocket session gets an isolated environment instance. Configuration via environment variables: PYRE_MAX_STEPS max steps before timeout (default 150) PYRE_SEED base random seed (default 42) PORT server port (default 8000) """ import os from pathlib import Path from typing import Any, Dict, List, Optional from pydantic import Field, BaseModel from fastapi import HTTPException from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from openenv.core.env_server.http_server import create_app from starlette.routing import Route try: from ..models import PyreAction, PyreObservation from .pyre_env_environment import PyreEnvironment from .narrative import build_narrative_observation, compute_visible_cells except (ImportError, ModuleNotFoundError): from models import PyreAction, PyreObservation from server.pyre_env_environment import PyreEnvironment from server.narrative import build_narrative_observation, compute_visible_cells MAX_STEPS = int(os.getenv("PYRE_MAX_STEPS", "150")) BASE_SEED = int(os.getenv("PYRE_SEED", "42")) def create_pyre_environment() -> PyreEnvironment: return PyreEnvironment(max_steps=MAX_STEPS, base_seed=BASE_SEED) app = create_app( create_pyre_environment, PyreAction, PyreObservation, env_name="pyre_env", ) _stateful_env: Optional[PyreEnvironment] = None def get_stateful_env() -> PyreEnvironment: """Return singleton env used by HTTP reset/step for browser workflows.""" global _stateful_env if _stateful_env is None: _stateful_env = create_pyre_environment() return _stateful_env # Remove stateless HTTP routes from create_app so these stateful overrides are used. app.routes[:] = [ r for r in app.routes if not ( isinstance(r, Route) and ( (r.path in {"/reset", "/step"} and "POST" in (r.methods or set())) or (r.path == "/state" and "GET" in (r.methods or set())) ) ) ] class ResetRequest(BaseModel): seed: Optional[int] = None difficulty: str = "medium" class StepRequest(BaseModel): action: str = Field(..., description="Action type: move | door | wait | look") direction: Optional[str] = Field(None, description="Cardinal direction for move/look") target_id: Optional[str] = Field(None, description="Door ID for door action") door_state: Optional[str] = Field(None, description="'open' or 'close' for door action") STATIC_DIR = Path(__file__).resolve().parent / "static" @app.get("/") def serve_frontend() -> FileResponse: """Serve the React frontend from server/static/index.html.""" html_path = STATIC_DIR / "index.html" if not html_path.exists(): # Fallback to the RPG viewer if index.html is missing rpg_path = STATIC_DIR / "viewer_rpg.html" if rpg_path.exists(): return FileResponse(str(rpg_path)) raise HTTPException(status_code=404, detail="Frontend file not found.") return FileResponse(str(html_path)) # Mount the static directory for assets (CSS, JS, etc.) if (STATIC_DIR / "assets").exists(): app.mount("/assets", StaticFiles(directory=str(STATIC_DIR / "assets")), name="assets") @app.post("/reset") def reset_episode(body: ResetRequest = ResetRequest()) -> Dict[str, Any]: env = get_stateful_env() obs = env.reset(seed=body.seed, difficulty=body.difficulty) return { "observation": obs.model_dump(), "reward": float(obs.reward or 0.0), "done": bool(obs.done), "metadata": obs.metadata or {}, } @app.post("/step") def step_episode(body: StepRequest) -> Dict[str, Any]: env = get_stateful_env() if getattr(env, "_fire_sim", None) is None: raise HTTPException(status_code=409, detail="No active episode. Call POST /reset first.") obs = env.step(PyreAction( action=body.action, direction=body.direction, target_id=body.target_id, door_state=body.door_state, )) return { "observation": obs.model_dump(), "reward": float(obs.reward or 0.0), "done": bool(obs.done), "metadata": obs.metadata or {}, } @app.get("/state") def get_state() -> Dict[str, Any]: env = get_stateful_env() return env.state.model_dump() @app.get("/scene") def get_scene() -> Dict[str, Any]: """Return a compact scene snapshot for external frontends. Response shape -------------- labels agent — position, health, status flags, perception summary episode — fire parameters, step counters, difficulty map — grid dimensions, exit positions, door registry surroundings — visible objects, blocked exits, audible signals, available action hints graph channels — ordered list of channel names (index guide) channel_info — human-readable description of each channel width / height grid — grid[y][x] = [cell_type, fire, smoke, is_agent, is_visible] cell_type: 0=floor 1=wall 2=door_open 3=door_closed 4=exit 5=obstacle fire / smoke: 0.0 (none) → 1.0 (max) is_agent / is_visible: 0 or 1 """ env = get_stateful_env() st = env.state # --- Build structured observation fields (no narrative) --- obs_data = build_narrative_observation( step_count=st.step_count, agent_x=st.agent_x, agent_y=st.agent_y, agent_alive=st.agent_alive, agent_evacuated=st.agent_evacuated, agent_health=st.agent_health, cell_grid=st.cell_grid, fire_grid=st.fire_grid, smoke_grid=st.smoke_grid, exit_positions=st.exit_positions, door_registry=st.door_registry, zone_map=st.zone_map, last_action_feedback=getattr(env, "_last_feedback", ""), wind_dir=st.wind_dir, w=st.grid_w, h=st.grid_h, ) # --- Visibility set for the graph layer --- if st.agent_alive and not st.agent_evacuated: visible_set = compute_visible_cells( st.agent_x, st.agent_y, st.cell_grid, st.smoke_grid, st.grid_w, st.grid_h, ) else: visible_set = set() # --- Labels --- labels: Dict[str, Any] = { "agent": { "x": st.agent_x, "y": st.agent_y, "health": st.agent_health, "health_status": obs_data.get("health_status", "Good"), "alive": st.agent_alive, "evacuated": st.agent_evacuated, "location": obs_data.get("location_label", ""), "smoke_level": obs_data.get("smoke_level", "none"), "fire_visible": obs_data.get("fire_visible", False), "fire_direction": obs_data.get("fire_direction", None), "last_action_feedback": obs_data.get("last_action_feedback", ""), }, "episode": { "id": st.episode_id, "step": st.step_count, "max_steps": st.max_steps, "template": st.template_name, "difficulty": getattr(env, "_difficulty", "medium"), "wind_dir": st.wind_dir, "fire_spread_rate": st.fire_spread_rate, "humidity": st.humidity, "fire_sources": st.fire_sources_count, }, "map": { "width": st.grid_w, "height": st.grid_h, "exit_positions": st.exit_positions, "door_registry": st.door_registry, }, "surroundings": { "visible_objects": obs_data.get("visible_objects", []), "blocked_exit_ids": obs_data.get("blocked_exit_ids", []), "audible_signals": obs_data.get("audible_signals", []), "available_actions": obs_data.get("available_actions_hint", []), }, } # --- 2-D multi-channel grid --- w, h = st.grid_w, st.grid_h grid: List[List[List[float]]] = [] for y in range(h): row: List[List[float]] = [] for x in range(w): idx = y * w + x cell_type = float(st.cell_grid[idx]) fire = round(st.fire_grid[idx], 4) smoke = round(st.smoke_grid[idx], 4) is_agent = 1.0 if (x == st.agent_x and y == st.agent_y) else 0.0 is_visible = 1.0 if (x, y) in visible_set else 0.0 row.append([cell_type, fire, smoke, is_agent, is_visible]) grid.append(row) graph: Dict[str, Any] = { "channels": ["cell_type", "fire", "smoke", "is_agent", "is_visible"], "channel_info": { "cell_type": "0=floor 1=wall 2=door_open 3=door_closed 4=exit 5=obstacle", "fire": "0.0=none to 1.0=fully burning", "smoke": "0.0=clear to 1.0=dense smoke", "is_agent": "1 if agent occupies this cell, else 0", "is_visible": "1 if within agent line-of-sight, else 0", }, "width": w, "height": h, "grid": grid, } return {"labels": labels, "graph": graph} def main(host: str = "0.0.0.0", port: int = 8000): import uvicorn port = int(os.getenv("PORT", port)) uvicorn.run(app, host=host, port=port) if __name__ == "__main__": main()