# Project export: HERALD: Hospital Emergency Room Agentic Live Digital-twin 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: HERALD is an autonomous digital twin of a hospital emergency room where every patient, nurse, doctor, and piece of equipment is a live AI agent. - Devpost: https://devpost.com/software/herald-hospital-emergency-room-agentic-live-digital-twin - GitHub: https://github.com/RyanDang363/berk-ai-hackathon - Demo: https://agentverse.ai/agents/details/agent1qty576zgxtvhugg4a4gr7pdzrhcq89g78f3kszhmd70a9ftlsamcj6a6h3w/profile - Team: 4 GitHub contributor(s) — Evan (13 commits), Claude Opus 4.8 (1M context) (7 commits), RyanDang363 (6 commits), unknown (2 commits) ## Devpost submission (written by the team) ### Inspiration ERs run on fragmented whiteboards, pagers, and tribal knowledge — there's no live, queryable model of who's where and what's free right now. We wanted to see if every physical entity in a hospital — patient, nurse, doctor, bed, oxygen unit — could be its own autonomous agent reasoning in real time. The promise of a "digital twin" you can just talk to ("admit MRN-0005", "discharge MRN-0002") felt like the natural interface for high-pressure clinical ops. ### What it does Models an entire emergency room as cooperating uAgents in one Bureau, reachable from ASI:One chat, with a live admin dashboard mirroring every state change. Runs MRN-driven intake (triage → propose bed/nurse/doctor → admit) and discharge (propose sign-off → mark discharged → free resources on resolve), all with human-in-the-loop confirmation. Detects critical events like low-oxygen alerts, auto-dispatches the nearest free nurse, and records a structured incident trace of every action taken. ### How we built it Python 3.11 + uagents: a public Orchestrator agent (mailbox + Chat Protocol) plus private entity agents, all sharing state behind a StorageInterface (in-memory default, Redis swap). Event-registry pattern (plan → confirm → resolve) so every event proposes read-only first and only mutates state after admin approval — keeping the demo deterministic. A FastAPI + vanilla-JS dashboard with interactive proposal cards for real-time, human-in-the-loop control of the ER floor. ### Challenges we ran into Getting a mailbox-backed public agent to coexist with private agents in a single Bureau — we de-risked it with a spike before committing to the architecture. Premature state mutation: proposals were decrementing availability before confirmation, so we refactored intake/discharge to be fully read-only until commit. A subtle data bug where seeded patients had no MRN made them undischargeable — a reminder that MRN is the join key across every flow. ### Accomplishments we're proud of A genuinely conversational ER: you run the whole floor — admit, assign, discharge, resolve — from plain ASI:One chat, with the dashboard updating live. Clean separation of intent → plan → commit → resolve, giving deterministic demos and idempotent, testable handlers backed by EARS specs. End-to-end autonomy on the oxygen event: drop detected, nearest free nurse dispatched, and the equipment swap applied automatically. ### What we learned Async, message-passing agents force you to think in terms of state machines and idempotency, not request/response — every trigger must be safe to fire twice. Deferring all writes until confirmation makes a system both safer and far easier to demo under pressure. Spec-driven development (README → LLD → EARS → tests → code) kept a multi-person build from drifting into mismatched message names and keys. ### What's next Swap the in-memory store for Redis-backed persistence and multi-room scaling, with real-time vitals streaming from monitoring devices. Smarter triage and assignment via LLM reasoning over full EHR history, plus predictive alerts (deterioration, bed-capacity, staffing gaps). EHR/FHIR integration and audit-grade compliance so HERALD can move from synthetic demo to a real clinical pilot. ## README (from the GitHub repository) # ER Room Digital Twin An autonomous digital twin of a hospital emergency room. Every physical entity — patient, nurse, doctor, bed, equipment — is modeled as a **uAgent** running in a single [Fetch.ai Bureau](https://fetch.ai/). A single `OrchestratorAgent` (reachable through ASI:One) receives natural-language commands, reasons with the ASI:One LLM, and dispatches real asynchronous uAgent messages to coordinate the ER in real time. > This is not a dashboard that shows data. It is a system that **acts**. --- ## Getting Started **Prerequisites:** Python 3.11+, [`uv`](https://docs.astral.sh/uv/) (`brew install uv`). ```bash # 1. Clone and enter the repo git clone https://github.com/RyanDang363/berk-ai-hackathon.git cd berk-ai-hackathon # 2. Set up environment variables cp .env.example .env # Edit .env — for a no-API-key local run, leave USE_MOCK=true # 3. Install dependencies uv sync # 4. Run (mock mode — no API keys needed) USE_MOCK=true uv run python -m er_twin.main # 5. Run the tests uv run pytest ``` **Mock mode (`USE_MOCK=true`)** swaps in deterministic keyword intent routing instead of the ASI:One LLM, `InMemoryStore` instead of Redis, and `NoopMemory` instead of Iris — but still runs the **real** agent coordination in-process over deterministically seeded state. Replies are state-derived and reproducible with no API keys. **With API keys (`USE_MOCK=false`):** set `ASIONE_API_KEY`, `REDIS_URL`, and `AGENT_MEMORY_*` in `.env` for the live ASI:One LLM, Redis-backed state, and Iris agent memory. **Architecture details:** see [ARCHITECTURE.md](ARCHITECTURE.md) for the subsystem map, message flows, and design decisions. --- ## The Problem Emergency rooms operate in controlled chaos — every room, patient, nurse, doctor, and piece of equipment is a moving variable: - **Reactive triage** — staff only respond after bottlenecks form - **No real-time resource awareness** — nurses waste time locating equipment - **Manual bed assignment** — slow and error-prone under surge conditions - **Critical event delays** — no autonomous escalation when a patient deteriorates - **Siloed systems** — no single source of truth for room, staff, and equipment state --- ## Architecture **One process, one `Bureau`.** The public `OrchestratorAgent` (`mailbox=True`, Chat Protocol, reachable from ASI:One) runs **inside** the same Bureau as all private ER entity agents. They communicate over in-process uAgent messaging — no cross-process hop, no Almanac overhead. ``` ASI:One (public chat) │ ChatMessage ▼ ┌─────────────────────────────────────────────────────────┐ │ ONE Bureau · ONE process · ONE asyncio event loop │ │ │ │ OrchestratorAgent ──── in-process ctx.send ────► │ │ (mailbox + Chat Protocol) │ │ │ │ │ ▼ │ │ Admissions · Triage · Patient×3 · Bed×4 · │ │ Nurse×2 · Doctor×2 · Equipment │ │ │ │ │ ▼ │ │ StorageInterface MemoryInterface │ │ (InMemory|Redis) (Noop|Iris) │ └──────────┬──────────────────────────────────────────────┘ │ er:events out/replay/*.json ▼ ▼ Dashboard (FastAPI) Replay timeline (/replay/…) co-hosted in-process ``` **State** lives behind a `StorageInterface` (InMemoryStore first, RedisStore via `REDIS_URL`). **Memory** lives behind a `MemoryInterface` (NoopMemory or Iris). The dashboard is co-hosted on the same asyncio event loop and reads the live store directly. **Two write paths (intentional hybrid).** Chat commands (ASI:One or `USE_MOCK`) run as async uAgent message hops correlated by `flow_id`. Dashboard confirm/resolve call the same domain functions on the shared store directly — FastAPI handlers have no uAgent `Context` to `ctx.send`. Both paths leave the ER in the same state. ### Key Architecture Decisions | Decision | Rationale | |---|---| | **Single-process Bureau** | One process + one event loop = one seam to debug. The Orchestrator (`mailbox=True`) is a Bureau member; entity agents are messaged in-process. | | **Only Orchestrator on Agentverse** | Private agents have no mailbox and no Agentverse profile. Patient records stay in the local process. | | **StorageInterface abstraction** | One interface (`get / set / update / delete / list_ids / publish`) with two backends. InMemoryStore is the default; RedisStore is a zero-handler-change swap via `REDIS_URL`. | | **Async, fire-and-forget messaging** | Chat-path `ctx.send` calls return immediately; replies arrive in separate `@on_message` handlers and are correlated by `flow_id`. | | **CommandGate serialization** | The Orchestrator runs one chat command at a time; new commands are FIFO-queued while one is in flight. A 30-second watchdog releases the gate if a reply never lands. | | **Dashboard co-hosted** | The FastAPI dashboard is scheduled on the Bureau's existing event loop (`create_task` + `bureau.run()`). The shared store is injected directly — the floor map reads live state. Commands stay on chat; the dashboard is a live view plus confirm/resolve. | | **IrisMemory for agent memory** | Iris (Redis Agent Memory) appends session events for long-term semantic recall. Recording and recall are best-effort and non-fatal — a memory failure never aborts a live command. | --- ## The Three Events Each event is triggerable via a natural-language command to the `OrchestratorAgent` through ASI:One. | Event | Trigger Phrase | Agent Flow | |---|---|---| | **1. Patient Intake** | *"A new patient arrived with chest pain"* | Orchestrator → AdmissionsAgent (intake + MRN) → TriageAgent (acuity) → BedAgent (assign) → NurseAgent + DoctorAgent (assign) → confirmation to chat | | **2. Low Oxygen Alert** | *"Bed 3's patient oxygen is dropping"* | EquipmentAgent lowers supply + emits `LowSupplyAlert` → Orchestrator finds replacement unit → NurseAgent dispatched → oxygen swap applied → status confirmation | | **3. Status Summary** | *"Show me what's happening in the ER"* | Orchestrator reads live state across all agents → synthesizes summary via ASI:One LLM → returns to chat | Events 1 and 2 are async uAgent pipelines on the **chat** path (multi-hop `ctx.send` correlated by `flow_id`). Event 3 is a synchronous, read-only store query. Dashboard confirm/resolve for intake and discharge apply the same domain mutations without going through the message bus. --- ## Agent Roster > Instance counts stay small so a local run is deterministic: 3 PatientAgents, 2 NurseAgents, 2 DoctorAgents, 4 BedAgents, a handful of EquipmentAgents. | Agent | Responsibility | |---|---| | **OrchestratorAgent** | Mailbox + Chat Protocol + ASI:One reasoning. Only agent on Agentverse. Dispatches uAgent messages to all Bureau agents; serializes commands via CommandGate. | | **AdmissionsAgent** | Patient intake: deduplication by MRN/name, EHR enrichment, patient slot registration. | | **TriageAgent** | Chief-complaint → acuity score (ESI 1–5) + specialty routing. | | **PatientAgent** (×3 pooled) | Tracks acuity, vitals, status, assigned bed, and care team. Pool of 3 reusable slots; `bind` / `discharge` manage lifecycle. | | **BedAgent** (×4) | Tracks occupancy, specialty, attached equipment. Responds to assign and release requests. | | **NurseAgent** (×2) | Tracks availability, assignments, location. Accepts dispatch and release requests. | | **DoctorAgent** (×2) | Tracks specialty, load (cap 3), assignments. Paged for acuity ≤ 2 patients. | | **EquipmentAgent** | One age [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 76 recognized source files, 538 KB. - CSS (language) — detected in the code - FastAPI (technology) — detected in the code - HTML (language) — detected in the code - JavaScript (language) — detected in the code - OpenAI (technology) — detected in the code - Python (language) — detected in the code - Redis (technology) — detected in the code - AI coding agent: Claude Code — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (87 of 87) ``` .env.example .gitignore .mcp.json dashboard/__init__.py dashboard/datasource.py dashboard/fixtures/er_state.json dashboard/orchestrator_client.py dashboard/pika_jobs.py dashboard/server.py dashboard/sim.py dashboard/static/app.js dashboard/static/floor.js dashboard/static/index.html dashboard/static/library.html dashboard/static/library.js dashboard/static/login.html dashboard/static/replay.html dashboard/static/replay.js dashboard/static/style.css er_twin/__init__.py er_twin/active_events.py er_twin/addresses.py er_twin/agents/__init__.py er_twin/agents/admissions.py er_twin/agents/bed.py er_twin/agents/doctor.py er_twin/agents/equipment.py er_twin/agents/nurse.py er_twin/agents/orchestrator.py er_twin/agents/patient.py er_twin/agents/stub.py er_twin/agents/triage.py er_twin/config.py er_twin/connect_orchestrator.py er_twin/display.py er_twin/ehr.py er_twin/events/__init__.py er_twin/events/base.py er_twin/events/discharge_flow.py er_twin/events/discharge.py er_twin/events/helpers.py er_twin/events/intake_flow.py er_twin/events/intake.py er_twin/events/oxygen.py er_twin/events/ping.py er_twin/events/registry.py er_twin/events/resolve.py er_twin/events/summary.py er_twin/main.py er_twin/memory.py er_twin/oxygen_coord.py er_twin/oxygen_flow.py er_twin/protocols.py er_twin/replay.py er_twin/status_summary.py er_twin/storage.py fixtures/ehr_master.json pyproject.toml README.md scripts/build_ehr.py scripts/build_pika_prompt.py scripts/capture_replay_frames.py scripts/redis_smoke.py scripts/replay_meta.py scripts/run_pika_identity_check.ps1 scripts/run_pika_keyframes.ps1 scripts/run_pika_replay.ps1 scripts/seed_redis.py skills-lock.json spikes/mailbox_inside_bureau_spike.py spikes/oxygen_async_flow_spike.py tests/conftest.py tests/test_dashboard.py tests/test_domain_invariants.py tests/test_ehr.py tests/test_event_discharge.py tests/test_event_intake.py tests/test_event_oxygen.py tests/test_event_registry.py tests/test_event_summary.py tests/test_integration_wiring.py tests/test_memory.py tests/test_orchestrator_skeleton.py tests/test_patient_pool.py tests/test_replay.py tests/test_storage.py uv.lock ``` ### Dependencies - pyproject.toml: authlib, fastapi, itsdangerous, openai@>=2.43.0, pydantic-settings, python-multipart, redis, redis-agent-memory, uagents, uagents-core, uvicorn ### Recent commits (newest first) - UI dashboard changes - edit gitignore - add to gitignore - clean up - Fix discharge not freeing nurse: seed p2.care_team + release by assignments - Merge PR #7: dashboard 3D map polish (bird logo + expanded floor) into main - Merge branch 'main' into feat/dashboard-3d-map-polish - Merge remote-tracking branch 'origin/main' - Polish dashboard exterior shell - Dashboard media library + replay polish, Pika keyframes job wiring - started patinet intake outake logic - Expand dashboard floor layout - name switch - Merge pull request #6 from RyanDang363/feat/dashboard-3d-map-polish - Refine dashboard 3D hospital shell - Docs: architecture, Fetch deliverables, agent profile README + handle/drift sync - Merge main into dashboard: port feat 3D map onto shared floor.js - Orchestrator @ERTwin handle/profile + AGENT.md connection guide - Polish dashboard 3D map interactions and styling - Live ASI:One/Agentverse integration + demo assembly ## Key source files (fetched from GitHub, selected and truncated for size) ### pyproject.toml ``` [project] name = "er-twin" version = "0.1.0" description = "ER Room Digital Twin — Fetch.ai uAgents + Bureau + ASI:One" requires-python = ">=3.11,<3.13" dependencies = [ "uagents", "uagents-core", "pydantic-settings", "redis", "redis-agent-memory", "fastapi", "uvicorn", "python-multipart", "itsdangerous", "authlib", "openai>=2.43.0", ] [dependency-groups] dev = [ "pytest>=8.0", "ruff>=0.4", "httpx", # Headless frame capture for the incident replay (scripts/capture_replay_frames.py, Phase 3). # After `uv sync`, run `uv run playwright install chromium` once to fetch the browser. "playwright>=1.40", ] [tool.uv] package = false [tool.ruff] line-length = 100 # Vendored third-party clone (Fetch.ai Innovation Lab examples) — not our code to lint. extend-exclude = ["fetch-ai-documentation/innovation-lab-examples"] [tool.pytest.ini_options] pythonpath = ["."] testpaths = ["tests"] ``` ### er_twin/main.py ```python """Single entry point for the ER Twin (LLD §5). Builds ONE `Bureau` holding the public OrchestratorAgent (mailbox + Chat Protocol, reachable from ASI:One) and every private entity agent — the PatientAgent pool plus the bed / nurse / doctor / equipment / admissions / triage agents — then `bureau.run()`. One process, one event loop (ORCH-SYS-001). The Bureau starts the Orchestrator's mailbox client; in-process messaging is the spike-proven seam (spikes/mailbox_inside_bureau_spike.py). All agents share one `InMemoryStore` (the demo-safe default behind `StorageInterface`). State is seeded deterministically before the Bureau runs (no async startup race): `seed_state` lays the clean inventory, then `seed_baseline` adds the mid-shift demo scenario (decision Gap 5 / R2-B) so the oxygen and summary commands are demoable in any order. Run: `USE_MOCK=true uv run python -m er_twin.main` Expected on startup: the Orchestrator logs its `agent1q…` address plus an Agentverse inspector URL. A line like "Agent mailbox not found: create one using the agent inspector" is EXPECTED until the one-time inspector connect — it is not a failure. """ from uagents import Bureau from er_twin.addresses import ORCHESTRATOR_ADDRESS, STUB_ADDRESS from er_twin.agents import admissions, bed, doctor, equipment, nurse, patient, triage from er_twin.agents import orchestrator as orch from er_twin.agents.orchestrator import orchestrator from er_twin.agents.stub import stub from er_twin.config import settings from er_twin.memory import make_memory from er_twin.storage import StorageInterface, make_store # Modules that seed a slice of the shared store (have init_state). Order is irrelevant. _ENTITY_MODULES = (patient, bed, nurse, doctor, equipment) # Modules contributing agents but no seeded inventory (handlers call domain fns on demand). _AGENT_ONLY_MODULES = (admissions, triage) def seed_state(store: StorageInterface) -> None: """Seed every entity's clean initial inventory into the shared store (deterministic, pre-run).""" for module in _ENTITY_MODULES: module.init_state(store) def _inventory_counts(store: StorageInterface) -> dict[str, int]: """Read back the live index sets so the boot banner reports what actually landed in the store (not the module constants), exposing a partial/failed seed instead of hiding it.""" return {e: len(store.list_ids(e)) for e in ("patient", "bed", "nurse", "doctor", "equipment")} def ensure_seeded(store: StorageInterface) -> dict[str, int]: """Seed the store, verify the core inventory landed, and re-seed once if it did not. With the demo-default `InMemoryStore` the seed is always durable. With a persistent `RedisStore` it must not be assumed: a prior/parallel run or a transient backend error can leave the keyspace without beds/nurses, after which the Orchestrator answers every intake with `no_bed_available` while the boot log still looks healthy. So we seed, read the indexes back, and retry once if beds or nurses are missing — turning a silent downstream failure into a loud, self-healing startup step. Returns the verified counts for the boot banner. """ seed_state(store) seed_baseline(store) counts = _inventory_counts(store) if counts["bed"] == 0 or counts["nurse"] == 0: seed_state(store) seed_baseline(store) counts = _inventory_counts(store) return counts def seed_baseline(store: StorageInterface) -> None: """Layer the mid-shift demo scenario on top of the clean seed (decision Gap 5 / R2-B/C). p1 waiting-room patient, p2 on bed-3 with oxygen unit o2_1; nurse1 busy with p2 (so the oxygen dispatch deterministically picks nurse2); doc2 carrying p2. Patient counter advanced to 2. """ store.set("er:counter:patient", {"value": 2}) store.set("er:patient:p1", { "id": "p1", "mrn": "MRN-0001", "name": "Sam Rivera", "chief_complaint": "observation after minor fall", "acuity": 4, "specialty": "general", "status": "in_triage", "vitals": {"heart_rate": 84, "blood_pressure": "128/78", "resp_rate": 16, "spo2": 98, "temperature_f": 98.4, "pain_score": 3}, "assigned_bed": None, "care_team": [], }) store.set("er:patient:p2", { "id": "p2", "mrn": "MRN-0002", "name": "Avery Chen", "chief_complaint": "shortness of breath", "acuity": 3, "specialty": "general", "status": "in_treatment", "vitals": {"heart_rate": 104, "blood_pressure": "136/84", "resp_rate": 24, "spo2": 92, "temperature_f": 99.1, "pain_score": 4}, "assigned_bed": "bed3", "care_team": ["nurse1", "doc2"], }) store.update("er:bed:bed3", {"occupied_by": "p2", "status": "occupied", "equipment": ["o2_1"]}) store.update("er:equipment:o2_1", {"supply_level": 55, "in_use_by": "p2", "location": "bed-3"}) store.update("er:nurse:nurse1", {"available": False, "location": "bed-3", "assignments": ["p2"]}) store.update("er:doctor:doc2", {"load": 1, "assignments": ["p2"]}) def build_bureau(store: StorageInterface) -> Bureau: bureau = Bureau() bureau.add(orchestrator) bureau.add(stub) for module in (*_ENTITY_MODULES, *_AGENT_ONLY_MODULES): for agent in module.build_agents(store): bureau.add(agent) return bureau def main() -> None: # Backend selection lives in the factories (LLD §4): USE_MOCK=true ⇒ InMemoryStore + NoopMemory # (zero-dependency demo); USE_MOCK=false with REDIS_URL / AGENT_MEMORY_* set ⇒ live Redis + Iris. store = make_store() memory = make_memory() counts = ensure_seeded(store) # seed + verify the inventory actually landed (self-healing) orch.set_store(store) # the Orchestrator coordinates intake over this same store orch.set_memory(memory) # ...and records/recalls ER events through this memory backend print(f"USE_MOCK = {settings.use_mock}") print(f"store = {ty [truncated — 1146 more characters] ``` ### dashboard/server.py ```python """FastAPI server for the read-only admin dashboard. @spec DASH-API-001, DASH-API-002, DASH-API-003, DASH-API-004, DASH-ERR-001, DASH-IN-002 @spec DASH-AUTH-001, DASH-AUTH-002, DASH-AUTH-003, DASH-AUTH-004, DASH-AUTH-005, DASH-AUTH-006 @spec DASH-AUTH-007, DASH-AUTH-008 Run: uvicorn dashboard.server:app --port 8050 Auth note: session-cookie login via Google OAuth (any account — no allowlist) or a hardcoded username/password fallback. A demo access gate, NOT real HIPAA compliance (the project uses synthetic data; production compliance is out of scope). """ import json import re from datetime import datetime from pathlib import Path from authlib.integrations.starlette_client import OAuth, OAuthError from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware from er_twin.config import settings from . import pika_jobs import dashboard.datasource as datasource from .datasource import active_events_list, current_events, derive_summary, live_snapshot _STATIC = Path(__file__).parent / "static" # Incident replay artifacts written by the Orchestrator (er_twin.replay, LLD §9.1). Resolved relative # to the repo root so the dashboard reads the same out/replay/ the agents write, regardless of CWD. _REPLAY_DIR = Path(__file__).parent.parent / "out" / "replay" # Incident ids are `{incident_type}-{n:04d}` — a strict allowlist also blocks path traversal. _INCIDENT_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") def _replay_file(incident_id: str) -> Path | None: """The `out/replay/{incident_id}.json` path if it exists and the id is safe, else None.""" if not _INCIDENT_ID_RE.fullmatch(incident_id): return None path = _REPLAY_DIR / f"{incident_id}.json" return path if path.is_file() else None app = FastAPI(title="ER Twin — Admin Dashboard") app.add_middleware(SessionMiddleware, secret_key=settings.dashboard_secret_key) app.mount("/static", StaticFiles(directory=_STATIC), name="static") # Google OAuth is registered only when credentials are configured (graceful degradation). oauth = OAuth() GOOGLE_ENABLED = bool(settings.google_client_id and settings.google_client_secret) if GOOGLE_ENABLED: oauth.register( name="google", client_id=settings.google_client_id, client_secret=settings.google_client_secret, server_metadata_url="https://accounts.google.com/.well-known/openid-configuration", client_kwargs={"scope": "openid email profile"}, ) _last_good: dict | None = None # --- Auth --------------------------------------------------------------------- def current_user(request: Request) -> str | None: return request.session.get("user") def require_api(request: Request) -> str: """Dependency: protected API routes return 401 when unauthenticated. @spec DASH-AUTH-004""" user = current_user(request) if not user: raise HTTPException(status_code=401, detail="authentication required") return user @app.get("/login") def login_page() -> FileResponse: return FileResponse(_STATIC / "login.html") @app.post("/login") async def login(request: Request) -> RedirectResponse: """Validate hardcoded credentials and establish a session. @spec DASH-AUTH-001, DASH-AUTH-002""" form = await request.form() username = str(form.get("username", "")) password = str(form.get("password", "")) if username == settings.dashboard_username and password == settings.dashboard_password: request.session["user"] = username return RedirectResponse("/", status_code=303) return RedirectResponse("/login?error=1", status_code=303) @app.get("/auth/config") def auth_config() -> JSONResponse: """Lets the login page show the Google button only when configured. @spec DASH-AUTH-007""" return JSONResponse({"google_enabled": GOOGLE_ENABLED}) @app.get("/auth/google") async def auth_google(request: Request): """Begin the Google OAuth flow. @spec DASH-AUTH-007""" if not GOOGLE_ENABLED: return RedirectResponse("/login?error=oauth_unconfigured", status_code=303) redirect_uri = request.url_for("auth_callback") return await oauth.google.authorize_redirect(request, redirect_uri) @app.get("/auth/callback") async def auth_callback(request: Request): """Complete OAuth: any authenticated Google account is allowed in. @spec DASH-AUTH-007, DASH-AUTH-008""" if not GOOGLE_ENABLED: return RedirectResponse("/login?error=oauth_unconfigured", status_code=303) try: token = await oauth.google.authorize_access_token(request) except OAuthError: return RedirectResponse("/login?error=1", status_code=303) userinfo = token.get("userinfo") or {} email = userinfo.get("email") if not email: return RedirectResponse("/login?error=1", status_code=303) request.session["user"] = email # no allowlist — any Google account is accepted return RedirectResponse("/", status_code=303) @app.get("/logout") def logout(request: Request) -> RedirectResponse: """Clear the session. @spec DASH-AUTH-005""" request.session.clear() return RedirectResponse("/login", status_code=303) # --- Pages & API (protected) -------------------------------------------------- @app.get("/") def index(request: Request): """Serve the dashboard, or redirect to login when unauthenticated. @spec DASH-AUTH-003""" if not current_user(request): return RedirectResponse("/login", status_code=303) return FileResponse(_STATIC / "index.html") @app.get("/api/state") def api_state(user: str = Depends(require_api)) -> JSONResponse: """Full read-only snapshot + derived KPIs. Falls back to last-good if the source is down.""" global _last_good try: snap = live_snapshot() except Exception: # noqa: BLE001 — source unavailable must never crash the server [truncated — 9981 more characters] ``` ### er_twin/__init__.py ```python """ER Room Digital Twin — Fetch.ai uAgents multi-agent system.""" ``` ### dashboard/__init__.py ```python """Read-only admin dashboard for the ER twin (see docs/llds/dashboard.lld.md).""" ``` ### er_twin/display.py ```python """Presentation-only id → friendly name map for chat and dashboard.""" DISPLAY_NAMES: dict[str, str] = { "nurse1": "Nurse Maya", "nurse2": "Nurse Chen", "doc1": "Dr. Smith", "doc2": "Dr. Patel", "bed1": "bed-1", "bed2": "bed-2", "bed3": "bed-3", "bed4": "bed-4", "o2_1": "oxygen unit o2-1", "o2_2": "replacement unit o2-2", } def display(entity_id: str | None) -> str: return DISPLAY_NAMES.get(entity_id, entity_id) if entity_id else "" ``` ### er_twin/oxygen_flow.py ```python """Oxygen flow correlation state (LLD §6).""" from dataclasses import dataclass, field @dataclass class OxygenFlow: """Per-flow context for the multi-hop oxygen event, keyed by `flow_id`.""" flow_id: str bed_id: str alert_equipment_id: str session_id: str | None = None chat_sender: str | None = None replacement_id: str | None = None nurse_id: str | None = None status: str = "started" lines: list[dict] = field(default_factory=list) ``` ### dashboard/orchestrator_client.py ```python """Seam for future command input (read-only baseline — NOT wired yet). @spec DASH-IN-001 (deferred) When command input is enabled later, implement `send_command` as a uAgents client that sends a `ChatMessage` to the Orchestrator's address (pattern: fetch-ai-documentation/uagents-chat-protocol.md, client-agent example). The HTTP route and frontend slot already exist behind `dashboard_allow_input`. """ def send_command(phrase: str) -> bool: """Forward a trigger phrase to the OrchestratorAgent. Not implemented in the read-only baseline.""" raise NotImplementedError( "Command input is deferred (DASH-IN-001). Enable dashboard_allow_input and implement " "the uAgents ChatMessage client to activate." ) ``` ### er_twin/addresses.py ```python """Deterministic agent addresses derived from the base seed (LLD section 5). Computed once at import time and used as constants — no runtime Almanac discovery. Each agent's seed is `{AGENT_SEED}-{role}`; entity pools append an index, e.g. `bed-1`. """ from uagents.crypto import Identity from er_twin.config import settings def seed_for(role: str) -> str: return f"{settings.agent_seed}-{role}" def address_for(role: str) -> str: return Identity.from_seed(seed_for(role), 0).address # Singleton agents referenced across the system. ORCHESTRATOR_ADDRESS = address_for("orchestrator") STUB_ADDRESS = address_for("stub") ADMISSIONS_ADDRESS = address_for("admissions") TRIAGE_ADDRESS = address_for("triage") def pool_address(role: str, index: int) -> str: """Address for one member of an entity pool, e.g. pool_address('bed', 1).""" return address_for(f"{role}-{index}") ``` ### er_twin/oxygen_coord.py ```python """Low-oxygen coordination pure functions (OXY-*).""" from __future__ import annotations import re from er_twin.agents import equipment, nurse from er_twin.display import display from er_twin.storage import StorageInterface def should_start_o2_dispatch(in_flight: dict[str, str], equipment_id: str) -> bool: return equipment_id not in in_flight def apply_oxygen_swap( store: StorageInterface, depleted_id: str, replacement_id: str, bed_id: str, nurse_id: str ) -> str | None: occupant = equipment.swap_oxygen_unit(store, depleted_id, replacement_id, bed_id) nurse.dispatch_nurse(store, nurse_id, bed_id) return occupant def format_oxygen_confirmation(bed_id: str, replacement_id: str, nurse_id: str) -> str: return ( f"Low O2 on {display(bed_id)} resolved: dispatched {display(nurse_id)} with " f"{display(replacement_id)}; patient SpO2 restored to 96%." ) def bed_from_text(text: str) -> str: match = re.search(r"bed\s*(\d+)", text.lower()) return f"bed{match.group(1)}" if match else "bed3" def cleanup_oxygen( flow_id: str, flows: dict, in_flight: dict, senders, ) -> None: flow = flows.pop(flow_id, None) if flow is not None: in_flight.pop(flow.alert_equipment_id, None) if flow.session_id: senders.forget(flow.session_id) ``` [65 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]