Update Contec monitor service, parser, dashboard and tunneling

This commit is contained in:
2026-09-12 19:56:21 +05:30
parent 2011544f53
commit 9d7c370836
72 changed files with 3966 additions and 10240 deletions
+314 -369
View File
@@ -2,18 +2,29 @@ import asyncio
import json
import logging
import socket
from typing import List, Optional
from datetime import datetime, timezone
from typing import List
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect, Request
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, VitalReading, TransmissionLog
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."""
@@ -70,61 +81,333 @@ 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)):
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()
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
"pending_transmissions": pending_transmissions
}
@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)):
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: str = None, patient_id: str = None, limit: int = 50, db: Session = Depends(get_db)):
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)
if patient_id:
query = query.filter(VitalReading.patient_id == patient_id)
return query.order_by(VitalReading.timestamp.desc()).limit(limit).all()
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)):
@@ -185,341 +468,3 @@ def get_network_info():
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.post("/register-patient")
def register_patient(
mrn: str,
gender: str,
dob: str,
name: str = "Unknown",
bed_id: str = None,
device_id: str = None,
db: Session = Depends(get_db)
):
"""
Register or update patient demographics in the monitor database.
Call this to seed patient data when the CMS monitor uses binary protocol
(which does not transmit demographics).
Example:
POST /api/register-patient?mrn=09091102&gender=Male&dob=1960-06-22&name=Shahil
"""
if not mrn:
raise HTTPException(status_code=400, detail="MRN is required")
clean_mrn = mrn.strip().upper()
# Check if patient already exists
patient = db.query(Patient).filter(
(Patient.mrn == clean_mrn) | (Patient.patient_id == clean_mrn)
).first()
if patient:
# Update existing record
patient.mrn = clean_mrn
patient.name = name or patient.name
patient.gender = gender or patient.gender
patient.dob = dob or patient.dob
if bed_id:
patient.bed_id = bed_id
else:
# Create new patient record
patient = Patient(
patient_id=clean_mrn,
mrn=clean_mrn,
name=name,
gender=gender,
dob=dob,
bed_id=bed_id,
)
db.add(patient)
# If device_id is provided, also create a vital reading to link patient to device
if device_id:
device = db.query(Device).filter(Device.device_id == device_id).first()
if device:
reading = VitalReading(
device_id=device_id,
patient_id=clean_mrn,
timestamp=datetime.now(timezone.utc),
transmitted=True,
)
db.add(reading)
db.commit()
db.refresh(patient)
return {
"status": "ok",
"patient_id": patient.patient_id,
"mrn": patient.mrn,
"name": patient.name,
"gender": patient.gender,
"dob": patient.dob,
"bed_id": patient.bed_id,
}
@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 or auto-register patient based on MRN only (essential to survive Render cloud SQLite wipeouts)
matched_patient = db.query(Patient).filter(
(Patient.mrn == in_mrn) | (Patient.patient_id == in_mrn)
).first()
if not matched_patient:
matched_patient = Patient(
patient_id=in_mrn,
mrn=in_mrn,
name="Shahil Kumar",
gender=gender or "Male",
dob=dob or "1960-06-22",
)
db.add(matched_patient)
db.commit()
db.refresh(matched_patient)
# 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,
}