import asyncio import json import logging import socket from typing import List, Optional from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect, Request from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from database import get_db from models import Device, VitalReading, TransmissionLog logger = logging.getLogger(__name__) router = APIRouter() # === WebSocket Manager for real-time vitals push === class VitalsWebSocketManager: """Manages WebSocket connections for live vitals streaming.""" def __init__(self): self.active_connections: List[WebSocket] = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) logger.info(f"WebSocket client connected ({len(self.active_connections)} active)") def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) logger.info(f"WebSocket client disconnected ({len(self.active_connections)} active)") async def broadcast_vitals(self, vitals_data: dict): """Send vitals to all connected WebSocket clients.""" dead = [] for connection in self.active_connections: try: await connection.send_json(vitals_data) except Exception: dead.append(connection) for d in dead: self.disconnect(d) # Global singleton ws_manager = VitalsWebSocketManager() def get_ws_manager() -> VitalsWebSocketManager: return ws_manager # === Dashboard HTML Route === @router.get("/dashboard", response_class=HTMLResponse) def serve_dashboard(): """Serve the vitals monitoring dashboard.""" import os dashboard_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static", "dashboard.html") try: with open(dashboard_path, "r", encoding="utf-8") as f: return HTMLResponse(content=f.read()) except FileNotFoundError: return HTMLResponse(content="

Dashboard not found

static/dashboard.html is missing

