import json from pathlib import Path from typing import Any import uvicorn from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from environment.avigilance_env import AvigilanceEnv from environment.models import AvigilanceAction FRONTEND_HTML = """ AvigilanceEnv Space Console

DGCA Monitoring Space

Avigilance Mission Console

20k+ Hybrid Corpus OpenEnv Ready Real Ingestion Plan Ready

Live Operations Deck

Run task resets, inspect payloads, and submit exact environment actions from one screen.

This Space root is entirely self-contained in app.py and now fronts a 20k-plus hybrid corpus with broader airline, airport, and incident coverage, plus a documented path for DGCA, AAIB, and AAI ingestion.

Open Control Deck Verification Flow
API Health Checking...

Polling live service metadata.

Runtime Loading...

Loading corpus coverage and manifest details.

Session No episode yet

Reset any task to initialize an active episode.

Reward Guardrail (0, 1)

Scores remain strictly inside the open interval.

Control Deck

Interactive Task Runner

Awaiting reset

The default seed is 42, which recreates the same scenario every time. If you want different observations and different rewards, change the seed or use Randomize Seed before resetting.

Observation Snapshot No observation loaded
Last Reward None yet
Episode State Idle
Accuracy Not available
Consistency Not available
Safety Alignment Not available
Why This Changes Depends on seed and action

Observation Payload

Backend response after reset or step.

Press Reset Episode to start.

Action Editor

Use typed JSON matching the task schema.

The helper can prefill a valid payload, but you can edit every field before submit.

Runtime Result

State, reward, or error output from the active request.

Waiting for actions.

Observation Interpretation

Plain-English explanation of what the latest observation is telling you.

Reset a task to see the live scenario translated into plain language.

Action Interpretation

Readable summary of the action payload currently in the editor.

Pick a task or load an example action to see what your JSON is doing.

Result Interpretation

Short explanation of reward, completion state, and backend response behavior.

Submit a step or fetch state to decode the response here.

Current Task Field Guide

Human-readable meaning of the most important fields for the selected task.

Verification Path

How This Space Should Be Reviewed

1. Reset a live task

Choose one of the three environments, seed it deterministically, and inspect the exact observation payload emitted by the backend.

2. Submit a typed action

Load the generated example or replace it with your own JSON, then submit the request against the same /step endpoint validators hit.

3. Verify contract behavior

Use the result panel and activity feed to confirm reward bounds, episode status, and endpoint availability before running repository validation scripts.

""" api_app = FastAPI( title="AvigilanceEnv", description="India Aviation Safety Monitoring OpenEnv — DGCA Early Warning System", version="1.1.0", ) _envs: dict[str, AvigilanceEnv] = {} _data_dir = Path(__file__).resolve().parent / "data" def _to_jsonable(value: Any) -> Any: if hasattr(value, "model_dump"): return value.model_dump(mode="json") return value def _dump_json(value: Any) -> str: return json.dumps(_to_jsonable(value), indent=2, ensure_ascii=True) def _load_corpus_manifest() -> dict[str, Any]: manifest_path = _data_dir / "corpus_manifest.json" if manifest_path.exists(): with manifest_path.open("r", encoding="utf-8") as handle: return json.load(handle) return { "version": "unknown", "summary": { "total_records": 0, "fto_profiles": 0, "incident_reports": 0, "resource_scenarios": 0, "unique_airports": 0, "unique_airlines": 0, }, "source_catalog": [], "space_ready": False, } def _reset_session(task_id: str = "task1", seed: int = 42) -> dict[str, Any]: env = AvigilanceEnv(task_id=task_id, seed=seed) _envs[task_id] = env obs = env.reset() return obs.model_dump(mode="json") def _step_session(action: AvigilanceAction) -> dict[str, Any]: env = _envs.get(action.task_id) if env is None: raise HTTPException(status_code=400, detail=f"No active episode for {action.task_id}. Call /reset first.") obs, reward, done, info = env.step(action) return { "observation": obs.model_dump(mode="json"), "reward": reward.model_dump(mode="json"), "done": done, "info": info, } def _get_state(task_id: str = "task1") -> dict[str, Any]: env = _envs.get(task_id) if env is None: return {"status": "no_active_episode"} return env.state() @api_app.get("/api/info") def api_info() -> dict[str, Any]: return { "name": "AvigilanceEnv", "description": "India Aviation Safety Monitoring — OpenEnv Early Warning System", "tasks": ["task1", "task2", "task3"], "status": "ready", } @api_app.post("/reset") def reset(task_id: str = "task1", seed: int = 42) -> JSONResponse: try: return JSONResponse(content=_reset_session(task_id=task_id, seed=seed)) except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc @api_app.post("/step") def step(action: AvigilanceAction) -> JSONResponse: try: return JSONResponse(content=_step_session(action)) except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc @api_app.get("/state") def state(task_id: str = "task1") -> JSONResponse: return JSONResponse(content=_get_state(task_id=task_id)) @api_app.get("/health") def health() -> dict[str, Any]: return {"status": "healthy", "env": "AvigilanceEnv"} @api_app.get("/metadata") def metadata() -> dict[str, Any]: manifest = _load_corpus_manifest() return { "name": "AvigilanceEnv", "description": "India Aviation Safety Monitoring OpenEnv — DGCA Early Warning System", "version": "1.2.0", "tasks": ["task1", "task2", "task3"], "walkthrough": "/#walkthrough", "corpus_summary": manifest.get("summary", {}), "corpus_version": manifest.get("version", "unknown"), "space_ready": manifest.get("space_ready", False), "ingestion_plan_path": "walkthrough/real-data-ingestion-plan.md", } @api_app.get("/walkthrough") def walkthrough() -> RedirectResponse: return RedirectResponse(url="/#walkthrough", status_code=307) @api_app.get("/", response_class=HTMLResponse) def frontend() -> HTMLResponse: return HTMLResponse(FRONTEND_HTML) @api_app.get("/schema") def schema() -> dict[str, Any]: from environment.models import AvigilanceAction as ActionModel from environment.models import AvigilanceObservation, AvigilanceReward return { "observation": AvigilanceObservation.model_json_schema(), "action": ActionModel.model_json_schema(), "state": AvigilanceReward.model_json_schema(), } @api_app.post("/mcp") def mcp(payload: dict | None = None) -> dict[str, Any]: body = payload or {} return { "jsonrpc": "2.0", "id": body.get("id"), "result": { "name": "AvigilanceEnv", "tools": ["reset", "step", "state"], }, } app = api_app def main() -> None: uvicorn.run(app, host="0.0.0.0", port=7860) if __name__ == "__main__": main()