# Project export: Aegis
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: Optimal first responder routing for emergency and disaster
- Devpost: https://devpost.com/software/aegis-z0yqbs
- GitHub: https://github.com/A-Sanil/CalHacks
- Video: https://www.youtube.com/embed/Wg6uBZwcifE?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 3 GitHub contributor(s) — A-Sanil (5 commits), tim fan (2 commits), Nandan (1 commits)
## Devpost submission (written by the team)
### Inspiration
During a disaster, the safest route can change in seconds. A road may become blocked, a fire perimeter may expand, visibility may deteriorate, or a new emergency call may suddenly become the highest priority. Traditional navigation systems optimize for ordinary travel. Emergency responders need something different: a system that can understand rapidly changing field reports, account for vehicle capacity and survivor urgency, and continuously recalculate routes as conditions evolve. We built Aegis Rescue to explore one question: What if emergency routing could react to new disaster intelligence as quickly as responders receive it? In January of 2025 all of us were stuck within the LA area during the Pacific Palisades Fire. Not only was our group put in distress, many of our friends and family within the area were displaced and had to quickly evacuate. After seeing in real time the destruction and chaos caused by the fires, we were all inspired to create some safety measure to create a more optimized approach for first responders and the general public to ensure the best approach to minimizing the destruction of these wildfires which have been abundant over the last few decades within California. Our Palisades Fire scenario demonstrates this idea through a fire engine conducting search-and-rescue missions while an ambulance independently handles medical calls and returns to the hospital when it reaches capacity. What It Does Aegis Rescue is a real-time disaster logistics and route-optimization platform. It combines live hazard intelligence, a continuously updated road graph, priority-aware vehicle routing, and an interactive tactical dashboard. The system can: Convert unstructured reports—such as 911 calls, fire department updates, weather alerts, and social posts—into structured changes to the road network. Increase the cost of dangerous roads or mark impassable roads as closed. Prioritize rescue sites based on urgency, demand, fire exposure, and travel cost. Coordinate multiple vehicles with different capacities, starting locations, and missions. Recalculate routes when fire conditions change during a trip. Compare continuing toward a rescue with retreating to safety. Display active routes, alternatives, hazards, rescue sites, and vehicle movement on a live map. Fall back to local routing if the backend is unavailable, keeping the demonstration resilient. At a high level, Aegis minimizes travel and hazard exposure while maximizing the priority of the people reached: $$\text{Route Score} = \text{Travel Cost} \mathbin{+} \lambda \cdot \text{Hazard Exposure} \mathbin{-} \mu \cdot \text{Priority Served}$$ As fire conditions evolve, the cost of traveling across a road becomes: $$\mathrm{LiveEdgeCost}(i,j)=\mathrm{BaseCost}(i,j)\times\mathrm{HazardMultiplier}(i,j)$$ A multiplier near (1) represents a clear road, while a very large multiplier represents an impassable route. How We Built It We designed Aegis around two specialized “brains” connected through a shared live-state layer: An extractor agent interprets unstructured field intelligence and converts it into structured node and edge updates. A deterministic OR-Tools optimization engine solves the mathematical routing problem using vehicle capacities, priorities, time windows, and current road conditions. We intentionally use AI for understanding language, not for calculating routes. Once the incoming intelligence has been structured, the optimization engine makes reproducible and explainable routing decisions. Redis acts as the system’s nervous system. It stores the road graph, hazard multipliers, rescue priorities, vehicle states, optimization contracts, and completed routes. Redis Pub/Sub allows new intelligence and route updates to move through the system in real time. The backend is built with Python and FastAPI. It supports: Capacitated Vehicle Routing Problems Traveling Salesperson Problems Priority-aware rescue selection Vehicle capacities and time windows Nearest-neighbor and 2-opt fallback algorithms Optional machine-learning warm starts Real-time route updates through WebSockets The frontend is a tactical dashboard built with React, TypeScript, Vite, and MapLibre GL. It visualizes: Historical fire progression Rescue locations and medical calls Emergency vehicle positions Active and alternative routes Road closures and hazard exposure Dynamic rerouting decisions For a realistic demonstration, we processed historical data from: LAFD Palisades Fire progression data OpenStreetMap road data USGS 3DEP elevation data EPA AirNow historical air-quality observations Historical fire progression is replayed while vehicles are moving, forcing the optimizer to respond to conditions that change during an active mission. Turning Emergency Audio Into Map Updates With Deepgram We used Deepgram Nova-3 to convert live police and emergency-dispatch audio into structured information that Aegis can act on. Deepgram transcribes noisy radio traffic and 911 calls in real time. We improve transcription accuracy with key terms for local roads, landmarks, emergency vocabulary, and place names such as “Palisades Drive,” “road closed,” “trapped,” and “mandatory evacuation.” The audio pipeline: Deepgram converts emergency audio into text. Speaker diarization separates callers, dispatchers, and field units. The system filters transcripts for relevant locations, hazards, casualties, and road closures. Our extraction agent converts those details into structured map updates. New emergency locations are added to the map as response points. Road hazards update the live graph and trigger route optimization. Responders immediately see the new incident and updated route on the tactical dashboard. Deepgram also applies smart formatting to addresses and unit numbers while redacting sensitive information such as phone, payment-card, and Social Security numbers. This allows Aegis to transform spoken reports into actionable map intelligence within seconds—without requiring dispatchers to manually re-enter information. Challenges We Faced Routing Through a Changing Environment A route that was optimal when a responder departed could become unsafe moments later. We needed to update road costs without rebuilding the entire graph every time a new hazard appeared. We addressed this by separating static road costs from live hazard multipliers. A hazard report becomes a small, constant-time Redis update, while the solver reads the latest combined cost whenever it recalculates a route. Combining AI With Deterministic Optimization Language models are useful for interpreting reports such as: “The fire crossed Route 4, and the road is now impassable.” However, they should not be trusted to perform safety-critical route optimization directly. We created a strict JSON contract between the extraction layer and the solver. This gave us the flexibility of natural-language understanding while keeping routing deterministic, testable, and explainable. Coordinating Independent Vehicles The fire engine and ambulance have different bases, capacities, objectives, and return conditions. Each vehicle needed its own operational state and route while still reacting to the same evolving disaster. We modeled the vehicles independently while allowing them to share the same live hazard graph. This lets both missions run concurrently without coupling their operational logic. Rendering Solver Output on Real Roads Optimization solvers return ordered graph nodes, while an interactive map needs complete road geometry. We built path-reconstruction logic that expands optimized routes into drawable road segments, allowing the dashboard to accurately display the paths selected by the backend. Keeping the Demo Reliable Live systems have many possible failure points. We added: Deterministic fallbacks for intelligence extraction Multiple fallback routing algorithms Semantic caching for repeated or similar reports Client-side route previews when the backend is unavailable Automated tests for routing, capacity, caching, traversal, and real-time updates These safeguards allow the demonstration to continue functioning even when an individual service is unavailable. What We Learned We learned that disaster routing is not simply a shortest-path problem. It is a dynamic vehicle-routing problem involving: Uncertainty Vehicle capacity Survivor urgency Time windows Hazard exposure Multiple vehicle types Rapidly changing constraints We also learned the value of giving each technology a focused responsibility: AI translates human language into structured operational updates. Redis maintains fast-changing shared state. Optimization algorithms make constrained routing decisions. The tactical dashboard gives humans visibility into why routes change. Most importantly, we learned that resilient systems need graceful degradation. During an emergency, partial functionality is far more useful than a system that fails completely when one service becomes unavailable. What’s Next We would like to expand Aegis Rescue with: Live feeds from Caltrans, NOAA, NASA FIRMS, and emergency dispatch systems Speech-to-text ingestion for radio and 911 traffic Confidence scores and human approval for uncertain intelligence Additional resources such as shelters, helicopters, supply trucks, and evacuation buses Traffic, fuel, crew-hour, and terrain constraints Predictive fire-spread modeling Offline-first operation for damaged communications infrastructure Field testing with emergency-management professionals Our long-term vision is a decision-support platform that helps responders transform fragmented disaster intelligence into coordinated action—without removing humans from the loop.
## README (from the GitHub repository)
# 🛰️ Project Aegis Route
### Real-time agentic optimization for wildfire search & rescue — **voice-first**, grounded in the January 2025 Palisades Fire.
    -red) 
