feat: Add basic chart for vital signs observations

api/pyproject.toml

@@ -36,6 +36,9 @@ target-version = "py311"
 [tool.ruff.lint]
 select = ["E", "F", "I", "UP", "B"]
 
+[tool.ruff.lint.flake8-bugbear]
+extend-immutable-calls = ["fastapi.Query", "fastapi.Depends", "fastapi.Path"]
+
 [tool.mypy]
 python_version = "3.11"
 # Enable important checks while being pragmatic about strictness

api/scripts/upload_templates.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""Upload openEHR templates to EHRBase.
+
+Usage:
+    python scripts/upload_templates.py [--check-only]
+
+This script uploads all OPT templates from the templates/ directory to EHRBase.
+Run this after starting EHRBase to ensure all required templates are available.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+import httpx
+
+# Add parent directory to path for imports
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.config import settings
+
+
+async def list_templates(client: httpx.AsyncClient) -> list[str]:
+    """List existing templates in EHRBase."""
+    response = await client.get("/openehr/v1/definition/template/adl1.4")
+    if response.status_code == 200:
+        templates = response.json()
+        return [t.get("template_id", "") for t in templates]
+    return []
+
+
+async def upload_template(client: httpx.AsyncClient, template_path: Path) -> bool:
+    """Upload a single template to EHRBase."""
+    template_content = template_path.read_text()
+    template_id = template_path.stem  # filename without extension
+
+    print(f"  Uploading {template_id}...")
+
+    response = await client.post(
+        "/openehr/v1/definition/template/adl1.4",
+        content=template_content,
+        headers={"Content-Type": "application/xml"},
+    )
+
+    if response.status_code in (200, 201, 204):
+        print(f"  ✓ {template_id} uploaded successfully")
+        return True
+    elif response.status_code == 409:
+        print(f"  ○ {template_id} already exists")
+        return True
+    else:
+        print(f"  ✗ {template_id} failed: {response.status_code}")
+        try:
+            error_detail = response.json()
+            print(f"    Error: {error_detail}")
+        except Exception:
+            print(f"    Response: {response.text[:500]}")
+        return False
+
+
+async def main(check_only: bool = False):
+    """Upload all templates to EHRBase."""
+    templates_dir = Path(__file__).parent.parent / "templates"
+
+    if not templates_dir.exists():
+        print(f"Templates directory not found: {templates_dir}")
+        sys.exit(1)
+
+    template_files = list(templates_dir.glob("*.opt"))
+    if not template_files:
+        print("No .opt template files found")
+        sys.exit(0)
+
+    print(f"Found {len(template_files)} template(s)")
+    print(f"EHRBase URL: {settings.ehrbase_url}")
+    print()
+
+    auth = None
+    if settings.ehrbase_user and settings.ehrbase_password:
+        auth = httpx.BasicAuth(settings.ehrbase_user, settings.ehrbase_password)
+
+    async with httpx.AsyncClient(base_url=settings.ehrbase_url, auth=auth) as client:
+        # Check EHRBase connectivity
+        try:
+            response = await client.get("/ehrbase/rest/status")
+            if response.status_code != 200:
+                print(f"EHRBase not ready: {response.status_code}")
+                sys.exit(1)
+            print("EHRBase is ready")
+        except httpx.ConnectError:
+            print("Cannot connect to EHRBase")
+            sys.exit(1)
+
+        # List existing templates
+        existing = await list_templates(client)
+        print(f"Existing templates: {len(existing)}")
+        for t in existing:
+            print(f"  - {t}")
+        print()
+
+        if check_only:
+            print("Check only mode - not uploading")
+            return
+
+        # Upload templates
+        print("Uploading templates...")
+        success_count = 0
+        for template_path in template_files:
+            if await upload_template(client, template_path):
+                success_count += 1
+
+        print()
+        print(f"Uploaded {success_count}/{len(template_files)} templates")
+
+
+if __name__ == "__main__":
+    check_only = "--check-only" in sys.argv
+    asyncio.run(main(check_only))

api/src/ehrbase/queries.py

@@ -1,17 +1,41 @@
 """Common AQL queries for clinical data retrieval."""
 
-# Get all vital signs for a patient
+# Get all vital signs (blood pressure + pulse) for a patient
 VITAL_SIGNS_QUERY = """
 SELECT
     c/uid/value as composition_id,
-    o/data[at0001]/events[at0006]/data[at0003]/items[at0004]/value/magnitude as systolic,
-    o/data[at0001]/events[at0006]/data[at0003]/items[at0005]/value/magnitude as diastolic,
-    o/data[at0001]/events[at0006]/time/value as time
+    c/context/start_time/value as recorded_at,
+    bp/data[at0001]/events[at0006]/data[at0003]/items[at0004]/value/magnitude as systolic,
+    bp/data[at0001]/events[at0006]/data[at0003]/items[at0005]/value/magnitude as diastolic,
+    pulse/data[at0002]/events[at0003]/data[at0001]/items[at0004]/value/magnitude as pulse_rate
 FROM EHR e
 CONTAINS COMPOSITION c
-CONTAINS OBSERVATION o[openEHR-EHR-OBSERVATION.blood_pressure.v2]
+CONTAINS (
+    OBSERVATION bp[openEHR-EHR-OBSERVATION.blood_pressure.v2] OR
+    OBSERVATION pulse[openEHR-EHR-OBSERVATION.pulse.v2]
+)
 WHERE e/ehr_id/value = $ehr_id
-ORDER BY o/data[at0001]/events[at0006]/time/value DESC
+ORDER BY c/context/start_time/value DESC
+"""
+
+# Get vital signs for a patient within a date range
+VITAL_SIGNS_DATE_RANGE_QUERY = """
+SELECT
+    c/uid/value as composition_id,
+    c/context/start_time/value as recorded_at,
+    bp/data[at0001]/events[at0006]/data[at0003]/items[at0004]/value/magnitude as systolic,
+    bp/data[at0001]/events[at0006]/data[at0003]/items[at0005]/value/magnitude as diastolic,
+    pulse/data[at0002]/events[at0003]/data[at0001]/items[at0004]/value/magnitude as pulse_rate
+FROM EHR e
+CONTAINS COMPOSITION c
+CONTAINS (
+    OBSERVATION bp[openEHR-EHR-OBSERVATION.blood_pressure.v2] OR
+    OBSERVATION pulse[openEHR-EHR-OBSERVATION.pulse.v2]
+)
+WHERE e/ehr_id/value = $ehr_id
+AND c/context/start_time/value >= $from_date
+AND c/context/start_time/value <= $to_date
+ORDER BY c/context/start_time/value DESC
 """
 
 # Get all encounters for a patient
