Upload 3 files

app/api/dashboard.py

@@ -0,0 +1,236 @@
+"""
+Multi-patient dashboard API endpoints.
+
+Provides aggregate views for monitoring multiple patients simultaneously.
+"""
+from typing import Any, Dict, List, Optional
+from datetime import datetime, timedelta
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import func, and_
+from sqlalchemy.orm import Session
+
+from app.db.session import get_session
+from app.models.ecg import ECGSample
+from app.models.schemas import DashboardStats, PatientSummary, AlertSummary
+
+router = APIRouter()
+
+
+@router.get("/stats", response_model=DashboardStats)
+def get_dashboard_stats(
+    session: Session = Depends(get_session),
+    hours: int = Query(24, ge=1, le=168, description="Time window in hours")
+) -> Dict[str, Any]:
+    """
+    Get aggregate statistics for the dashboard.
+
+    Returns:
+        - Total patients monitored
+        - Total samples processed
+        - Alert distribution (none, notify, escalate)
+        - Average scores
+        - Energy savings estimate
+    """
+    cutoff_time = datetime.utcnow() - timedelta(hours=hours)
+
+    # Total samples in time window
+    total_samples = session.query(func.count(ECGSample.id)).filter(
+        ECGSample.created_at >= cutoff_time
+    ).scalar() or 0
+
+    # Unique patients
+    unique_patients = session.query(func.count(func.distinct(ECGSample.patient_id))).filter(
+        ECGSample.created_at >= cutoff_time
+    ).scalar() or 0
+
+    # Alert distribution
+    alert_counts = session.query(
+        ECGSample.alert_level,
+        func.count(ECGSample.id)
+    ).filter(
+        ECGSample.created_at >= cutoff_time
+    ).group_by(ECGSample.alert_level).all()
+
+    alert_distribution = {level: count for level, count in alert_counts}
+
+    # Average score
+    avg_score = session.query(func.avg(ECGSample.score)).filter(
+        ECGSample.created_at >= cutoff_time
+    ).scalar() or 0.0
+
+    # Label distribution
+    label_counts = session.query(
+        ECGSample.label,
+        func.count(ECGSample.id)
+    ).filter(
+        ECGSample.created_at >= cutoff_time
+    ).group_by(ECGSample.label).all()
+
+    label_distribution = {label: count for label, count in label_counts}
+
+    # Estimated energy savings (assume 60% average from gating)
+    estimated_energy_savings_pct = 60.0
+
+    return {
+        "time_window_hours": hours,
+        "total_samples": total_samples,
+        "unique_patients": unique_patients,
+        "alert_distribution": alert_distribution,
+        "label_distribution": label_distribution,
+        "avg_score": round(float(avg_score), 3),
+        "estimated_energy_savings_pct": estimated_energy_savings_pct,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+
+
+@router.get("/patients", response_model=List[PatientSummary])
+def get_patient_summaries(
+    session: Session = Depends(get_session),
+    alert_level: Optional[str] = Query(None, description="Filter by alert level"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> List[Dict[str, Any]]:
+    """
+    Get summary information for all patients.
+
+    Returns list of patients with their latest sample and alert status.
+    """
+    # Subquery to get latest sample per patient
+    from sqlalchemy import distinct
+    from sqlalchemy.sql import exists
+
+    # Get distinct patient IDs
+    patient_ids = session.query(distinct(ECGSample.patient_id)).all()
+    patient_ids = [pid[0] for pid in patient_ids]
+
+    summaries = []
+
+    for patient_id in patient_ids[:limit]:
+        # Get latest sample for this patient
+        latest_sample = session.query(ECGSample).filter(
+            ECGSample.patient_id == patient_id
+        ).order_by(ECGSample.created_at.desc()).first()
+
+        if not latest_sample:
+            continue
+
+        # Filter by alert level if specified
+        if alert_level and latest_sample.alert_level != alert_level:
+            continue
+
+        # Count total samples for this patient
+        sample_count = session.query(func.count(ECGSample.id)).filter(
+            ECGSample.patient_id == patient_id
+        ).scalar() or 0
+
+        # Count alerts
+        alert_count = session.query(func.count(ECGSample.id)).filter(
+            and_(
+                ECGSample.patient_id == patient_id,
+                ECGSample.alert_level.in_(['notify', 'escalate'])
+            )
+        ).scalar() or 0
+
+        summaries.append({
+            "patient_id": patient_id,
+            "latest_label": latest_sample.label,
+            "latest_score": round(float(latest_sample.score or 0.0), 3),
+            "latest_alert_level": latest_sample.alert_level,
+            "latest_hr": latest_sample.hr,
+            "last_updated": latest_sample.created_at.isoformat(),
+            "total_samples": sample_count,
+            "alert_count": alert_count,
+        })
+
+    # Sort by alert level priority (escalate > notify > none)
+    alert_priority = {'escalate': 0, 'notify': 1, 'none': 2, None: 3}
+    summaries.sort(key=lambda x: alert_priority.get(x['latest_alert_level'], 3))
+
+    return summaries
+
+
+@router.get("/alerts", response_model=List[AlertSummary])
+def get_recent_alerts(
+    session: Session = Depends(get_session),
+    hours: int = Query(24, ge=1, le=168),
+    alert_level: Optional[str] = Query(None, description="Filter: notify or escalate"),
+    limit: int = Query(50, ge=1, le=500),
+) -> List[Dict[str, Any]]:
+    """
+    Get recent alerts across all patients.
+
+    Returns samples with alert_level in ['notify', 'escalate'], sorted by recency.
+    """
+    cutoff_time = datetime.utcnow() - timedelta(hours=hours)
+
+    query = session.query(ECGSample).filter(
+        and_(
+            ECGSample.created_at >= cutoff_time,
+            ECGSample.alert_level.in_(['notify', 'escalate'])
+        )
+    )
+
+    if alert_level:
+        query = query.filter(ECGSample.alert_level == alert_level)
+
+    alerts = query.order_by(ECGSample.created_at.desc()).limit(limit).all()
+
+    return [
+        {
+            "sample_id": alert.id,
+            "patient_id": alert.patient_id,
+            "alert_level": alert.alert_level,
+            "label": alert.label,
+            "score": round(float(alert.score or 0.0), 3),
+            "hr": alert.hr,
+            "timestamp": alert.created_at.isoformat(),
+        }
+        for alert in alerts
+    ]
+
+
+@router.get("/patient/{patient_id}/history")
+def get_patient_history(
+    patient_id: str,
+    session: Session = Depends(get_session),
+    hours: int = Query(24, ge=1, le=168),
+    limit: int = Query(100, ge=1, le=1000),
+) -> Dict[str, Any]:
+    """
+    Get historical data for a specific patient.
+
+    Returns time series of samples, labels, scores, alerts.
+    """
+    cutoff_time = datetime.utcnow() - timedelta(hours=hours)
+
+    samples = session.query(ECGSample).filter(
+        and_(
+            ECGSample.patient_id == patient_id,
+            ECGSample.created_at >= cutoff_time
+        )
+    ).order_by(ECGSample.created_at.asc()).limit(limit).all()
+
+    history = [
+        {
+            "sample_id": s.id,
+            "label": s.label,
+            "score": round(float(s.score or 0.0), 3),
+            "alert_level": s.alert_level,
+            "hr": s.hr,
+            "timestamp": s.created_at.isoformat(),
+        }
+        for s in samples
+    ]
+
+    # Compute summary stats
+    alert_count = sum(1 for s in samples if s.alert_level in ['notify', 'escalate'])
+    avg_score = sum(s.score or 0.0 for s in samples) / max(len(samples), 1)
+
+    return {
+        "patient_id": patient_id,
+        "time_window_hours": hours,
+        "sample_count": len(samples),
+        "alert_count": alert_count,
+        "avg_score": round(float(avg_score), 3),
+        "history": history,
+    }

app/api/streaming.py

@@ -0,0 +1,150 @@
+"""
+WebSocket streaming endpoint for real-time ECG monitoring.
+
+Supports:
+- Live ECG signal streaming
+- Real-time gating, inference, and rule evaluation
+- Multi-patient concurrent monitoring
+"""
+import asyncio
+import json
+from typing import Any, Dict, List, Optional
+from datetime import datetime
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends
+from sqlalchemy.orm import Session
+
+from app.db.session import get_session
+from app.ml.gating import gate_signal
+from app.ml.inference import infer_ecg
+from app.rules.engine import evaluate_ecg_rules
+from app.models.ecg import ECGSample
+
+router = APIRouter()
+
+
+class ConnectionManager:
+    """Manages WebSocket connections for multiple patients."""
+
+    def __init__(self):
+        self.active_connections: Dict[str, WebSocket] = {}
+
+    async def connect(self, patient_id: str, websocket: WebSocket):
+        await websocket.accept()
+        self.active_connections[patient_id] = websocket
+
+    def disconnect(self, patient_id: str):
+        if patient_id in self.active_connections:
+            del self.active_connections[patient_id]
+
+    async def send_message(self, patient_id: str, message: Dict[str, Any]):
+        if patient_id in self.active_connections:
+            await self.active_connections[patient_id].send_json(message)
+
+    def get_active_patients(self) -> List[str]:
+        return list(self.active_connections.keys())
+
+
+manager = ConnectionManager()
+
+
+@router.websocket("/ws/ecg/{patient_id}")
+async def ecg_stream(websocket: WebSocket, patient_id: str):
+    """
+    WebSocket endpoint for streaming ECG data.
+
+    Protocol:
+    - Client sends: {"signal": [float, ...], "patient_context": {...}}
+    - Server responds: {"status": "ok", "result": {...}, "timestamp": "..."}
+
+    The server runs gating → inference → rules and streams back results.
+    """
+    await manager.connect(patient_id, websocket)
+
+    try:
+        while True:
+            # Receive signal chunk from client
+            data = await websocket.receive_json()
+
+            signal = data.get("signal", [])
+            patient_context = data.get("patient_context", {})
+            patient_context["patient_id"] = patient_id
+
+            if not signal:
+                await websocket.send_json({
+                    "status": "error",
+                    "message": "Empty signal",
+                    "timestamp": datetime.utcnow().isoformat()
+                })
+                continue
+
+            # Process signal through pipeline
+            try:
+                # Step 1: Gating
+                gated, gating_meta = gate_signal(signal)
+
+                # Step 2: Inference
+                model_output = infer_ecg(gated, original_len=len(signal), gating_meta=gating_meta)
+
+                # Step 3: Rules
+                rules_result = evaluate_ecg_rules(patient_context, model_output)
+
+                # Build response
+                result = {
+                    "patient_id": patient_id,
+                    "label": model_output.get("label"),
+                    "score": round(float(model_output.get("score", 0.0)), 3),
+                    "hr": model_output.get("hr"),
+                    "alert_level": rules_result.get("alert_level"),
+                    "explanations": rules_result.get("explanations", []),
+                    "gating": {
+                        "ratio": round(gating_meta.get("ratio", 1.0), 3),
+                        "selected_windows": gating_meta.get("selected_windows", 0),
+                        "total_windows": gating_meta.get("total_windows", 0),
+                        "energy_saved_pct": round((1 - gating_meta.get("ratio", 1.0)) * 100, 1),
+                    },
+                    "timestamp": datetime.utcnow().isoformat(),
+                }
+
+                await websocket.send_json({
+                    "status": "ok",
+                    "result": result,
+                    "timestamp": datetime.utcnow().isoformat(),
+                })
+
+            except Exception as e:
+                await websocket.send_json({
+                    "status": "error",
+                    "message": str(e),
+                    "timestamp": datetime.utcnow().isoformat(),
+                })
+
+    except WebSocketDisconnect:
+        manager.disconnect(patient_id)
+        print(f"Patient {patient_id} disconnected")
+
+
+@router.get("/active-streams")
+async def get_active_streams() -> Dict[str, Any]:
+    """Get list of active patient streams."""
+    return {
+        "active_patients": manager.get_active_patients(),
+        "count": len(manager.get_active_patients()),
+    }
+
+
+@router.post("/broadcast-alert")
+async def broadcast_alert(alert: Dict[str, Any]):
+    """
+    Broadcast an alert to all connected patients (admin use).
+
+    Example: System-wide maintenance notification.
+    """
+    for patient_id in manager.get_active_patients():
+        await manager.send_message(patient_id, {
+            "type": "system_alert",
+            "message": alert.get("message", ""),
+            "timestamp": datetime.utcnow().isoformat(),
+        })
+
+    return {"status": "ok", "recipients": len(manager.get_active_patients())}