> **Turn the chaos of disaster comms into an O(1) live state — then reroute the instant someone speaks.**
Wildfire search-and-rescue runs on *voice*: 911 calls, dispatch radio, field units on the net. Aegis listens to that traffic, extracts what changed (a bridge is out, six trapped at the high school), writes it to a live graph in **O(1)**, and re-optimizes multi-vehicle rescue routes — continuously. Then it **speaks the new plan back** to the crews.
---
## 🎤 Voice is the front door — powered by Deepgram
Voice isn't bolted on; it's the **primary I/O of the whole system**. Without speech-to-text there is no data to optimize on. Aegis uses Deepgram across four load-bearing surfaces:
| Deepgram product | Where it lives in Aegis | Why it's essential |
|---|---|---|
| **STT — Nova-3** | The front door: 911 calls + dispatch radio → `ingest.Signal` → negative edge weights | No transcript = no live state = nothing to solve |
| **STT streaming** | Live radio-scanner feed → graph updates as the call is still happening | Reroutes mid-incident |
| **TTS — Aura-2** | The solver's route → spoken advisory to field units | Hands-free instructions in a moving truck; closes the loop |
| **Voice Agent API** | *Talk* to Aegis: "new emergency at grid E5" → it acts + replies | Voice as the control surface — the marquee demo |
**SAR-tuned, not a vanilla call:** keyterm prompting (street names + survival vocab survive radio noise), speaker **diarization** (caller vs dispatcher), `smart_format`/`numerals` for addresses & unit numbers, and PII **redaction** for real 911 audio.
### ⭐ The marquee moment — reroute by voice, on a coordinate grid
Mid-mission, the dispatcher just talks. The **Deepgram Voice Agent** (one socket: Nova-3 → an LLM with function-calling → Aura-2) decides to call an Aegis tool:
> **Dispatcher (out loud):** "Aegis, new emergency — two people trapped at grid **echo five**, critical."
> **→** agent calls `add_sar_location(grid="E5", demand=2, priority="critical")`
> **→** a new node drops onto the coordinate grid in Redis, wired to its nearest roads
> **→** the **real OR-Tools solver** re-optimizes over live edge costs
> **Aegis (spoken back):** "Copy. New rescue site at grid E5, node 10, priority critical, two victims. Rerouting 2 units, total cost 43."
```text
🎙️ speech ──Nova-3──▶ intent + args ──▶ add_sar_location / block_road / reroute
│
O(1) edit ▼ (new node + edges on the grid)
Redis live state
│ build_contract → run_solver (OR-Tools)
▼
new routes ──Aura-2──▶ 🔊 spoken advisory + 🗺️ dashboard
```
Grid squares use NATO phonetics too — "echo five", "E-5", and `E5` all resolve to the same lat/lng inside the Palisades operating box.
---
## 🏗️ Architecture
```text
AUDIO BRAIN STATE (O(1)) SOLVE VOICE OUT
911 call ─┐
radio ────┼─▶ Deepgram ─▶ Extractor Agent ─▶ Redis live layer ─▶ build_contract ─▶ Aura-2 TTS 🔊
field ────┘ STT (+ semantic base × multiplier run_solver + pub/sub 🗺️
cache) node priorities (OR-Tools CVRP)
▲
GraphRAG static knowledge graph
(roads, terrain — indexed once)
```
Two layers, decoupled on purpose:
- **GraphRAG = static** knowledge graph (roads, terrain) — pre-indexed once.
- **Redis = live** state — real-time edge weights & node priorities, O(1) updates, independent of when you solve.
---
## ⚡ Quickstart
```bash
python -m venv .venv && .venv\Scripts\activate # (Windows) | source .venv/bin/activate
pip install -r requirements.txt
pytest -q # 46 tests, fully offline (fakeredis, no API keys needed)
python voice_demo.py # the Deepgram story: 911 calls → reroutes → spoken advisories
```
Everything runs **offline by default** — `fakeredis`, sidecar transcripts, and a graceful solver fallback — so the demo always works on stage. Add a Deepgram key to go live (below).
---
## 🗂️ The voice layer
| Module | Role |
|---|---|
| `voice/deepgram_stt.py` | **Ears.** Audio → transcript → `ingest.Signal` (batch + live streaming, SAR keyterms, diarize, redact) |
| `voice/deepgram_tts.py` | **Mouth.** Advisory text → spoken Aura-2 audio for field units |
| `voice/voice_agent.py` | **Brain + voice.** Deepgram Voice Agent with function-calling tools: `add_sar_location`, `block_road`, `reroute`, `status` |
| `voice/grid.py` | Coordinate grid ↔ lat/lng (A–H × 1–8), NATO phonetics, nearest-node wiring |
| `voice/dispatch_pipeline.py` | The full loop: `process_call(audio)` → STT → extractor → Redis → solver → TTS |
| `voice/samples/` | Realistic 911/dispatch transcripts (hit real graph roads & nodes) |
The same tool dispatch backs both the **live** Voice Agent (`run_live()`) and the **offline** path (`handle_text()`) — so what you test offline is exactly what runs live.
---
## 🧪 Tests & metrics
- **46 tests passing** (38 system + 8 new voice) — `pytest -q`, no keys required.
- **Stress:** 12,000 ticks → 480 solves, **96% semantic-cache hit**, **4.84M tokens saved**, all invariants held.
- Voice-triggered reroute calls the **real OR-Tools CVRP** solver with a Dijkstra live-cost matrix — a blocked road or a new node reroutes via true shortest path.
---
## 🔑 Going live with Deepgram
```bash
export DEEPGRAM_API_KEY=dg_xxx # unlocks Nova-3 STT, Aura-2 TTS, Voice Agent
python voice/make_sample_audio.py # synth the sample calls to real .wav (Aura-2)
python voice_demo.py # now transcribes REAL audio with Nova-3
python voice_demo.py --live # talk to Aegis: live Voice Agent (needs a mic)
```
Optional env: `DEEPGRAM_STT_MODEL` (default `nova-3`), `DEEPGRAM_TTS_MODEL` (default `aura-2-thalia-en`), `AEGIS_AGENT_LLM` (Voice Agent think model).
---
## 📚 More docs
- [COMPLETE.md](COMPLETE.md) — full system walkthrough & PDF→code mapping
- [INTEGRATION.md](INTEGRATION.md) — solver ↔ frontend wiring
- [DEMO.md](DEMO.md) — live demo script
- [SLIDES.md](SLIDES.md) — presentation deck (also rendered: `Project-Aegis-Route.pptx`)
---
Built for CalHacks. GraphRAG + Redis + LLM agent + OR-Tools + React — with **Deepgram** as the voice that drives it all.
## Detected evidence (automated analysis)
Indexed codebase: 93 recognized source files, 330 KB.
- CSS (language) — detected in the code
- FastAPI (technology) — detected in the code
- HTML (language) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — detected in the code
- React (technology) — detected in the code
- Redis (technology) — detected in the code
- TypeScript (language) — detected in the code
- OpenAI (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (120 of 121)
```
.gitignore
COMPLETE.md
conftest.py
contract_builder.py
data/processed/palisades/fire_timeline.geojson
data/processed/palisades/hazard_route_matrix.json
data/processed/palisades/replay_timeline.json
data/processed/palisades/road_graph_25.json
data/processed/palisades/roads_display.geojson
data/processed/palisades/smoke_pm25_timeline.json
data/raw/palisades/airnow_20250107.dat
data/raw/palisades/airnow_20250108.dat
data/raw/palisades/airnow_20250109.dat
data/raw/palisades/airnow_20250110.dat
data/raw/palisades/airnow_20250111.dat
data/raw/palisades/airnow_monitoring_sites.dat
data/raw/palisades/elevation_usgs_3dep.tif
data/raw/palisades/fire_progression_lafd.geojson
data/raw/palisades/roads_osm_overpass.json
data/raw/palisades/SOURCES.md
DEMO.md
demo.py
extractor_agent.py
frontend/.env.example
frontend/.gitignore
frontend/index.html
frontend/package.json
frontend/README.md
frontend/src/App.tsx
frontend/src/components/DemoGuide.tsx
frontend/src/components/DynamicMissionMap.tsx
frontend/src/components/IncidentFeed.tsx
frontend/src/components/NodeSelector.tsx
frontend/src/components/OperationsSidebar.tsx
frontend/src/components/PalisadesMap.tsx
frontend/src/components/Shared.tsx
frontend/src/components/TacticalMap.tsx
frontend/src/data/scenario.ts
frontend/src/enhancements.css
frontend/src/main.tsx
frontend/src/services/solver.ts
frontend/src/simulation/dynamicMission.ts
frontend/src/simulation/graph.ts
frontend/src/simulation/palisadesMission.ts
frontend/src/styles.css
frontend/src/types.ts
frontend/src/vite-env.d.ts
frontend/tsconfig.json
frontend/vite.config.ts
graph_seed.py
graph_traversal.py
ingest.py
INTEGRATION.md
pyproject.toml
README.md
redis_store.py
requirements.txt
samples/sample_optimize_request.json
scripts/precompute_palisades_hazards.py
scripts/prepare_palisades_demo.py
scripts/replay_palisades_to_redis.py
scripts/run_backend.ps1
scripts/run_backend.sh
scripts/run_frontend.ps1
scripts/run_frontend.sh
scripts/seed_redis.py
scripts/simulate_dynamic_routes.py
scripts/train_model.py
semantic_cache.py
sim_realtime.py
SLIDES.md
solver_bridge.py
solver/__init__.py
solver/adapters/__init__.py
solver/adapters/aegis_contract.py
solver/algorithms/__init__.py
solver/algorithms/cvrp.py
solver/algorithms/ml_warmstart.py
solver/algorithms/nearest_neighbor.py
solver/algorithms/path_utils.py
solver/algorithms/tsp_lkh.py
solver/algorithms/two_opt.py
solver/api/app.py
solver/api/frontend_graph.py
solver/api/optimize.py
solver/config.py
solver/core/__init__.py
solver/core/matrix.py
solver/core/schema.py
solver/core/solver.py
solver/redis_client.py
solvers/__init__.py
solvers/your_solver_adapter.py
stress.py
tests/conftest.py
tests/test_algorithms.py
tests/test_api.py
tests/test_contract.py
tests/test_integration.py
tests/test_optimize_endpoint.py
tests/test_pubsub.py
tests/test_realtime.py
tests/test_semantic_cache.py
tests/test_solver_bridge.py
tests/test_solver.py
tests/test_store.py
tests/test_traversal.py
tests/test_visit_penalties.py
tests/test_voice.py
viz/index.html
voice_demo.py
voice/__init__.py
voice/deepgram_stt.py
voice/deepgram_tts.py
voice/dispatch_pipeline.py
voice/grid.py
voice/make_sample_audio.py
voice/samples/911_canyon_road_fire.txt
voice/samples/911_highschool_trapped.txt
voice/samples/911_hwy9_bridge.txt
[1 more files omitted for size]
```
### Dependencies
- frontend/package.json: @turf/simplify@^7.3.5, @types/react@latest, @types/react-dom@latest, @vitejs/plugin-react@latest, maplibre-gl@^5.24.0, react@latest, react-dom@latest, typescript@latest, vite@latest
- pyproject.toml: elkai@>=2.0, fakeredis@>=2.21, fastapi@>=0.110, httpx@>=0.27, httpx@>=0.27, numpy@>=1.26, ortools@>=9.9, pydantic@>=2.6, pytest@>=8.0, pytest-asyncio@>=0.23, redis@>=5.0, shapely@>=2.1, torch@>=2.2, uvicorn[standard]@>=0.27, websockets@>=12.0
- requirements.txt: deepgram-sdk@>=3.7, elkai@>=2.0, fakeredis@>=2.21, fastapi@>=0.110, httpx@>=0.27, numpy@>=1.26, ortools@>=9.9, pydantic@>=2.6, pytest@>=8.0, pytest-asyncio@>=0.23, redis@>=5.0, shapely@>=2.1, torch@>=2.2, uvicorn[standard]@>=0.27, websockets@>=12.0
### Recent commits (newest first)
- Add Deepgram voice layer: STT 911 ingest, Aura-2 advisories, Voice Agent reroute-by-grid; polish README
- Update SLIDES.md to reflect the full final system
- Add concurrent ambulance workflow and demo controls
- Build final Palisades rescue operations demo
- Wire frontend to live OR-Tools solver via /optimize endpoint
- complete: assemble solver backend + nested frontend/ + adapter scaffold
- add solver
- Project Aegis Route: full system (code, tests, docs)
## Key source files (fetched from GitHub, selected and truncated for size)
### INTEGRATION.md
```markdown
# Integrating your solver
Aegis decouples **state** (Redis, O(1) live edge/node updates) from **solving**
(on demand). Your friend's solver only has to speak one small contract.
## The seam
`solver_bridge.run_solver(store, solver_fn)` does three things:
1. reads the contract from Redis (`aegis:contract:latest`),
2. calls `solver_fn(contract, store)`,
3. writes the returned solution back to Redis (`aegis:routes:latest`) **and**
publishes `aegis:routes:update` for the dashboard.
So your only job is to provide `solver_fn(contract, store) -> solution`.
## Input: the contract
```python
{
"timestamp": float,
"disaster_state": "escalating_wildfire",
"vehicles": [
{"id": "V1", "type": "air_evac", "capacity": 4, "start_node": 0}
],
"target_nodes": [
{"id": 7, "priority": 0.92, "demand": 3, "time_window": [0, 30]}
],
"dynamic_edge_modifiers": [
{"edge": [2, 3], "multiplier": 4.0, "reason": "fire jumped Hwy 9 bridge"}
]
}
```
## Live edge weights (the whole point of Redis)
Never hard-code distances. Ask Redis for the **current** cost:
```python
cost = store.get_final_cost(i, j) # base_cost * live multiplier, or None if no edge
```
The helper `solvers/your_solver_adapter.build_cost_matrix(contract, store)` turns
all contract nodes into a dense matrix for VRP-style solvers in one call.
## Output: the solution (return EXACTLY this)
```python
{
"vehicles": [
{"id": "V1", "route": [0, 7, 3], "load": 3, "cost": 12.5}
],
"total_cost": 12.5,
"dropped": [9] # target ids you could not serve
}
```
## Wire it up
```python
from solver_bridge import run_solver
from solvers.your_solver_adapter import adapter
solution = run_solver(store, adapter)
```
Fill the three `TODO` blocks in `solvers/your_solver_adapter.py`:
1. import the solver, 2. call it, 3. map its output to the schema above.
A capacity-aware greedy `mock_solver` already lives in `solver_bridge.py` so the
loop is testable before the real solver lands.
```
### DEMO.md
```markdown
# Aegis Rescue: Palisades Fire demo
## Scenario
Fire Station 6 is the fixed safe base outside the cumulative Palisades Fire footprint. Three rescue sites—N9, N15, and N21—begin inside the fire.
The demo repeats one clear operational loop:
1. Rank the waiting rescue sites using fire-distance priority and round-trip route cost.
2. Leave Fire Station 6 for one selected site.
3. Follow the lowest-cost route, balancing travel time against distance and duration inside fire.
4. Load the survivors at the site.
5. Return to Fire Station 6 before beginning another rescue.
6. Mark the site safe only after the responder reaches the station.
The mission ends after all three groups return to the station.
## Independent ambulance
A second vehicle starts at Hospital 14 and runs concurrently with the fire truck. It has its own position, route, fire-weight updates, medical-call queue, and blue/purple map styling.
The ambulance workflow is:
1. Leave Hospital 14 for the lowest-cost pending medical call.
2. Collect one person and continue directly to another call.
3. At 3/3 onboard, stop collecting and return to Hospital 14 automatically.
4. Unload all three people.
5. Resume the remaining medical-call route from the hospital.
6. Make a final hospital return when the queue is empty, even if below capacity.
The demo uses five medical calls so both the full-capacity return and post-drop-off route resumption are visible.
## Dynamic rerouting
Historical LAFD fire progression continues during each trip. Fire-exposed edges remain traversable, but their risk and travel-time weights rise for every interval spent inside the cumulative footprint.
The first eligible mid-travel update adds a small deterministic demo flare-up at the responder. The current road edge is replaced by a temporary yellow `LIVE N1000` node and proportional forward/backward segments. The optimizer compares:
- distance remaining toward the site or station;
- distance required to retreat;
- current fire exposure;
- alternate road costs;
- rescue-site priority.
This makes the continue-versus-retreat decision distance-dependent without restarting the responder from the previous node.
## Visual language
- Green marker: Fire Station 6
- Red markers: rescue sites inside fire
- Cyan route segments: clear travel
- Amber to red route segments: increasing fire exposure
- Dashed routes: second- and third-lowest-cost alternatives
- White/cyan marker: responder
- White/blue marker and blue-purple route: ambulance
- Blue marker: Hospital 14
- Purple markers: medical calls
- Yellow ring: temporary live split node
## Run
Open `http://127.0.0.1:5173`, click **Start rescue demo**, and watch each complete station-to-site-to-station cycle. Use **Reset** to restart the scenario.
```
### requirements.txt
```
fastapi>=0.110
uvicorn[standard]>=0.27
pydantic>=2.6
redis>=5.0
numpy>=1.26
ortools>=9.9
elkai>=2.0
torch>=2.2
websockets>=12.0
httpx>=0.27
shapely>=2.1
# dev / tests
pytest>=8.0
pytest-asyncio>=0.23
fakeredis>=2.21
deepgram-sdk>=3.7 # voice: STT (nova-3), TTS (aura-2), Voice Agent API
```
### pyproject.toml
```
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "aoe-solver"
version = "0.1.0"
description = "Agentic Optimization Engine — real-time CVRP/TSP solver"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"fastapi>=0.110",
"uvicorn[standard]>=0.27",
"pydantic>=2.6",
"redis>=5.0",
"numpy>=1.26",
"ortools>=9.9",
"elkai>=2.0",
"torch>=2.2",
"websockets>=12.0",
"httpx>=0.27",
"shapely>=2.1",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
"fakeredis>=2.21",
"httpx>=0.27",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["solver*"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
```
### frontend/package.json
```
{
"name": "aegis-route-dashboard",
"private": true,
"version": "0.1.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "tsc -b && vite build",
"preview": "vite preview"
},
"dependencies": {
"@turf/simplify": "^7.3.5",
"maplibre-gl": "^5.24.0",
"react": "latest",
"react-dom": "latest"
},
"devDependencies": {
"@types/react": "latest",
"@types/react-dom": "latest",
"@vitejs/plugin-react": "latest",
"typescript": "latest",
"vite": "latest"
}
}
```
### frontend/src/main.tsx
```typescript
import { StrictMode } from 'react'
import { createRoot } from 'react-dom/client'
import App from './App'
import './styles.css'
createRoot(document.getElementById('root')!).render(
,
)
```
### solver/api/app.py
```python
from __future__ import annotations
import asyncio
import json
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from pathlib import Path
from solver.config import settings
from solver.core.matrix import CostMatrixBuilder
from solver.core.schema import ProblemPayload, SolveResponse
from solver.core.solver import SolverEngine
from solver.api.optimize import router as optimize_router
from solver.redis_client import create_redis_client
class ConnectionManager:
"""Broadcasts route updates to dashboard clients."""
def __init__(self) -> None:
self._connections: list[WebSocket] = []
self._lock = asyncio.Lock()
async def connect(self, websocket: WebSocket) -> None:
await websocket.accept()
async with self._lock:
self._connections.append(websocket)
async def disconnect(self, websocket: WebSocket) -> None:
async with self._lock:
if websocket in self._connections:
self._connections.remove(websocket)
async def broadcast(self, message: dict[str, Any]) -> None:
dead: list[WebSocket] = []
async with self._lock:
connections = list(self._connections)
for ws in connections:
try:
await ws.send_json(message)
except Exception:
dead.append(ws)
for ws in dead:
await self.disconnect(ws)
def _incoming_event(payload: ProblemPayload) -> dict[str, Any]:
return {
"type": "update_received",
"timestamp": payload.timestamp.isoformat(),
"disaster_state": payload.disaster_state,
"edge_modifiers_applied": [
{"edge": list(m.edge), "multiplier": m.multiplier, "reason": m.reason}
for m in payload.dynamic_edge_modifiers
],
}
async def _solve_and_broadcast(payload: ProblemPayload) -> SolveResponse:
await manager.broadcast(_incoming_event(payload))
engine = get_engine()
result = engine.solve(payload)
await manager.broadcast(result.to_dashboard_event())
return result
VIZ_DIR = Path(__file__).resolve().parents[2] / "viz"
PALISADES_DATA_DIR = Path(__file__).resolve().parents[2] / "data" / "processed"
manager = ConnectionManager()
_engine: SolverEngine | None = None
def get_engine() -> SolverEngine:
global _engine
if _engine is None:
try:
client = create_redis_client()
client.ping()
_engine = SolverEngine.from_redis(client)
except Exception:
_engine = SolverEngine(CostMatrixBuilder())
return _engine
@asynccontextmanager
async def lifespan(app: FastAPI):
# Pre-load solver (and ML model) at startup
get_engine()
yield
if _engine is not None:
_engine._matrix_builder.invalidate_cache()
app = FastAPI(
title="AOE Solver",
description="Real-time CVRP/TSP solver for Agentic Optimization Engine",
version="0.1.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Live dashboard solve path (frontend OptimizationRequest -> real CVRP solver)
app.include_router(optimize_router)
if VIZ_DIR.is_dir():
app.mount("/viz", StaticFiles(directory=str(VIZ_DIR), html=True), name="viz")
if PALISADES_DATA_DIR.is_dir():
app.mount("/data", StaticFiles(directory=str(PALISADES_DATA_DIR)), name="processed-data")
@app.get("/")
async def root() -> RedirectResponse:
return RedirectResponse(url="/viz/")
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok", "service": "aoe-solver"}
@app.post("/solve", response_model=SolveResponse)
async def solve(payload: ProblemPayload) -> SolveResponse:
"""
Step C entrypoint: accept JSON matrix payload, return optimal routes.
Also broadcasts to WebSocket subscribers for live dashboard updates.
"""
try:
result = await _solve_and_broadcast(payload)
except KeyError as exc:
raise HTTPException(
status_code=503,
detail=f"Baseline matrix unavailable: {exc}. Seed Redis or pass baseline_matrix.",
) from exc
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
return result
@app.websocket("/ws/routes")
async def websocket_routes(websocket: WebSocket) -> None:
"""
Dashboard integration socket.
Clients receive `route_update` events after each /solve call.
Clients may also send ProblemPayload JSON to trigger inline re-solve.
"""
await manager.connect(websocket)
try:
while True:
raw = await websocket.receive_text()
data = json.loads(raw)
if data.get("type") == "ping":
await websocket.send_json({"type": "pong"})
continue
payload = ProblemPayload.model_validate(data)
await _solve_and_broadcast(payload)
except WebSocketDisconnect:
await manager.disconnect(websocket)
except Exception as exc:
await manager.disconnect(websocket)
raise exc
def main() -> None:
import uvicorn
uvicorn.run(
"solver.api.app:app",
host=settings.solver_host,
port=settings.solver_port,
reload=False,
)
if __name__ == "__main__":
main()
```
### frontend/src/App.tsx
```typescript
import { useEffect, useMemo, useRef, useState } from 'react'
import simplify from '@turf/simplify'
import { PalisadesMap } from './components/PalisadesMap'
import './enhancements.css'
import { averageSmoke, evaluateEdges, optimizeRouteAlternatives, priorityFromFireDistance, type HazardSnapshot, type MissionPlan, type ReplayEvent, type RoadGraph, type WeightedEdge } from './simulation/palisadesMission'
const API = import.meta.env.VITE_API_BASE_URL ?? 'http://127.0.0.1:8000'
const FIRE_STATION = 6
const RESCUE_SITES = [9, 15, 21]
const HOSPITAL = 14
const AMBULANCE_CAPACITY = 3
const MEDICAL_CALLS = [10, 24, 18, 0, 22]
const DEMO_START_FIRE_INDEX = 3
const EMPTY_PLAN: MissionPlan = { order: [], edges: [], coordinates: [], eta: 0, risk: 0, blockedEdges: 0 }
type MissionPhase = 'outbound' | 'returning' | 'complete'
type VirtualSplit = { id: number; position: [number, number]; originalSource: number; originalTarget: number; edges: WeightedEdge[] }
function initialFireState(events: ReplayEvent[]) {
const fireEvents = events.filter((event) => event.type === 'fire_perimeter')
const selected = fireEvents[Math.min(DEMO_START_FIRE_INDEX, fireEvents.length - 1)]
if (!selected) return { collection: null, timestamp: '', eventIndex: 0, latest: null }
const selectedIndex = events.indexOf(selected)
const features = events.slice(0, selectedIndex + 1).filter((event) => event.type === 'fire_perimeter').map((event) => simplify(event.payload, { tolerance: 0.00035, highQuality: false }))
return { collection: { type: 'FeatureCollection', features }, timestamp: selected.timestamp, eventIndex: selectedIndex + 1, latest: selected }
}
function localFirePatch([longitude, latitude]: [number, number]) {
const radiusKm = 0.55
const coordinates: [number, number][] = []
for (let step = 0; step <= 32; step++) {
const angle = step / 32 * Math.PI * 2
coordinates.push([longitude + Math.cos(angle) * radiusKm / (111 * Math.cos(latitude * Math.PI / 180)), latitude + Math.sin(angle) * radiusKm / 111])
}
return { type: 'Feature', properties: { source: 'demo_local_flare_up' }, geometry: { type: 'Polygon', coordinates: [coordinates] } }
}
function App() {
const [graph, setGraph] = useState(null)
const [events, setEvents] = useState([])
const [hazardMatrix, setHazardMatrix] = useState>({})
const [eventIndex, setEventIndex] = useState(0)
const [running, setRunning] = useState(false)
const [started, setStarted] = useState(false)
const [currentNode, setCurrentNode] = useState(FIRE_STATION)
const [remainingSites, setRemainingSites] = useState([...RESCUE_SITES])
const [rescuedSites, setRescuedSites] = useState([])
const [activeSite, setActiveSite] = useState(null)
const [phase, setPhase] = useState('outbound')
const [fire, setFire] = useState(null)
const [fireTimestamp, setFireTimestamp] = useState('')
const [pm25, setPm25] = useState(0)
const [latest, setLatest] = useState(null)
const [loadError, setLoadError] = useState('')
const [movement, setMovement] = useState(null)
const [movementDelay, setMovementDelay] = useState(650)
const [interrupted, setInterrupted] = useState(false)
const [virtualSplit, setVirtualSplit] = useState(null)
const [ambulanceNode, setAmbulanceNode] = useState(HOSPITAL)
const [medicalCalls, setMedicalCalls] = useState([...MEDICAL_CALLS])
const [collectedCalls, setCollectedCalls] = useState([])
const [ambulanceOnboard, setAmbulanceOnboard] = useState(0)
const [ambulanceDelivered, setAmbulanceDelivered] = useState(0)
const [ambulancePhase, setAmbulancePhase] = useState<'collecting' | 'returning' | 'complete'>('collecting')
const [ambulanceMovement, setAmbulanceMovement] = useState(null)
const [ambulanceDelay, setAmbulanceDelay] = useState(850)
const [playbackSpeed, setPlaybackSpeed] = useState(0.65)
const ambulanceProgress = useRef<{ position: [number, number]; segmentIndex: number; remainingRatio: number } | null>(null)
const movementProgress = useRef<{ position: [number, number]; segmentIndex: number; remainingRatio: number } | null>(null)
const virtualNodeSequence = useRef(1000)
const midpathDemoTriggered = useRef(false)
useEffect(() => {
Promise.all([
fetch(`${API}/data/palisades/road_graph_25.json`).then((response) => response.json()),
fetch(`${API}/data/palisades/replay_timeline.json`).then((response) => response.json()),
fetch(`${API}/data/palisades/hazard_route_matrix.json`).then((response) => response.json()),
]).then(([nextGraph, nextEvents, nextMatrix]) => {
const seed = initialFireState(nextEvents)
setGraph(nextGraph); setEvents(nextEvents); setHazardMatrix(nextMatrix)
setFire(seed.collection); setFireTimestamp(seed.timestamp); setEventIndex(seed.eventIndex); setLatest(seed.latest)
}).catch(() => setLoadError('Start the API on port 8000, then refresh.'))
}, [])
const snapshot = hazardMatrix[fireTimestamp]
const exposureByRoute = snapshot?.route_exposure_steps ?? {}
const distances = snapshot?.node_distances_km ?? {}
const priorities = useMemo(() => Object.fromEntries(graph?.nodes.map((node) => [node.id, priorityFromFireDistance(distances[String(node.id)])]) ?? []), [graph, distances])
const weighted = useMemo(() => {
if (!graph) return []
const base = evaluateEdges(graph, exposureByRoute, pm25)
if (!virtualSplit) return base
return [...base.filter((edge) => !((edge.source === virtualSplit.originalSource && edge.target === virtualSplit.originalTarget) || (edge.source === virtualSplit.originalTarget && edge.target === virtualSplit.originalSource))), ...virtualSplit.edges]
}, [graph, exposureByRoute, pm25, virtualSplit])
const rankedSites = useMemo(() => remainingSites.map((site) => {
const outbound = optimizeRouteAlt
[truncated — 15922 more characters]
```
### conftest.py
```python
import os, sys
sys.path.insert(0, os.path.dirname(__file__))
```
### stress.py
```python
"""stress.py -- heavy continuous run; invariants asserted EVERY tick."""
import time
from sim_realtime import run_simulation
def main(seeds=range(1, 16), ticks=800):
t0 = time.time()
agg = {"ticks": 0, "solves": 0, "tokens_saved": 0}
rates = []
for sd in seeds:
st = run_simulation(ticks=ticks, seed=sd, solve_every=25)
agg["ticks"] += st["ticks"]; agg["solves"] += st["solves"]
agg["tokens_saved"] += st["cache"]["tokens_saved"]; rates.append(st["cache"]["hit_rate_pct"])
dt = time.time() - t0
print(f"OK {agg['ticks']} live ticks across {len(list(seeds))} seeds in {dt:.1f}s "
f"({agg['ticks']/dt:.0f} ticks/s) -- ALL INVARIANTS HELD")
print(f" solver runs={agg['solves']} tokens_saved={agg['tokens_saved']:,} "
f"avg_cache_hit={sum(rates)/len(rates):.1f}%")
if __name__ == "__main__":
main()
```
[82 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]