# 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 - Built for Berkeley AI Hackathon 2026 **Hackathon Technical Specification — Developer Reference** Fetch.ai uAgents + Bureau · ASI:One · in-memory/Redis state · Pika MCP replay (via Claude Code) > Target: 24-hour build · Python 3.11+ · Local Bureau, one Agentverse mailbox **One-liner:** *Fetch.ai coordinates the ER response; ASI:One exposes the public chat interface; StorageInterface/Redis records the event trace; Claude Code CLI invokes Pika MCP to turn that trace into replay media.* > **Connecting as a teammate or judge?** See [`AGENT.md`](AGENT.md) for how to reach the canonical ER Twin Orchestrator on ASI:One / Agentverse — and why you should **not** re-register your own copy. --- ## Feasibility Verdict **✅ Feasible — with one critical architecture choice** **Single-process Bureau (verified P1 default).** Everything runs in **one Python process, one `Bureau`**: the public `OrchestratorAgent` (`mailbox=True`, `publish_agent_details=True`, Chat Protocol) is added to the same Bureau as the private ER entity agents, and they communicate via in-process uAgent messaging. ASI:One reaches only the Orchestrator. > **Verified by spike.** Official Fetch docs do not prominently showcase `mailbox=True` agents inside a Bureau, but our local spike on `uagents==0.25.2` (`spikes/mailbox_inside_bureau_spike.py`) proves that **`Bureau.run_async` starts a member agent's mailbox client** and that **in-process Orchestrator ↔ entity messaging works** in this project environment. If Agentverse/ASI:One smoke testing fails, we fall back to the documented **two-process pattern** (standalone Orchestrator process + separate Bureau process) — see *Architecture Alternatives and Fallbacks*. **Why one process:** the riskiest seam (does the mailbox client start, and does internal messaging work?) is now proven, in-process, with one event loop and one command to debug. The two-process split would trade that proven seam for an *untested* cross-process endpoint hop — worse for a 24-hour build. **Conclusion:** all agents are real uAgents inside a single local Bureau; only the `OrchestratorAgent` gets an Agentverse mailbox + Chat Protocol + ASI:One so you can talk to the system from outside. **Demo priority:** Talking to the ASI:One orchestrator and watching it trigger ER events. Everything else is built in service of that one interaction loop. --- ## Overview Emergency rooms operate in controlled chaos — every room, patient, nurse, doctor, and piece of equipment is a moving variable. This project builds an autonomous digital twin of a hospital emergency room where every physical entity is modeled as a uAgent, agents coordinate in real time via in-process Bureau messaging, and a single `OrchestratorAgent` — reachable through ASI:One — responds to critical events autonomously. > 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 (creates a local .venv) uv sync # 4. Run the system (mock mode — no ASI:One key needed): ONE process, ONE Bureau USE_MOCK=true uv run python -m er_twin.main # 5. Run the tests uv run pytest ``` **Mock mode:** `USE_MOCK=true` skips the *external* services — deterministic keyword intent lookup instead of the ASI:One LLM, `InMemoryStore` instead of Redis, `NoopMemory` instead of Iris — but still runs the **real** agent coordination in-process over deterministically seeded state. So replies are state-derived (not canned) and reproducible with **no API keys** (see the `USE_MOCK` contract in [docs/TEAM.md](docs/TEAM.md)). Set `USE_MOCK=false` (+ keys) for the live ASI:One LLM, Redis, and Iris. **How it works:** [ARCHITECTURE.md](ARCHITECTURE.md) maps the implemented system — subsystems, the three event flows, and the state / memory / replay layers. **Who builds what:** see [docs/TEAM.md](docs/TEAM.md) for the ownership map and git workflow, and [STATUS.md](STATUS.md) for live progress. --- ## Core Problem Emergency rooms suffer from cascading inefficiencies caused by static, reactive coordination: - **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 This is the **Fetch.ai-native path**: build directly on uAgents, run everything in **one process / one `Bureau`**, with the public `OrchestratorAgent` (mailbox) added to the same Bureau as the private ER entity agents. ASI:One discovers and chats with **only** the Orchestrator — it is the single external surface. State lives behind a `StorageInterface` (InMemoryStore first, Redis later). After an event runs, the system exports an incident trace that the **Claude Code CLI → Pika MCP** turn into replay media — an automated post-processing step, *not* part of the Fetch runtime. **Data-driven replay (LLD §9.1).** Beyond the narrative brief, every milestone also captures a full-state ER snapshot (with a real `ts`) into `out/replay/{incident}.json`. A `/replay/{incident}` page replays it on the **same SVG floor map as the live dashboard** (shared `floor.js`), tweening tokens between snapshots in real time. The milestone keyframes are rasterized to PNGs (`scripts/capture_replay_frames.py`, Playwright) and fed to Pika `generate_keyframes_video` (`scripts/run_pika_keyframes.ps1`) for a time-compressed start→end clip; the returned `video_url` is written back into the incident file and embedded in a gated `/library` page that lists every incident this session. So Pika reconstructs **ground-truth state**, not a hallucination — and if Pika is skipped, `/replay/{incident}` still plays the reconstruction. `er:events` / `REPLAY-LOG-002` are unchanged (`ts` lives only on the snapshot records). **Single-process runtime (P1 default):** - **One entry point — `er_twin/main.py`.** Builds a single `Bureau`, adds the `OrchestratorAgent` (`mailbox=True`, `publish_agent_details=True`, `Protocol(spec=chat_protocol_spec)`) **and** all private entity agents (Admissions, Triage, Patient(s), Bed(s), Nurse(s), Doctor(s), Equipment(s)), then `bureau.run()`. - **Orchestrator** is the only public surface (Agentverse mailbox + ASI:One). It handles the 3 NL demo triggers, dispatches **in-process** uAgent messages to the private agents, and writes the event log + `out/incident_replay_brief.json`. - **Entity agents** have **no mailbox** and **no Agentverse profiles** — private by design. > **Not** the "other framework → uAgent Adapter → Agentverse" path, and **not** hosting every > entity as a public Agentverse agent. Pika MCP is never called by uAgents directly — only by the > Claude Code CLI. The **two-process** split (standalone Orchestrator + separate Bureau) is the > documented fallback if ASI:One smoke testing fails — see *Architecture Alternatives and Fallbacks*. **Implementation notes (carry into P1 code):** - **Async, not request/response.** uAgents messaging is fire-and-forget: a chat handler `ctx.send`s and returns; the reply arrives later in a *separate* `@on_message` handler. The Orchestrator must store `{session/request id → user sender address}` and send the final `ChatMessage` from the response handler. (This matters more than the process-count decision.) - **Construct, don't import, the chat protocol:** `chat = Protocol(spec=chat_protocol_spec)` then `orchestrator.include(chat)`. There is no importable `chat_proto`. - **Pin Python `>=3.11,<3.13 [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.]