@@ -37,3 +61,12 @@
 CONTAINS INSTRUCTION i[openEHR-EHR-INSTRUCTION.medication_order.v3]
 WHERE e/ehr_id/value = $ehr_id
 """
+
+# Count vital signs for a patient
+VITAL_SIGNS_COUNT_QUERY = """
+SELECT COUNT(c/uid/value) as count
+FROM EHR e
+CONTAINS COMPOSITION c
+WHERE e/ehr_id/value = $ehr_id
+AND c/archetype_details/template_id/value = 'open-cis.vital-signs.v1'
+"""

api/src/ehrbase/templates.py

@@ -1,9 +1,18 @@
 """Template management for EHRBase."""
 
+import logging
+from pathlib import Path
 from typing import Any
 
 from src.ehrbase.client import ehrbase_client
 
+logger = logging.getLogger(__name__)
+
+# Template files that must be registered in EHRBase
+REQUIRED_TEMPLATES = [
+    "open-cis.vital-signs.v1",
+]
+
 
 async def list_templates() -> list[dict[str, Any]]:
     """List all available operational templates."""
@@ -24,3 +33,74 @@ async def get_template_example(template_id: str, format: str = "FLAT") -> dict[s
     )
     response.raise_for_status()
     return response.json()
+
+
+async def get_registered_template_ids() -> list[str]:
+    """Get list of template IDs registered in EHRBase."""
+    try:
+        templates = await list_templates()
+        return [t.get("template_id", "") for t in templates]
+    except Exception as e:
+        logger.warning(f"Failed to list EHRBase templates: {e}")
+        return []
+
+
+async def upload_template_file(template_id: str, template_content: str) -> bool:
+    """Upload a single template to EHRBase."""
+    try:
+        await upload_template(template_content)
+        logger.info(f"Template {template_id} uploaded successfully")
+        return True
+    except Exception as e:
+        error_msg = str(e)
+        # 409 means template already exists - that's OK
+        if "409" in error_msg:
+            logger.info(f"Template {template_id} already exists")
+            return True
+        logger.error(f"Failed to upload template {template_id}: {e}")
+        return False
+
+
+async def ensure_templates_registered() -> dict[str, bool]:
+    """
+    Ensure all required templates are registered in EHRBase.
+
+    Called during API startup. Returns a dict mapping template_id to success status.
+    """
+    results: dict[str, bool] = {}
+
+    # Find templates directory (relative to this file)
+    templates_dir = Path(__file__).parent.parent.parent / "templates"
+
+    if not templates_dir.exists():
+        logger.warning(f"Templates directory not found: {templates_dir}")
+        return results
+
+    # Check what's already registered
+    try:
+        existing = await get_registered_template_ids()
+        logger.info(f"EHRBase has {len(existing)} registered template(s)")
+    except Exception as e:
+        logger.warning(f"Could not connect to EHRBase to check templates: {e}")
+        return results
+
+    # Upload any missing required templates
+    for template_id in REQUIRED_TEMPLATES:
+        template_file = templates_dir / f"{template_id}.opt"
+
+        if not template_file.exists():
+            logger.warning(f"Template file not found: {template_file}")
+            results[template_id] = False
+            continue
+
+        if template_id in existing:
+            logger.info(f"Template {template_id} already registered")
+            results[template_id] = True
+            continue
+
+        # Read and upload
+        logger.info(f"Uploading template {template_id}...")
+        template_content = template_file.read_text()
+        results[template_id] = await upload_template_file(template_id, template_content)
+
+    return results

api/src/main.py

@@ -8,6 +8,7 @@
 from src.config import settings
 from src.db.client import prisma
 from src.ehrbase.client import ehrbase_client
+from src.ehrbase.templates import ensure_templates_registered
 from src.encounters.router import router as encounters_router
 from src.observations.router import router as observations_router
 from src.patients.router import router as patients_router
@@ -24,6 +25,17 @@ async def lifespan(app: FastAPI):
         logger.info("Database connected successfully")
     except Exception as e:
         logger.error(f"Failed to connect to database: {e}")
+
+    # Ensure openEHR templates are registered in EHRBase
+    try:
+        template_results = await ensure_templates_registered()
+        if template_results:
+            for template_id, success in template_results.items():
+                status = "registered" if success else "FAILED"
+                logger.info(f"Template {template_id}: {status}")
+    except Exception as e:
+        logger.warning(f"Could not ensure templates are registered: {e}")
+
     yield
     # Shutdown
     if prisma.is_connected():