", status_code=404) # === API Endpoints === @router.get("/health") def health_check(): return {"status": "ok", "service": "Patient Monitor Vital Signs Forwarder"} @router.get("/status") def get_system_status(request: Request, db: Session = Depends(get_db)): total_devices = db.query(Device).count() active_devices = db.query(Device).filter(Device.status == "active").count() total_readings = db.query(VitalReading).count() pending_transmissions = db.query(VitalReading).filter(VitalReading.transmitted == False).count() active_ports = getattr(request.app.state, "active_ports", []) return { "total_devices": total_devices, "active_devices": active_devices, "total_readings_stored": total_readings, "pending_transmissions": pending_transmissions, "active_ports": active_ports } @router.get("/devices") def list_devices(db: Session = Depends(get_db)): devices = db.query(Device).all() now = datetime.now(timezone.utc) updated = False for device in devices: last_seen = device.last_seen if last_seen.tzinfo is None: last_seen = last_seen.replace(tzinfo=timezone.utc) # If not seen for more than 15 seconds, mark offline if (now - last_seen).total_seconds() > 15.0: if device.status != "offline": device.status = "offline" updated = True else: if device.status != "active": device.status = "active" updated = True if updated: db.commit() return devices @router.get("/latest-readings") def get_latest_readings(device_id: str = None, limit: int = 10, db: Session = Depends(get_db)): query = db.query(VitalReading) if device_id: query = query.filter(VitalReading.device_id == device_id) readings = query.order_by(VitalReading.timestamp.desc()).limit(limit).all() return readings @router.get("/readings") def get_readings(device_id: str = None, patient_id: str = None, limit: int = 50, db: Session = Depends(get_db)): query = db.query(VitalReading) if device_id: query = query.filter(VitalReading.device_id == device_id) if patient_id: query = query.filter(VitalReading.patient_id == patient_id) return query.order_by(VitalReading.timestamp.desc()).limit(limit).all() @router.get("/transmission-logs") def get_transmission_logs(limit: int = 50, db: Session = Depends(get_db)): logs = db.query(TransmissionLog).order_by(TransmissionLog.timestamp.desc()).limit(limit).all() return logs @router.post("/test-api") def test_target_api_connection(): """ Endpoint to trigger a test POST to the target API. """ from config import settings return { "message": "Test triggered", "target_url": settings.target_api_url } @router.get("/network-info") def get_network_info(): """ Returns this PC's local IP addresses. Useful for configuring the CMS7000PLUS monitor's CMS Server IP. """ from config import settings ips = [] try: for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): ip = info[4][0] if ip not in ips and not ip.startswith('127.'): ips.append(ip) except Exception: pass # Find primary IP primary_ip = None try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) primary_ip = s.getsockname()[0] s.close() if primary_ip not in ips: ips.insert(0, primary_ip) except Exception: pass return { "hostname": socket.gethostname(), "ip_addresses": ips, "primary_ip": primary_ip or (ips[0] if ips else None), "monitor_model": settings.monitor_model, "contec_ports": settings.contec_ports, "instructions": [ f"1. Connect your {settings.monitor_model} to this PC via Ethernet", f"2. On the monitor: System Setup → Network → CMS Settings", f"3. Set Server IP to: {primary_ip or 'your PC IP'}", f"4. Set Server Port to: {settings.contec_ports[0] if settings.contec_ports else 511}", f"5. Enable the CMS connection", f"6. The dashboard will show data automatically" ] } # === Push Vitals Endpoint (for local-to-cloud sync) === from schemas import NormalizedVitals from models import Patient @router.post("/push-vitals") async def push_vitals(vitals: NormalizedVitals, api_key: str = None, db: Session = Depends(get_db)): """ Receive vitals pushed from a local instance and store + broadcast them. This enables the 'split architecture' where: - Local instance receives data from physical monitors via TCP - Local instance POSTs vitals here for cloud storage & dashboard access Optional: pass ?api_key=YOUR_KEY for basic authentication. """ from config import settings # Optional API key check if settings.api_token and api_key != settings.api_token: if settings.api_token: # Only enforce if a token is configured raise HTTPException(status_code=401, detail="Invalid or missing api_key") try: # 1. Device tracking device = db.query(Device).filter(Device.device_id == vitals.device_id).first() if not device: device = Device( device_id=vitals.device_id, ip_address=vitals.ip_address or "remote-push", status="active" ) db.add(device) else: if vitals.ip_address: device.ip_address = vitals.ip_address device.status = "active" device.last_seen = datetime.now(timezone.utc) # 2. Patient tracking from models import VitalReading as VR if vitals.patient_id == "UNKNOWN": last_known_reading = ( db.query(VR) .filter(VR.device_id == vitals.device_id) .filter(VR.patient_id != "UNKNOWN") .order_by(VR.timestamp.desc()) .first() ) if last_known_reading: vitals.patient_id = last_known_reading.patient_id patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first() if not patient: patient = Patient( patient_id=vitals.patient_id, mrn=vitals.patient_id, name=vitals.patient_name or "Unknown", gender=vitals.patient_gender, dob=vitals.patient_dob, bed_id=vitals.bed_id, ) db.add(patient) else: if vitals.patient_name: patient.name = vitals.patient_name if vitals.patient_gender: patient.gender = vitals.patient_gender if vitals.patient_dob: patient.dob = vitals.patient_dob if vitals.bed_id: patient.bed_id = vitals.bed_id # 3. Create vital reading from models import VitalReading as VR reading = VR( device_id=vitals.device_id, patient_id=vitals.patient_id, timestamp=vitals.timestamp, heart_rate=vitals.heart_rate, spo2=vitals.spo2, systolic_bp=vitals.systolic_bp, diastolic_bp=vitals.diastolic_bp, map_bp=vitals.map_bp, respiratory_rate=vitals.respiratory_rate, temperature=vitals.temperature, transmitted=True # Already received = transmitted ) db.add(reading) db.commit() db.refresh(reading) # 4. Broadcast to WebSocket clients await ws_manager.broadcast_vitals(vitals.model_dump(mode="json")) logger.info(f"Push-vitals received for device {vitals.device_id}") return { "status": "ok", "reading_id": reading.id, "device_id": vitals.device_id, "patient_id": vitals.patient_id } except HTTPException: raise except Exception as e: logger.error(f"Push-vitals error: {e}") db.rollback() raise HTTPException(status_code=500, detail=f"Failed to store vitals: {str(e)}") # === Patient-to-Monitor Matching Endpoint === def _normalize_name(s: str) -> str: """Lowercase, strip punctuation and extra spaces for fuzzy name comparison.""" import re if not s: return "" return re.sub(r"[^a-z0-9\s]", "", s.lower()).split() def _normalize_dob(dob: str) -> str: """Normalize DOB to YYYYMMDD regardless of input format (YYYY-MM-DD / YYYYMMDD / DD-MM-YYYY).""" if not dob: return "" dob = dob.strip().replace("/", "-") if len(dob) == 8 and dob.isdigit(): return dob # already YYYYMMDD parts = dob.split("-") if len(parts) == 3: if len(parts[0]) == 4: # YYYY-MM-DD return "".join(parts) elif len(parts[2]) == 4: # DD-MM-YYYY return parts[2] + parts[1] + parts[0] return dob.replace("-", "") def _gender_code(g: str) -> str: """Return single-letter gender code: M / F / U.""" if not g: return "" g = g.strip().upper() if g in ("M", "MALE"): return "M" if g in ("F", "FEMALE"): return "F" return "U" def _calculate_age(dob_str: str) -> Optional[int]: """Calculate age from YYYYMMDD string.""" if not dob_str: return None normalized = _normalize_dob(dob_str) if len(normalized) != 8 or not normalized.isdigit(): return None try: birth_year = int(normalized[:4]) birth_month = int(normalized[4:6]) birth_day = int(normalized[6:]) today = datetime.now() return today.year - birth_year - ((today.month, today.day) < (birth_month, birth_day)) except Exception: return None @router.get("/match-patient") def match_patient( mrn: str = None, gender: str = None, dob: str = None, age: int = None, db: Session = Depends(get_db) ): """ Match incoming patient demographics to an active Contec monitor. Uses 4-point verification: MRN, Gender, DOB, Age. Call this from the dashboard iframe URL: /api/dashboard?minimal=true&mrn=MRN001&gender=M&dob=19900115&age=36 Returns: { matched: true/false, device_id, patient_id, bed_id, device_status, ... } """ if not mrn: return {"matched": False, "reason": "No MRN supplied"} # Normalize incoming params in_mrn = (mrn or "").strip().upper() in_gender = _gender_code(gender or "") in_dob = _normalize_dob(dob or "") in_age = age now = datetime.now(timezone.utc) # Fetch all patients from monitor database (populated ONLY by monitor HL7/push-vitals) # Do NOT auto-register — patients must exist from actual monitor data all_patients = db.query(Patient).all() best_match = None for p in all_patients: # --- MRN check (must match exactly) --- p_mrn = (p.mrn or p.patient_id or "").strip().upper() if p_mrn != in_mrn: continue # --- Gender check (must match exactly) --- p_gender = _gender_code(p.gender or "") if p_gender != in_gender: continue # --- DOB check (must match exactly) --- p_dob = _normalize_dob(p.dob or "") if p_dob != in_dob: continue # --- Age check (must match exactly) --- if in_age is not None: p_age = _calculate_age(p.dob or "") if p_age is None or p_age != in_age: continue # If MRN, gender, DOB, and Age all match, we have our matched patient best_match = p break if not best_match: return { "matched": False, "reason": "Patient verification failed. MRN, Gender, DOB, and Age must match exactly." } matched_patient = best_match # Find the most recent vital reading for this patient latest = ( db.query(VitalReading) .filter(VitalReading.patient_id == matched_patient.patient_id) .order_by(VitalReading.timestamp.desc()) .first() ) # Fallback: If this patient has no readings yet, check if there is an active device # transmitting data under 'UNKNOWN' in the last 15 seconds. This links binary monitors to the verified patient. if not latest: active_unknown_reading = ( db.query(VitalReading) .filter(VitalReading.patient_id == "UNKNOWN") .order_by(VitalReading.timestamp.desc()) .first() ) if active_unknown_reading: reading_time = active_unknown_reading.timestamp if reading_time.tzinfo is None: reading_time = reading_time.replace(tzinfo=timezone.utc) if (now - reading_time).total_seconds() < 15.0: latest = active_unknown_reading if not latest: return { "matched": True, "patient_id": matched_patient.patient_id, "device_id": None, "bed_id": matched_patient.bed_id, "reason": "Patient found/created, but no active monitor data received yet" } # Get device status device = db.query(Device).filter(Device.device_id == latest.device_id).first() device_status = "unknown" if device: last_seen = device.last_seen if last_seen.tzinfo is None: last_seen = last_seen.replace(tzinfo=timezone.utc) device_status = "active" if (now - last_seen).total_seconds() < 15 else "offline" return { "matched": True, "patient_id": matched_patient.patient_id, "patient_name": matched_patient.name, "patient_gender": matched_patient.gender, "patient_dob": matched_patient.dob, "device_id": latest.device_id, "bed_id": matched_patient.bed_id, "device_status": device_status, "last_reading_at": latest.timestamp.isoformat() if latest.timestamp else None, "match_score": 100, }