import asyncio
import json
import time
import uuid
from collections import OrderedDict
from collections.abc import AsyncIterator
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from pathlib import Path
import httpx
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import FileResponse, Response
from fastapi.staticfiles import StaticFiles
from openworld_radio_twin import __version__
from openworld_radio_twin.artifacts import build_artifact_bundle, render_radio_map_png
from openworld_radio_twin.batch import (
BatchDatasetRequest,
BatchPlan,
build_batch_plan,
generate_batch_dataset,
)
from openworld_radio_twin.config import get_settings
from openworld_radio_twin.engines import EngineContext, EngineExecutor, EngineRegistry
from openworld_radio_twin.models import (
BuildingContextResponse,
EngineName,
SceneRequest,
SceneResponse,
SearchResult,
SimulationRequest,
SimulationResponse,
)
from openworld_radio_twin.providers.base import SCENE_BOUNDARY_POLICY, BuildingProviderError
from openworld_radio_twin.providers.map_data import MapDataProvider
from openworld_radio_twin.providers.surfaces import SurfaceProviderError
from openworld_radio_twin.providers.terrain import TerrainProviderError
from openworld_radio_twin.scenes import (
CompiledScene,
cached_scene,
compile_web_scene,
environment_context,
scene_provenance,
scene_response,
)
from openworld_radio_twin.simulation.building_context import (
building_geometry_context,
placed_building_geometry_context,
)
from openworld_radio_twin.simulation.radio_map import build_coverage_products
[docs]
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
settings = get_settings()
client = httpx.AsyncClient(
follow_redirects=True,
timeout=settings.http_request_timeout_seconds,
)
app.state.http_client = client
app.state.provider = MapDataProvider(settings, client)
app.state.sionna_lock = asyncio.Lock()
app.state.batch_lock = asyncio.Lock()
app.state.batch_worker = ThreadPoolExecutor(max_workers=1, thread_name_prefix="owrt-batch")
app.state.batch_jobs = OrderedDict()
app.state.batch_tasks = set()
app.state.simulations = OrderedDict()
app.state.environments = OrderedDict()
yield
for task in list(app.state.batch_tasks):
task.cancel()
await asyncio.gather(*app.state.batch_tasks, return_exceptions=True)
app.state.batch_worker.shutdown(wait=True, cancel_futures=True)
await client.aclose()
app = FastAPI(
title="OpenWorld Radio Twin API",
version=__version__,
lifespan=lifespan,
docs_url="/api/docs",
openapi_url="/api/openapi.json",
)
settings = get_settings()
app.mount("/assets", StaticFiles(directory=settings.web_root / "assets"), name="assets")
API_REVISION = 13
engine_registry = EngineRegistry.load_default(settings.engine_spec_directory)
engine_executor = EngineExecutor(engine_registry, settings.model_deployment_file)
def _engine_runtime() -> dict[str, dict[str, object]]:
return engine_executor.runtime_catalog()
[docs]
@app.get("/", include_in_schema=False)
async def index() -> FileResponse:
return FileResponse(settings.web_root / "index.html")
[docs]
@app.get("/batch", include_in_schema=False)
async def batch_index() -> FileResponse:
return FileResponse(settings.web_root / "batch.html")
[docs]
@app.get("/api/health")
async def health() -> dict[str, object]:
runtime = _engine_runtime()
return {
"status": "ok",
"version": __version__,
"location_provider": settings.location_provider,
"api_revision": API_REVISION,
"capabilities": {
"viewport_buildings": True,
"aligned_building_grid": True,
"reference_dataset_contract": True,
"batch_generation": True,
"batch_resume_and_extend": True,
"empty_scene_mode": True,
"selective_case_artifacts": True,
"radio_metrics": ["path_gain", "rss", "sinr", "association"],
"building_provider": "source_registry",
"building_sources": True,
"native_building_geometry": True,
"terrain_provider": "usgs_3dep_with_global_fallback",
"terrain_following_radio_map": True,
"semantic_surface_materials": True,
"extensible_engine_registry": True,
"compiled_scene_cache": True,
},
"engines": {entry["id"]: entry["runtime"] for entry in engine_registry.catalog(runtime)},
}
[docs]
@app.get("/api/engines")
async def engines() -> list[dict[str, object]]:
return engine_registry.catalog(_engine_runtime())
[docs]
@app.post("/api/batches/plan", response_model=BatchPlan)
async def plan_batch(payload: BatchDatasetRequest) -> BatchPlan:
return build_batch_plan(payload)
async def _run_batch_job(
app: FastAPI,
payload: BatchDatasetRequest,
plan: BatchPlan,
resume: bool,
) -> None:
job = app.state.batch_jobs[plan.dataset_slug]
def update_progress(completed: int, total: int, message: str) -> None:
job.update(completed_cases=completed, total_cases=total, message=message)
try:
async with app.state.batch_lock:
job.update(status="running", message="Preparing shared scene assets")
output = await generate_batch_dataset(
payload,
plan,
settings.dataset_root,
app.state.provider,
app.state.provider.settings.max_buildings,
app.state.sionna_lock,
app.state.batch_worker,
update_progress,
resume=resume,
)
manifest = json.loads((output / "dataset_manifest.json").read_text(encoding="utf-8"))
job.update(
status=manifest["status"],
completed_cases=plan.total_cases,
message="Dataset generation completed",
output_directory=str(output),
)
except Exception as exc:
job.update(status="failed", message="Dataset generation failed", error=str(exc))
[docs]
@app.post("/api/batches", status_code=202)
async def start_batch(
request: Request,
payload: BatchDatasetRequest,
resume: bool = Query(default=False),
) -> dict[str, object]:
plan = build_batch_plan(payload)
if any(
case.request.engine == EngineName.sionna.value
for scene in plan.scenes
for case in scene.cases
):
runtime = _engine_runtime()[EngineName.sionna.value]
if not runtime["available"]:
raise HTTPException(status_code=409, detail=str(runtime["message"]))
output = settings.dataset_root.expanduser().resolve() / plan.dataset_slug
existing_job = request.app.state.batch_jobs.get(plan.dataset_slug)
if existing_job and existing_job["status"] in {"queued", "running"}:
raise HTTPException(
status_code=409,
detail=f"Dataset '{plan.dataset_slug}' is already being generated",
)
if output.exists() and not resume:
raise HTTPException(
status_code=409,
detail=(
f"Dataset '{plan.dataset_slug}' already exists; choose a new dataset name "
"or enable resume/extend"
),
)
job = {
"job_id": plan.dataset_slug,
"dataset_name": plan.dataset_name,
"status": "queued",
"completed_cases": 0,
"total_cases": plan.total_cases,
"message": "Waiting for the batch worker",
"output_directory": None,
"error": None,
}
request.app.state.batch_jobs[plan.dataset_slug] = job
task = asyncio.create_task(_run_batch_job(request.app, payload, plan, resume))
request.app.state.batch_tasks.add(task)
task.add_done_callback(request.app.state.batch_tasks.discard)
return job
[docs]
@app.get("/api/batches")
async def list_batches(request: Request) -> list[dict[str, object]]:
return list(reversed(request.app.state.batch_jobs.values()))
[docs]
@app.get("/api/batches/{job_id}")
async def batch_status(request: Request, job_id: str) -> dict[str, object]:
job = request.app.state.batch_jobs.get(job_id)
if job is None:
raise HTTPException(status_code=404, detail="Batch job was not found")
return job
[docs]
@app.get("/api/batches/{job_id}/manifest")
async def batch_manifest(request: Request, job_id: str) -> FileResponse:
job = request.app.state.batch_jobs.get(job_id)
if job is None:
raise HTTPException(status_code=404, detail="Batch job was not found")
if job["status"] not in {"completed", "completed_with_errors"}:
raise HTTPException(status_code=409, detail="Batch dataset is not complete")
path = Path(str(job["output_directory"])) / "dataset_manifest.json"
return FileResponse(path, filename=f"{job_id}-manifest.json")
[docs]
@app.get("/api/search", response_model=list[SearchResult])
async def search(
request: Request, q: str = Query(min_length=2, max_length=120)
) -> list[SearchResult]:
try:
return await request.app.state.provider.search(q)
except httpx.HTTPError as exc:
raise HTTPException(
status_code=502, detail="Location search is temporarily unavailable"
) from exc
[docs]
@app.get("/api/building-sources")
async def building_sources(request: Request) -> list[dict[str, object]]:
provider = request.app.state.provider
sources = provider.building_sources.describe()
for source in sources:
source["automatic"] = (
source["automatic"] and source["id"] in provider.settings.automatic_building_sources
)
return [
{
"id": "auto",
"title": "Automatic",
"source_ids": list(provider.settings.automatic_building_sources),
},
{"id": "global", "title": "Global (Overture → OpenStreetMap)"},
*sources,
]
[docs]
@app.get("/api/buildings", response_model=BuildingContextResponse)
async def building_context(
request: Request,
latitude: float = Query(ge=-90, le=90),
longitude: float = Query(ge=-180, le=180),
radius_m: int = Query(default=750, ge=100, le=2500),
building_source: str = Query(default="auto", pattern=r"^(?:[a-z][a-z0-9_-]*|3dbag)$"),
) -> BuildingContextResponse:
warnings: list[str] = []
try:
result = await request.app.state.provider.buildings(
latitude,
longitude,
radius_m,
request.app.state.provider.settings.max_view_buildings,
source=building_source,
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except BuildingProviderError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
view_limit = request.app.state.provider.settings.max_view_buildings
limit_reached = result.source_count >= view_limit
warnings.extend(result.warnings)
if limit_reached:
warnings.append("Building provider limit reached; visible geometry may be incomplete.")
try:
geometry_context = building_geometry_context(result.buildings, longitude, latitude)
except ValueError as exc:
raise HTTPException(
status_code=502,
detail=f"Building source returned geometry that cannot be placed: {exc}",
) from exc
provenance: dict[str, str | int | float | bool] = {
"building_provider": result.provider,
**result.source_provenance,
"query_radius_m": radius_m,
"scene_boundary_policy": SCENE_BOUNDARY_POLICY,
"source_building_count": result.source_count,
"building_count": len(result.buildings),
"boundary_excluded_building_count": result.boundary_excluded_count,
"building_limit": view_limit,
"building_limit_reached": limit_reached,
"persistent_source_storage": False,
}
if result.release:
provenance["building_release"] = result.release
return BuildingContextResponse(
buildings=result.buildings,
building_geometry=geometry_context,
warnings=warnings,
data_provenance=provenance,
)
async def _compile_scene_or_raise(request: Request, payload: SceneRequest) -> CompiledScene:
"""Compile through the shared cache, mapping acquisition failures to HTTP errors."""
provider = request.app.state.provider
try:
return await compile_web_scene(payload, provider.settings, provider)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except BuildingProviderError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except (TerrainProviderError, SurfaceProviderError, httpx.HTTPError) as exc:
raise HTTPException(status_code=502, detail=f"Environment data failed: {exc}") from exc
[docs]
@app.post("/api/scenes", response_model=SceneResponse)
async def compile_scene_endpoint(request: Request, payload: SceneRequest) -> SceneResponse:
"""Compile a geographic scene into the cache shared with ``rt.load_scene``."""
scene = await _compile_scene_or_raise(request, payload)
return await asyncio.to_thread(scene_response, scene, request.app.state.provider.settings)
[docs]
@app.get("/api/scenes/{scene_id}", response_model=SceneResponse)
async def compiled_scene(request: Request, scene_id: str) -> SceneResponse:
"""Read a previously compiled scene back from the cache."""
settings_in_use = request.app.state.provider.settings
try:
scene = cached_scene(scene_id, settings_in_use)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except FileNotFoundError as exc:
raise HTTPException(status_code=404, detail="Scene is not in the cache") from exc
return await asyncio.to_thread(scene_response, scene, settings_in_use)
[docs]
@app.post("/api/simulations", response_model=SimulationResponse)
async def simulate(request: Request, payload: SimulationRequest) -> SimulationResponse:
started = time.perf_counter()
try:
payload.engine_config = engine_registry.validate_engine_config(
payload.engine,
payload.engine_config,
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
runtime = _engine_runtime().get(
payload.engine,
{"available": False, "message": "A server-side execution adapter is not configured."},
)
if not runtime.get("available"):
raise HTTPException(status_code=409, detail=str(runtime.get("message")))
scene = await _compile_scene_or_raise(request, payload.scene_request)
settings_in_use = request.app.state.provider.settings
reference = payload.scene_origin
buildings = scene.buildings.buildings
environment = scene.environment
warnings: list[str] = list(scene.buildings.warnings)
if scene.buildings.source_count >= settings_in_use.max_buildings:
warnings.append("Building provider limit reached; scene geometry may be incomplete.")
try:
execution = await engine_executor.run(
EngineContext(
payload,
buildings,
environment,
request.app.state.sionna_lock,
scene_assets=scene.assets,
)
)
except Exception as exc:
label = engine_registry.get(payload.engine).label
raise HTTPException(status_code=500, detail=f"{label} failed: {exc}") from exc
radio_data = execution.radio_data
engine_details = execution.details
warnings.extend(execution.warnings)
products = build_coverage_products(
payload,
radio_data,
buildings,
environment if payload.engine == EngineName.sionna.value else None,
)
response = SimulationResponse(
simulation_id=uuid.uuid4().hex[:12],
engine=payload.engine,
elapsed_ms=round((time.perf_counter() - started) * 1000),
coverage=products.coverage,
buildings=buildings,
building_geometry=placed_building_geometry_context(
scene.assets.local_buildings, reference.longitude, reference.latitude
),
warnings=warnings,
engine_details=engine_details,
data_provenance={
**scene_provenance(scene, settings_in_use),
"transmitter_count": len(payload.transmitters),
},
environment=environment_context(environment),
scene=scene_response(scene, settings_in_use),
)
request.app.state.simulations[response.simulation_id] = (
payload,
response,
radio_data,
products.building_mask,
)
request.app.state.environments[response.simulation_id] = environment
request.app.state.simulations.move_to_end(response.simulation_id)
while len(request.app.state.simulations) > 16:
expired_id, _ = request.app.state.simulations.popitem(last=False)
request.app.state.environments.pop(expired_id, None)
return response
[docs]
@app.get("/api/simulations/{simulation_id}/raster.png")
async def simulation_raster(
request: Request,
simulation_id: str,
metric: str = Query(default="rss", pattern="^(path_gain|rss|sinr|association)$"),
mask_buildings: bool = True,
feather_edges: bool = True,
) -> Response:
stored = request.app.state.simulations.get(simulation_id)
if stored is None:
raise HTTPException(status_code=404, detail="Simulation result is no longer available")
_, _, radio_data, building_mask = stored
content = render_radio_map_png(
radio_data,
metric,
building_mask,
mask_buildings=mask_buildings,
feather_edges=feather_edges,
)
return Response(
content=content,
media_type="image/png",
headers={"Cache-Control": "no-store"},
)
[docs]
@app.get("/api/simulations/{simulation_id}/artifacts.zip")
async def simulation_artifacts(request: Request, simulation_id: str) -> Response:
stored = request.app.state.simulations.get(simulation_id)
if stored is None:
raise HTTPException(status_code=404, detail="Simulation result is no longer available")
payload, result, radio_data, building_mask = stored
environment = request.app.state.environments.get(simulation_id)
bundle = build_artifact_bundle(payload, result, radio_data, building_mask, environment)
return Response(
content=bundle,
media_type="application/zip",
headers={
"Content-Disposition": f'attachment; filename="owrt-{simulation_id}.zip"',
},
)
