# Project export: RELAY – Voice, Search, Stream: Save patients from rigs to ER 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: TreeHacks 2026 - Tagline: Lost or delayed info kills. RELAY finds and streams the patient’s info to the ER in real time: faster, more accurate handover, while medics keep their hands on the patient, not the keyboard or phone. - Devpost: https://devpost.com/software/aria-voice-search-stream-save-patients-from-rigs-to-er - GitHub: https://github.com/vinamra57/relay - Video: https://www.youtube.com/embed/5CBSHbyvYiM?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 5 GitHub contributor(s) — Julbarth (16 commits), Vinamra Agarwal (12 commits), Claude (6 commits), Aditi Oak (4 commits), Cursor (2 commits) ## Devpost submission (written by the team) ### Inspiration The biggest problem is lack of information from delays in accessing patients histories and present information gets lost. 1/3 patients are at risk of treatment mistake in ERs and 40% are critical. And this starts in the ambulance. Paramedics see and do everything; by the time the patient reaches the ER, a lot of that context is lost. Studies put missing information in ambulance-to-ER handovers at around 30%. That means wrong or delayed treatments, duplicated work, and risk that could be avoided. There needs to be a better way to find and request information autonomously, accuratly document and stream it right to the hospital so medics can keep their hands and attention on the patient. This is why we built RELAY. ### What it does Relay is an AI-powered emergency response system that turns paramedic voice and background data into a single, real-time handover from ambulance to ER. It addresses the past, present, and future of the patient in one pipeline: so the ER sees the full picture even before the patient arrives. Devices: laptop with microphone in the ambulance, and a browser-based dashboard in the hospital. Past: patient history (hands-off search) Today, finding past context (e.g., allergies, conditions, medications, prior care, etc.,) often means nurses or medics calling GPs and chasing records by phone, or typing notes into official documentation needed already during the ambulance drive. We outsource that to agents so medics stay with the patient. Once we have core information (full name, address, age, gender) from the paramedic’s voice, we trigger in parallel: GP outreach: We resolve the GP/practice (e.g. lookup number by name and location), place a call (ElevenLabs + Twilio), and request data. No medic on the phone. Medical history: We query FHIR R4 (e.g. Synthea for dev; in production, HIE or Particle Health–style APIs) for corresponding patient records. So the team knows before arrival what drugs are safe and what to avoid. Relative data: We stream the relative data to the hospital much faster such that they can be connected sooner and provide information as they are one of the most critical sources. We decided not to use artificial agents here as this requires sensitive handling. All of that is merged into one patient story. Hands-off: documentation, requesting, and consolidation are done by the system and agents, not by someone on a keyboard or phone. Present: what’s happening now (voice → structure, no scribbling) Voice is the only input. A wearable mic streams the audio records of the treating medic. NEMSIS-compliant information is extracted (e.g., demographics, vitals, procedures, medications, impressions) so the ePCR fills itself as the paramedic speaks. No writing, no tapping; such that the medics can keep their hands on the patient through the full ride. Future: what the hospital needs to be ready (stream + speed) Right now the hospital is often called 1–2 minutes out if the medics have the time and gets the full ePCR at the end. We flip that: we stream information to the hospital en route. The hospital sees a live dashboard over WebSockets—active cases, NEMSIS updates, GP and medical-DB responses—and interfaces for urgency-based summaries for the handover. So the ER can prepare earlier: right team, right room, right drugs. Speed comes from (1) parallel GP + DB fetch, (2) continuous stream instead of one report at the end, and (3) less manual documentation so care isn’t delayed. How it fits together Handover gap: By recording and structuring everything from voice and pulling history in parallel, we aim for a complete handover and lower the risk of missing information. Delay in care: Because there’s only so much one human can do on a ride, we outsource documentation and information-gathering to the system and agents; more information is available faster and more accurately. Delay in hospital prep: We use a stream of information so the hospital can prep quicker. Human labor and error: We reduce calling around, handwritten notes, and late data entry by using secure electronic flows and agents to call, orchestrate, search, and consolidate. ### How we built it We built an end‑to‑end ambulance‑to‑hospital pipeline that streams hands‑free voice capture, extracts structured clinical data, and pushes a live hospital prep dashboard. Core stack Frontend: HTML/CSS/JS dashboards for paramedic + hospital views, real‑time updates via WebSockets. Backend: FastAPI + WebSockets, event bus for fan‑out, SQLite for persistence. AI: LLM‑based NEMSIS ePCR extraction and hospital summaries (Anthropic/OpenAI/Modal‑ready via a model‑agnostic client, Perplexity for targeted search). Voice: ElevenLabs streaming transcription. Data: NEMSIS‑compliant JSON, transcript segments with timestamps, and vitals playback for realism. Pipeline Wearable mic streams audio to the backend, producing continuous, timestamped segments. ElevenLabs transcribes in real time, merging partials into committed transcript updates. Each committed chunk updates a NEMSIS‑compliant ePCR record (patient info, vitals, impressions, procedures, meds). Structured data is persisted and pushed over WebSockets so the hospital view stays 1:1 with the ambulance stream. Once core identity is complete (name, age/DOB, address, gender), the system triggers parallel enrichment: medical history lookup and GP outreach. The hospital dashboard renders an inbound critical brief, contraindication warnings, prep actions, live vitals trends, and a full NEMSIS ePCR tab for deeper drill‑down. ### Challenges we ran into Real‑time coordination: Keeping transcript, NEMSIS extraction, and hospital UI in sync with low latency. Data completeness vs. speed: Balancing early summaries with partial information without misleading clinicians. UI clarity under pressure: Designing a dashboard that is dense but still readable in seconds. Integration complexity: Orchestrating voice, LLM extraction, and streaming updates reliably across the pipeline. ### Accomplishments we're proud of A hands‑free, end‑to‑end pipeline from ambulance voice to a live hospital dashboard. NEMSIS‑compliant structured extraction updated in real time. A hospital UI that highlights critical prep actions and contraindications immediately. A system that streams the ePCR 1:1 to the ER before arrival. ### What we learned Software engineering is dead. ### What's next for Relay – Voice, Search, Stream: Save patients from rigs to ER Production‑grade integrations with HIEs and hospital EHR systems. Clinical decision support tuned to EMS protocols and ER workflows. Voice‑first clinician queries for fast retrieval of critical facts. Hospital prep automation (stroke, STEMI, trauma activations) with tighter rules and audit trails. Security + compliance hardening for real clinical deployment. A custom wearable, 3D printed including the microphone. Adding AI-based predictive and diagnostic tools for early warning systems. ## README (from the GitHub repository) # Relay AI Emergency Response System — Automated ePCR for Paramedics. Relay captures real-time audio from paramedics, transcribes it, and extracts structured NEMSIS v3.5 medical data using LLMs. It enriches patient records by querying FHIR health information exchanges and contacting GPs, then streams everything to a hospital dashboard for live monitoring. ## Architecture ``` Paramedic (audio) ──WebSocket──▶ Transcription ──▶ NEMSIS Extraction (Claude) │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ FHIR Lookup GP Caller Medical DB │ (Twilio + │ │ ElevenLabs) │ ▼ ▼ ▼ Clinical Insights + Alerts │ Event Bus (Pub/Sub) │ ▼ Hospital Dashboard (WebSocket) ``` ## Quick Start ```bash # Install dependencies pip install -r requirements.txt # Configure environment cp .env.example .env # Edit .env with your API keys (at minimum ANTHROPIC_API_KEY) # Run uvicorn app.main:app --reload ``` - Paramedic UI: http://localhost:8000/ - Hospital Dashboard: http://localhost:8000/hospital ## Key Environment Variables | Variable | Description | Default | |----------|-------------|---------| | `ANTHROPIC_API_KEY` | Claude API key for NEMSIS extraction | required | | `ELEVENLABS_API_KEY` | Voice agent for GP calls | optional | | `TWILIO_ACCOUNT_SID` / `TWILIO_AUTH_TOKEN` | Outbound calling | optional | | `LLM_PROVIDER` | `auto`, `anthropic`, or `openai` | `auto` | | `DATABASE_URL` | PostgreSQL connection string for production | SQLite default | | `SEED_DEMO_CASES` | Populate demo cases on startup | `true` | | `GP_CALLS_ENABLED` | Enable outbound GP calls | `false` | | `DUMMY_MODE` | Mock data generation for development | `false` | | `GCP_PROJECT_ID` / `GCP_PUBSUB_TOPIC` | Multi-instance event streaming | optional | See `.env.example` for the full list. ## Project Structure ``` app/ main.py # FastAPI app entrypoint config.py # Environment configuration database.py # SQLite / PostgreSQL abstraction routers/ stream.py # WebSocket audio streaming + transcription cases.py # REST API for case CRUD hospital.py # Hospital dashboard WebSocket gp_call.py # GP call workflow endpoints services/ nemsis_extractor.py # Transcript → NEMSIS v3.5 via LLM llm.py # LLM abstraction (Anthropic / OpenAI) transcription.py # Real-time audio transcription voice_agent.py # ElevenLabs + Twilio voice calling fhir_client.py # FHIR R4 patient record queries gp_caller.py # GP contact workflow gp_documents.py # PDF/OCR extraction of GP records gp_lookup.py # GP practice phone lookup clinical_insights.py # Clinical alerts and recommendations medical_db.py # Medical history queries event_bus.py # Pub/Sub or in-memory event streaming models/ nemsis.py # NEMSIS v3.5 data structures case.py # Case API models clinical.py # Clinical decision support models transcript.py # Transcript segment models static/ # Frontend UIs (paramedic + hospital) docs/ # Deployment guides tests/ # pytest test suite ``` ## Testing ```bash pytest ``` Requires 60% minimum code coverage. Config in `pyproject.toml`. ## Deployment See [`docs/gcp-setup.md`](docs/gcp-setup.md) for Cloud Run deployment with Cloud SQL and Pub/Sub. ## Tech Stack - **Backend**: FastAPI, WebSockets, Pydantic - **AI**: Anthropic Claude, OpenAI (fallback), ElevenLabs - **Healthcare**: FHIR R4, NEMSIS v3.5 - **Database**: SQLite (dev) / PostgreSQL (prod) - **Telephony**: Twilio - **Cloud**: GCP Cloud Run, Pub/Sub, Cloud SQL - **Python**: 3.11+ ## Detected evidence (automated analysis) Indexed codebase: 58 recognized source files, 433 KB. - Anthropic (technology) — detected in the code - 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 - AI coding agent: Claude Code — evidence: commit authorship or trailers - AI coding agent: Cursor — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (63 of 63) ``` .env.example .gitignore app/__init__.py app/config.py app/database.py app/main.py app/models/__init__.py app/models/case.py app/models/clinical.py app/models/medical_history.py app/models/nemsis.py app/models/summary.py app/models/transcript.py app/routers/__init__.py app/routers/cases.py app/routers/gp_call.py app/routers/hospital.py app/routers/stream.py app/services/__init__.py app/services/clinical_insights.py app/services/core_info_checker.py app/services/event_bus.py app/services/fhir_client.py app/services/gp_caller.py app/services/gp_documents.py app/services/gp_lookup.py app/services/llm.py app/services/medical_db.py app/services/nemsis_extractor.py app/services/qa.py app/services/summary.py app/services/transcription.py app/services/vitals_dataset.py app/services/voice_agent.py data/vitals/bidmc_01_Numerics.csv data/vitals/README.md docs/gcp-setup.md pyproject.toml README.md requirements.txt static/app_enhanced.js static/app.js static/hospital.css static/hospital.html static/hospital.js static/index_enhanced.html static/index.html static/style_enhanced.css static/style.css tests/__init__.py tests/conftest.py tests/test_api.py tests/test_database.py tests/test_event_bus.py tests/test_fhir_client.py tests/test_gp_caller.py tests/test_gp_lookup.py tests/test_medical_db.py tests/test_models.py tests/test_services.py tests/test_summary.py tests/test_voice_agent.py tests/test_websocket.py ``` ### Dependencies - requirements.txt: aiosqlite@==0.20.0, anthropic@>=0.40.0, asyncpg@==0.30.0, fastapi@==0.115.6, google-cloud-pubsub@==2.31.0, httpx@>=0.28.0, openai@==1.59.3, pdf2image@==1.17.0, pydantic@>=2.10.4,<3, pypdf@>=4.2.0, pyright@==1.1.408, pytesseract@==0.3.10, pytest@==9.0.2, pytest-asyncio@==1.3.0, pytest-cov@==7.0.0, python-dotenv@==1.0.1, ruff@==0.15.1, twilio@==9.10.1, uvicorn[standard]@==0.34.0, websockets@==14.1 ### Recent commits (newest first) - Add comprehensive project README - Merge pull request #2 from vinamra57/vin-jb - Add GP document OCR and expose PDF for hospital - Add hospital WS heartbeat to prevent disconnects - Fix pending task warning in extraction loop - Remove GP records source row - Add GP status indicators and disable calls by default - Gate audio streaming on voice activity - Disable dummy transcription and require live voice - Trigger GP status when contact detected - Improve stream latency and demo FHIR history - Add gp_documents, config and static updates - use sonnet instead of haiku - Fix GP contact resolution: wait for full phone before triggering call - Rename app from Aria Health to Relay - done for the night - Pass patient details and chief complaint to GP calling agent - Fix hospital auto-select render - Revert "Merge main into jb branch" - Tune vitals update interval ## Key source files (fetched from GitHub, selected and truncated for size) ### docs/gcp-setup.md ```markdown # GCP Setup (Cloud SQL + Pub/Sub + Cloud Run) ## Architecture (recommended) - Cloud Run: runs the FastAPI backend (WebSockets + API) - Cloud SQL (Postgres): primary database - Pub/Sub: cross-instance event fanout for hospital dashboard - Optional: Cloud Storage + CDN for static hosting (or serve static from Cloud Run) ## New env vars Set these in your environment (Cloud Run or local): - `DATABASE_URL` (Postgres connection string) - `DATABASE_MAX_CONNECTIONS` (default 5) - `GCP_PROJECT_ID` (e.g., `my-project`) - `GCP_PUBSUB_TOPIC` (topic name or full topic path) - `GCP_PUBSUB_SUBSCRIPTION_PREFIX` (optional, default `relay-events`) ## Cloud SQL (Postgres) 1. Create a Cloud SQL Postgres instance. 2. Create a database (e.g., `relay`) and user. 3. Make sure your Cloud Run service account has `Cloud SQL Client` role. ### Local dev (Cloud SQL Auth Proxy) 1. Download and run the proxy: ```bash ./cloud-sql-proxy :: --port 5432 ``` 2. Set: ```bash export DATABASE_URL='postgresql://USER:PASSWORD@127.0.0.1:5432/relay' ``` 3. Start the app as usual; tables are created on startup. ### Cloud Run connection (Unix socket) Use the Cloud SQL connection name in your Cloud Run service and set: ```bash export DATABASE_URL='postgresql://USER:PASSWORD@/relay?host=/cloudsql/::' ``` ## Pub/Sub (for multi-instance streaming) 1. Enable Pub/Sub API. 2. Create a topic (e.g., `relay-events`). 3. Set: ```bash export GCP_PROJECT_ID='my-project' export GCP_PUBSUB_TOPIC='relay-events' ``` 4. Ensure your Cloud Run service account has `Pub/Sub Publisher` and `Pub/Sub Subscriber` roles. ## Cloud Run deploy 1. Build a container. 2. Deploy with: - WebSockets enabled (default for Cloud Run) - Min instances (optional for low-latency) - Environment variables from above ## Notes - If `GCP_PROJECT_ID` and `GCP_PUBSUB_TOPIC` are set but `google-cloud-pubsub` isn't installed, the app falls back to in-memory events (single instance only). - If `DATABASE_URL` is not set, the app defaults to local SQLite. ``` ### requirements.txt ``` fastapi==0.115.6 uvicorn[standard]==0.34.0 websockets==14.1 aiosqlite==0.20.0 asyncpg==0.30.0 google-cloud-pubsub==2.31.0 openai==1.59.3 anthropic>=0.40.0 python-dotenv==1.0.1 pydantic>=2.10.4,<3 httpx>=0.28.0 twilio==9.10.1 pypdf>=4.2.0 # PDF reading (gp_documents) pytesseract==0.3.10 pdf2image==1.17.0 # Dev dependencies pytest==9.0.2 pytest-asyncio==1.3.0 pytest-cov==7.0.0 ruff==0.15.1 pyright==1.1.408 ``` ### pyproject.toml ``` [project] name = "relay" version = "0.1.0" description = "AI Emergency Response System - Automated ePCR for Paramedics" requires-python = ">=3.11" [tool.pytest.ini_options] testpaths = ["tests"] asyncio_mode = "auto" addopts = [ "-v", "--tb=short", "--cov=app", "--cov-report=term-missing", "--cov-fail-under=60", ] [tool.coverage.run] source = ["app"] branch = true omit = ["tests/*", "*/__init__.py"] [tool.coverage.report] show_missing = true exclude_also = [ "pragma: no cover", "if TYPE_CHECKING:", "if __name__ == .__main__.:", "raise NotImplementedError", ] [tool.ruff] target-version = "py311" line-length = 100 src = ["app", "tests"] [tool.ruff.lint] select = ["E", "W", "F", "I", "B", "UP", "N", "S", "C4", "RUF"] ignore = [ "S101", # assert in tests is fine "E501", # line length handled by formatter ] [tool.ruff.lint.per-file-ignores] "tests/**/*.py" = ["S101", "S106"] [tool.ruff.lint.isort] known-first-party = ["app"] [tool.pyright] include = ["app"] exclude = ["**/__pycache__", ".venv", "tests"] typeCheckingMode = "basic" pythonVersion = "3.11" reportMissingImports = "error" reportMissingTypeStubs = false ``` ### app/main.py ```python import logging from pathlib import Path from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from app.config import GP_DOCUMENT_PATH from app.database import close_db, init_db from app.routers import cases, gp_call, hospital, stream logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Starting Relay...") await init_db() logger.info("Database initialized") yield await close_db() logger.info("Relay shut down") app = FastAPI( title="Relay", description="AI Emergency Response System - Automated ePCR for Paramedics", version="0.1.0", lifespan=lifespan, ) # Include routers app.include_router(stream.router) app.include_router(cases.router) app.include_router(hospital.router) app.include_router(gp_call.router, prefix="/api") # Serve static files app.mount("/static", StaticFiles(directory="static"), name="static") @app.get("/") async def serve_paramedic_ui(): return FileResponse("static/index_enhanced.html") @app.get("/hospital") async def serve_hospital_ui(): return FileResponse("static/hospital.html") @app.get("/api/documents/gp-record") async def get_gp_record(): path = Path(GP_DOCUMENT_PATH) if not path.exists(): raise HTTPException(status_code=404, detail="GP document not found") return FileResponse(path, media_type="application/pdf", filename=path.name) ``` ### static/app.js ```javascript // Relay - Paramedic UI let currentCaseId = null; let ws = null; let mediaStream = null; let audioContext = null; let processorNode = null; let segmentCount = 0; let downstreamState = { gp: null, medDb: null }; // --- Case Management --- async function newCase() { try { const resp = await fetch("/api/cases", { method: "POST", headers: { "Content-Type": "application/json" }, body: "{}" }); const data = await resp.json(); currentCaseId = data.id; document.getElementById("caseIdDisplay").textContent = `Case: ${currentCaseId.slice(0, 8)}...`; document.getElementById("btnStartStream").disabled = false; document.getElementById("btnNewCase").disabled = true; setStatus("active", "Ready"); clearUI(); } catch (e) { console.error("Failed to create case:", e); setStatus("error", "Error"); } } // --- Audio Streaming --- async function startStream() { if (!currentCaseId) return; // Connect WebSocket const proto = location.protocol === "https:" ? "wss:" : "ws:"; ws = new WebSocket(`${proto}//${location.host}/ws/stream/${currentCaseId}`); ws.onopen = () => { console.log("WebSocket connected"); startMicrophone(); }; ws.onmessage = (event) => { const msg = JSON.parse(event.data); handleServerMessage(msg); }; ws.onclose = () => { console.log("WebSocket closed"); stopMicrophone(); }; ws.onerror = (e) => { console.error("WebSocket error:", e); setStatus("error", "WS Error"); }; document.getElementById("btnStartStream").disabled = true; document.getElementById("btnStopStream").disabled = false; document.getElementById("recordingIndicator").style.display = "flex"; setStatus("active", "Streaming"); } function stopStream() { if (ws && ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: "end_call" })); ws.close(); } stopMicrophone(); document.getElementById("btnStartStream").disabled = true; document.getElementById("btnStopStream").disabled = true; document.getElementById("btnNewCase").disabled = false; document.getElementById("recordingIndicator").style.display = "none"; setStatus("idle", "Completed"); } async function startMicrophone() { try { mediaStream = await navigator.mediaDevices.getUserMedia({ audio: { sampleRate: 16000, channelCount: 1, echoCancellation: true, noiseSuppression: true, }, }); audioContext = new AudioContext({ sampleRate: 16000 }); const source = audioContext.createMediaStreamSource(mediaStream); // Use ScriptProcessorNode for PCM capture processorNode = audioContext.createScriptProcessor(4096, 1, 1); processorNode.onaudioprocess = (e) => { if (ws && ws.readyState === WebSocket.OPEN) { const float32 = e.inputBuffer.getChannelData(0); const int16 = float32ToInt16(float32); const base64 = arrayBufferToBase64(int16.buffer); ws.send(JSON.stringify({ type: "audio_chunk", data: base64 })); } }; source.connect(processorNode); processorNode.connect(audioContext.destination); } catch (e) { console.error("Microphone access failed:", e); setStatus("error", "Mic Error"); } } function stopMicrophone() { if (processorNode) { processorNode.disconnect(); processorNode = null; } if (audioContext) { audioContext.close(); audioContext = null; } if (mediaStream) { mediaStream.getTracks().forEach((t) => t.stop()); mediaStream = null; } } // --- Audio Helpers --- function float32ToInt16(float32Array) { const int16 = new Int16Array(float32Array.length); for (let i = 0; i < float32Array.length; i++) { const s = Math.max(-1, Math.min(1, float32Array[i])); int16[i] = s < 0 ? s * 0x8000 : s * 0x7fff; } return int16; } function arrayBufferToBase64(buffer) { const bytes = new Uint8Array(buffer); let binary = ""; for (let i = 0; i < bytes.byteLength; i++) { binary += String.fromCharCode(bytes[i]); } return btoa(binary); } // --- Server Message Handling --- function handleServerMessage(msg) { switch (msg.type) { case "transcript_partial": updatePartialTranscript(msg.text); break; case "transcript_committed": addCommittedSegment(msg.text); break; case "nemsis_update": updateNEMSIS(msg.nemsis); break; case "core_info_complete": onCoreInfoComplete(); break; case "medical_db_complete": downstreamState.medDb = msg.medical_db_response; showDownstream(downstreamState.gp, downstreamState.medDb); break; case "gp_call_complete": downstreamState.gp = msg.gp_response; showDownstream(downstreamState.gp, downstreamState.medDb); break; case "downstream_complete": showDownstream(msg.gp_response, msg.medical_db_response); break; case "error": console.error("Server error:", msg.message); setStatus("error", "Error"); break; } } // --- UI Updates --- function updatePartialTranscript(text) { let partial = document.getElementById("partialText"); if (!partial) { partial = document.createElement("div"); partial.id = "partialText"; partial.className = "transcript-partial"; document.getElementById("transcriptArea").appendChild(partial); } partial.textContent = text; scrollTranscript(); } function addCommittedSegment(text) { // Remove partial const partial = document.getElementById [truncated — 4673 more characters] ``` ### tests/conftest.py ```python import os import pytest import pytest_asyncio from fastapi.testclient import TestClient from httpx import ASGITransport, AsyncClient # In-memory DB and no external API keys for tests os.environ["OPENAI_API_KEY"] = "" os.environ["ELEVENLABS_API_KEY"] = "" os.environ["PERPLEXITY_API_KEY"] = "" os.environ["DATABASE_PATH"] = ":memory:" os.environ["DATABASE_URL"] = "" os.environ["SEED_DEMO_CASES"] = "false" from app.database import close_db, init_db from app.main import app # Disable external service calls — tests use synthetic dummy data import app.services.fhir_client as _fhir_mod import app.services.gp_lookup as _gp_lookup_mod _fhir_mod.FHIR_SERVERS = [] _gp_lookup_mod.PERPLEXITY_API_KEY = "" @pytest_asyncio.fixture async def db(): """Provide a fresh in-memory database for each test.""" import app.database as db_mod # Close any existing connection if db_mod._db is not None: try: await db_mod._db.close() except Exception: pass db_mod._db = None # Override module-level config directly (avoids fragile importlib.reload) db_mod.DATABASE_PATH = ":memory:" db_mod.DATABASE_URL = "" db_mod.SEED_DEMO_CASES = False await init_db() database = await db_mod.get_db() yield database await close_db() @pytest.fixture def client(db): """Provide a synchronous TestClient for HTTP endpoint tests.""" return TestClient(app) @pytest_asyncio.fixture async def async_client(db): """Provide an async httpx client for async HTTP tests.""" async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as ac: yield ac ``` ### tests/test_gp_lookup.py ```python """Tests for GP lookup service — Perplexity Sonar API integration.""" from app.services.gp_lookup import _validate_phone, lookup_gp_phone # --- Phone Validation --- class TestValidatePhone: def test_valid_us_number(self): assert _validate_phone("+1-555-0123") == "+1-555-0123" def test_valid_digits_only(self): assert _validate_phone("5550123456") == "5550123456" def test_valid_e164(self): assert _validate_phone("+15550123456") == "+15550123456" def test_valid_formatted(self): assert _validate_phone("(555) 012-3456") == "(555) 012-3456" def test_too_short(self): assert _validate_phone("123") is None def test_empty(self): assert _validate_phone("") is None def test_none(self): assert _validate_phone(None) is None def test_no_digits(self): assert _validate_phone("not a number") is None # --- Dummy Mode Lookup --- async def test_lookup_dummy_mode_returns_result(): """In dummy mode, lookup returns synthetic GP contact.""" result = await lookup_gp_phone( gp_name="Dr. Wilson", location="Springfield", practice_name="Greenfield Medical Center", ) assert result is not None assert "phone" in result assert "practice_name" in result assert "address" in result assert "source" in result async def test_lookup_dummy_mode_with_practice_name(): """Dummy mode includes practice name if provided.""" result = await lookup_gp_phone( gp_name="Dr. Smith", location="Chicago", practice_name="Smith Family Practice", ) assert result is not None assert result["practice_name"] == "Smith Family Practice" async def test_lookup_dummy_mode_without_practice_name(): """Dummy mode generates practice name from GP name.""" result = await lookup_gp_phone( gp_name="Dr. Jones", location="New York", ) assert result is not None assert "Dr. Jones" in result["practice_name"] async def test_lookup_dummy_mode_phone_valid(): """Dummy mode returns a validatable phone number.""" result = await lookup_gp_phone( gp_name="Dr. Wilson", location="Springfield", ) assert result is not None phone = _validate_phone(result["phone"]) assert phone is not None ``` ### app/config.py ```python import os from pathlib import Path from dotenv import load_dotenv load_dotenv() ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY", "") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY", "") # LLM configuration LLM_PROVIDER = os.getenv("LLM_PROVIDER", "auto") LLM_DEFAULT_TIER = os.getenv("LLM_DEFAULT_TIER", "fast") LLM_MODEL_FAST = os.getenv("LLM_MODEL_FAST", "") LLM_MODEL_STANDARD = os.getenv("LLM_MODEL_STANDARD", "") LLM_MODEL_HIGH = os.getenv("LLM_MODEL_HIGH", "") # Demo/Debug mode (explicit) DUMMY_MODE = os.getenv("DUMMY_MODE", "false").lower() in ("1", "true", "yes", "on") VOICE_DUMMY = os.getenv("VOICE_DUMMY", "false").lower() in ("1", "true", "yes", "on") GP_CALLS_ENABLED = os.getenv("GP_CALLS_ENABLED", "false").lower() in ("1", "true", "yes", "on") DATABASE_PATH = os.getenv("DATABASE_PATH", "relay.db") DATABASE_URL = os.getenv("DATABASE_URL", "") DATABASE_MAX_CONNECTIONS = int(os.getenv("DATABASE_MAX_CONNECTIONS", "5")) SEED_DEMO_CASES = os.getenv("SEED_DEMO_CASES", "true").lower() == "true" BASE_DIR = Path(__file__).resolve().parent.parent GP_DOCUMENT_PATH = os.getenv( "GP_DOCUMENT_PATH", str(BASE_DIR / "data" / "doc" / "Medical Record.pdf"), ) GP_DOCUMENT_DELAY_SECONDS = int(os.getenv("GP_DOCUMENT_DELAY_SECONDS", "60")) GP_CALL_PENDING_SECONDS = int(os.getenv("GP_CALL_PENDING_SECONDS", "8")) # Perplexity Sonar API (GP contact resolution) PERPLEXITY_API_KEY = os.getenv("PERPLEXITY_API_KEY", "") # Demo override for medical history lookup FHIR_DEMO_PATIENT_URL = os.getenv( "FHIR_DEMO_PATIENT_URL", "https://hapi.fhir.org/baseR4/Patient/131273059/$everything", ) # Twilio (outbound voice calls) TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID", "") TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN", "") TWILIO_PHONE_NUMBER = os.getenv("TWILIO_PHONE_NUMBER", "") # ElevenLabs Conversational AI (voice agent) ELEVENLABS_AGENT_ID = os.getenv("ELEVENLABS_AGENT_ID", "") ELEVENLABS_PHONE_NUMBER_ID = os.getenv("ELEVENLABS_PHONE_NUMBER_ID", "") # Hospital callback number for GP voicemail HOSPITAL_CALLBACK_NUMBER = os.getenv("HOSPITAL_CALLBACK_NUMBER", "+1-555-0100") # Email address for GPs to send medical records to RECORDS_EMAIL = os.getenv("RECORDS_EMAIL", "records@relay.health") # GCP (optional) GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID", "") GCP_PUBSUB_TOPIC = os.getenv("GCP_PUBSUB_TOPIC", "") GCP_PUBSUB_SUBSCRIPTION_PREFIX = os.getenv("GCP_PUBSUB_SUBSCRIPTION_PREFIX", "relay-events") ``` ### tests/test_voice_agent.py ```python """Tests for voice agent service — ElevenLabs + Twilio outbound calls.""" from app.services.voice_agent import _dummy_call, place_gp_call # --- Dummy Call --- class TestDummyCall: def test_returns_expected_keys(self): result = _dummy_call("John Smith", "case-123") assert "call_sid" in result assert "conversation_id" in result assert "status" in result assert "transcript" in result def test_status_is_dummy(self): result = _dummy_call("Jane Doe", "case-456") assert result["status"] == "dummy" def test_transcript_contains_patient_name(self): result = _dummy_call("Alice Johnson", "case-789") assert "Alice Johnson" in result["transcript"] def test_call_sid_contains_case_id(self): result = _dummy_call("Test", "case-abc") assert "case-abc" in result["call_sid"] def test_conversation_id_contains_case_id(self): result = _dummy_call("Test", "case-abc") assert "case-abc" in result["conversation_id"] def test_no_case_id(self): result = _dummy_call("Test", None) assert result["call_sid"] is not None assert result["status"] == "dummy" # --- Place GP Call (dummy mode) --- async def test_place_gp_call_dummy_mode(): """In dummy mode, place_gp_call returns synthetic result without API call.""" result = await place_gp_call( phone_number="+1-555-0123", patient_name="John Smith", patient_dob="1980-01-15", hospital_callback="+1-555-0100", case_id="test-case-001", ) assert result["status"] == "dummy" assert "John Smith" in result["transcript"] assert result["call_sid"] is not None assert result["conversation_id"] is not None async def test_place_gp_call_dummy_no_dob(): """Dummy mode works without DOB.""" result = await place_gp_call( phone_number="+1-555-0123", patient_name="Jane Doe", patient_dob=None, case_id="test-case-002", ) assert result["status"] == "dummy" assert "Jane Doe" in result["transcript"] async def test_place_gp_call_dummy_transcript_has_medical_info(): """Dummy transcript includes clinical information.""" result = await place_gp_call( phone_number="+1-555-0123", patient_name="Test Patient", patient_dob="1990-01-01", case_id="test-case-003", ) transcript = result["transcript"] # Dummy transcript should mention medications or allergies assert "medication" in transcript.lower() or "allerg" in transcript.lower() ``` ### tests/test_event_bus.py ```python """Tests for the CaseEventBus pub/sub system.""" import asyncio from app.services.event_bus import CaseEventBus class TestCaseEventBus: async def test_global_subscribe_receives_events(self): bus = CaseEventBus() queue = bus.subscribe_all() await bus.publish("case-1", {"type": "test", "data": "hello"}) event = queue.get_nowait() assert event["case_id"] == "case-1" assert event["type"] == "test" assert event["data"] == "hello" async def test_case_specific_subscribe(self): bus = CaseEventBus() q1 = bus.subscribe("case-1") q2 = bus.subscribe("case-2") await bus.publish("case-1", {"type": "update"}) event = q1.get_nowait() assert event["case_id"] == "case-1" assert q2.empty() async def test_unsubscribe_all(self): bus = CaseEventBus() queue = bus.subscribe_all() bus.unsubscribe_all(queue) await bus.publish("case-1", {"type": "test"}) assert queue.empty() async def test_unsubscribe_case(self): bus = CaseEventBus() queue = bus.subscribe("case-1") bus.unsubscribe("case-1", queue) await bus.publish("case-1", {"type": "test"}) assert queue.empty() async def test_multiple_global_subscribers(self): bus = CaseEventBus() q1 = bus.subscribe_all() q2 = bus.subscribe_all() await bus.publish("case-1", {"type": "test"}) assert not q1.empty() assert not q2.empty() assert q1.get_nowait()["type"] == "test" assert q2.get_nowait()["type"] == "test" async def test_both_global_and_case_subscribers(self): bus = CaseEventBus() global_q = bus.subscribe_all() case_q = bus.subscribe("case-1") await bus.publish("case-1", {"type": "test"}) assert not global_q.empty() assert not case_q.empty() async def test_unsubscribe_nonexistent_case(self): bus = CaseEventBus() queue: asyncio.Queue = asyncio.Queue() # Should not raise bus.unsubscribe("nonexistent", queue) async def test_unsubscribe_cleans_up_empty_set(self): bus = CaseEventBus() queue = bus.subscribe("case-1") bus.unsubscribe("case-1", queue) assert "case-1" not in bus._subscribers async def test_publish_adds_case_id(self): bus = CaseEventBus() queue = bus.subscribe_all() await bus.publish("my-case", {"type": "hello"}) event = queue.get_nowait() assert event["case_id"] == "my-case" ``` [42 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]