import asyncio import json import logging import socket from typing import List from pydantic import BaseModel from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from sqlalchemy import or_ from database import get_db from models import Device, Patient, VitalReading, TransmissionLog logger = logging.getLogger(__name__) router = APIRouter() class PatientAssignRequest(BaseModel): device_id: str mrn: str name: Optional[str] = None dob: Optional[str] = None gender: Optional[str] = None bed_id: Optional[str] = None # === 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="
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(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() return { "total_devices": total_devices, "active_devices": active_devices, "total_readings_stored": total_readings, "pending_transmissions": pending_transmissions } @router.get("/devices") def list_devices(db: Session = Depends(get_db)): devices = db.query(Device).all() return devices @router.get("/latest-readings") def get_latest_readings( limit: int = 10, mrn: Optional[str] = None, name: Optional[str] = None, dob: Optional[str] = None, device_id: Optional[str] = None, db: Session = Depends(get_db) ): query = db.query(VitalReading) if device_id: query = query.filter(VitalReading.device_id == device_id) patient_filters = [] if mrn: mrn_clean = mrn.strip() patient_filters.append(Patient.mrn == mrn_clean) patient_filters.append(Patient.patient_id == mrn_clean) if name: name_clean = name.strip() patient_filters.append(Patient.name.ilike(f"%{name_clean}%")) if dob: dob_clean = dob.strip() patient_filters.append(Patient.dob.ilike(f"%{dob_clean}%")) if patient_filters: matching_patients = db.query(Patient).filter(or_(*patient_filters)).all() matching_dev_ids = [p.device_id for p in matching_patients if p.device_id] matching_mrns = [p.mrn for p in matching_patients if p.mrn] + [p.patient_id for p in matching_patients if p.patient_id] conds = [] if mrn: conds.append(VitalReading.patient_id == mrn.strip()) if matching_mrns: conds.append(VitalReading.patient_id.in_(matching_mrns)) if matching_dev_ids: conds.append(VitalReading.device_id.in_(matching_dev_ids)) if conds: query = query.filter(or_(*conds)) readings = query.order_by(VitalReading.timestamp.desc()).limit(limit).all() return readings @router.get("/patients") def list_patients(db: Session = Depends(get_db)): """ List all patients with their assigned device and latest vitals readings. """ patients = db.query(Patient).filter(Patient.patient_id != "UNKNOWN", Patient.mrn != "UNKNOWN").all() results = [] for p in patients: # Find latest vitals for this patient or this patient's assigned device latest_reading = ( db.query(VitalReading) .filter( or_( VitalReading.patient_id == p.mrn, VitalReading.patient_id == p.patient_id, (VitalReading.device_id == p.device_id) if p.device_id else False ) ) .order_by(VitalReading.timestamp.desc()) .first() ) device = db.query(Device).filter(Device.device_id == p.device_id).first() if p.device_id else None results.append({ "id": p.id, "patient_id": p.patient_id, "mrn": p.mrn or p.patient_id, "name": p.name or "Unknown", "gender": p.gender, "dob": p.dob, "bed_id": p.bed_id, "device_id": p.device_id, "device_status": device.status if device else "unassigned", "device_ip": device.ip_address if device else None, "latest_vitals": { "heart_rate": latest_reading.heart_rate if latest_reading else None, "spo2": latest_reading.spo2 if latest_reading else None, "systolic_bp": latest_reading.systolic_bp if latest_reading else None, "diastolic_bp": latest_reading.diastolic_bp if latest_reading else None, "map_bp": latest_reading.map_bp if latest_reading else None, "respiratory_rate": latest_reading.respiratory_rate if latest_reading else None, "temperature": latest_reading.temperature if latest_reading else None, "timestamp": latest_reading.timestamp.isoformat() if (latest_reading and latest_reading.timestamp) else None, } if latest_reading else None }) return results @router.post("/patient/assign") @router.post("/patients/assign") async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db)): """ Assign patient demographics (MRN, Name, DOB, Gender, Bed) to a specific monitor/device_id. """ from contec_server import set_assigned_patient device_id = req.device_id.strip() mrn = req.mrn.strip() if req.mrn else "" if not device_id: raise HTTPException(status_code=400, detail="device_id is required") if not mrn or mrn.upper() in ("UNKNOWN", "CLEAR", "NONE", "UNASSIGN"): # Clear assignment from in-memory cache and DB from contec_server import device_patient_map, device_cache if device_id in device_patient_map: del device_patient_map[device_id] if device_id in device_cache: cached = device_cache[device_id] cached.patient_id = "UNKNOWN" cached.patient_name = None cached.patient_gender = None cached.patient_dob = None cached.bed_id = None db.query(Patient).filter(Patient.device_id == device_id).update({"device_id": None}) db.commit() try: await ws_manager.broadcast_vitals({ "event": "patient_assigned", "device_id": device_id, "patient_id": "UNKNOWN", "mrn": "UNKNOWN", "patient_name": None, "name": None, "patient_gender": None, "gender": None, "patient_dob": None, "dob": None, "bed_id": None }) except Exception: pass return { "status": "success", "message": f"Device {device_id} unassigned successfully" } # 1. Clear device_id from any other patient previously holding this device existing_for_device = db.query(Patient).filter(Patient.device_id == device_id).all() for p in existing_for_device: if p.mrn != mrn and p.patient_id != mrn: p.device_id = None # 2. Find or create patient by MRN patient = db.query(Patient).filter(or_(Patient.mrn == mrn, Patient.patient_id == mrn)).first() if not patient: patient = Patient( patient_id=mrn, mrn=mrn, name=req.name or "Unknown", gender=req.gender, dob=req.dob, bed_id=req.bed_id, device_id=device_id ) db.add(patient) else: patient.patient_id = mrn patient.mrn = mrn if req.name is not None: patient.name = req.name if req.gender is not None: patient.gender = req.gender if req.dob is not None: patient.dob = req.dob if req.bed_id is not None: patient.bed_id = req.bed_id patient.device_id = device_id # 3. Associate recent readings for this device that were marked "UNKNOWN" db.query(VitalReading).filter( VitalReading.device_id == device_id, VitalReading.patient_id == "UNKNOWN" ).update({"patient_id": mrn}) db.commit() db.refresh(patient) # 4. Update in-memory server mapping p_data = { "patient_id": patient.mrn, "mrn": patient.mrn, "name": patient.name, "gender": patient.gender, "dob": patient.dob, "bed_id": patient.bed_id, } set_assigned_patient(device_id, p_data) # 5. Broadcast to connected WebSocket clients try: await ws_manager.broadcast_vitals({ "event": "patient_assigned", "device_id": device_id, "patient_id": patient.mrn, "mrn": patient.mrn, "patient_name": patient.name, "name": patient.name, "patient_gender": patient.gender, "gender": patient.gender, "patient_dob": patient.dob, "dob": patient.dob, "bed_id": patient.bed_id, }) except Exception as e: logger.debug(f"Broadcast of patient_assigned failed: {e}") return { "status": "success", "message": f"Patient {patient.name} ({mrn}) assigned to device {device_id}", "patient": p_data } @router.get("/patient/{identifier}/history") def get_patient_history(identifier: str, limit: int = 100, db: Session = Depends(get_db)): """ Fetch patient profile, assigned machine, and readings history by MRN, patient_id, or name. """ patient = db.query(Patient).filter( or_( Patient.mrn == identifier, Patient.patient_id == identifier, Patient.name.ilike(f"%{identifier}%") ) ).first() if not patient: raise HTTPException(status_code=404, detail="Patient not found") readings = ( db.query(VitalReading) .filter(or_(VitalReading.patient_id == patient.mrn, VitalReading.patient_id == patient.patient_id)) .order_by(VitalReading.timestamp.desc()) .limit(limit) .all() ) device = db.query(Device).filter(Device.device_id == patient.device_id).first() if patient.device_id else None return { "patient": { "mrn": patient.mrn or patient.patient_id, "name": patient.name, "gender": patient.gender, "dob": patient.dob, "bed_id": patient.bed_id, "device_id": patient.device_id, }, "device": { "device_id": device.device_id if device else None, "ip_address": device.ip_address if device else None, "status": device.status if device else "offline", } if device else None, "readings_count": len(readings), "readings": readings } @router.get("/readings") def get_readings( device_id: Optional[str] = None, patient_id: Optional[str] = None, mrn: Optional[str] = None, name: Optional[str] = None, gender: Optional[str] = None, dob: Optional[str] = None, limit: int = 50, offset: int = 0, db: Session = Depends(get_db) ): """ Fetch vital readings filtered by patient MRN, Name, DOB, Gender, or Device ID. """ query = db.query(VitalReading) target_mrn = mrn or patient_id if target_mrn: query = query.filter(VitalReading.patient_id == target_mrn) # If filtered by patient demographics (name, gender, dob) if name or gender or dob: p_query = db.query(Patient) if name: p_query = p_query.filter(Patient.name.ilike(f"%{name}%")) if gender: p_query = p_query.filter(Patient.gender.ilike(f"%{gender}%")) if dob: p_query = p_query.filter(Patient.dob.ilike(f"%{dob}%")) matching_patients = p_query.all() matching_mrns = [p.mrn for p in matching_patients if p.mrn] + [p.patient_id for p in matching_patients if p.patient_id] matching_devices = [p.device_id for p in matching_patients if p.device_id] query = query.filter( or_( VitalReading.patient_id.in_(matching_mrns), VitalReading.device_id.in_(matching_devices) ) ) if device_id: query = query.filter(VitalReading.device_id == device_id) return query.order_by(VitalReading.timestamp.desc()).offset(offset).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 datetime import datetime, timezone @router.post("/push-vitals") async def push_vitals(vitals: NormalizedVitals, api_key: Optional[str] = None, db: Session = Depends(get_db)): """ Receive vitals pushed from a local instance and store + broadcast them to connected WebSocket clients. """ from config import settings if settings.api_token and api_key != settings.api_token: raise HTTPException(status_code=401, detail="Invalid or missing api_key") try: now = datetime.now(timezone.utc) # 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", last_seen=now ) db.add(device) else: if vitals.ip_address: device.ip_address = vitals.ip_address device.status = "active" device.last_seen = now # 2. Patient tracking mrn = vitals.patient_id patient = None if mrn and mrn != "UNKNOWN": patient = db.query(Patient).filter(or_(Patient.mrn == mrn, Patient.patient_id == mrn)).first() if not patient: patient = Patient( patient_id=mrn, mrn=mrn, name=vitals.patient_name or "Unknown", gender=vitals.patient_gender, dob=vitals.patient_dob, bed_id=vitals.bed_id, device_id=vitals.device_id ) db.add(patient) else: if vitals.patient_name and vitals.patient_name != "Unknown": 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 patient.device_id = vitals.device_id elif vitals.device_id: # Check if there's already a patient assigned to this device patient = db.query(Patient).filter(Patient.device_id == vitals.device_id, Patient.patient_id != "UNKNOWN").first() if patient: mrn = patient.mrn or patient.patient_id # 3. Create VitalReading reading = VitalReading( device_id=vitals.device_id, patient_id=mrn or "UNKNOWN", timestamp=vitals.timestamp or now, 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 ) db.add(reading) db.commit() db.refresh(reading) # 4. Broadcast to WebSocket clients vitals_dict = vitals.model_dump(mode="json") vitals_dict["mrn"] = mrn or (patient.mrn if patient else None) vitals_dict["patient_id"] = vitals_dict["mrn"] or vitals.patient_id vitals_dict["name"] = vitals.patient_name or (patient.name if patient else None) vitals_dict["patient_name"] = vitals_dict["name"] vitals_dict["gender"] = vitals.patient_gender or (patient.gender if patient else None) vitals_dict["patient_gender"] = vitals_dict["gender"] vitals_dict["dob"] = vitals.patient_dob or (patient.dob if patient else None) vitals_dict["patient_dob"] = vitals_dict["dob"] vitals_dict["bed_id"] = vitals.bed_id or (patient.bed_id if patient else None) await ws_manager.broadcast_vitals(vitals_dict) return { "status": "ok", "reading_id": reading.id, "device_id": vitals.device_id, "patient_id": vitals_dict["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)}")