Compare commits

...
17 Commits
Author SHA1 Message Date
rkcsh ce16210525 feat: automatically forward merged non-null vitals (HR, SpO2, Temp, BP) during cloud sync 2026-09-18 15:59:16 +05:30
rkcsh 1da4595666 clean: remove references to /home/prathiyuman paths and remove nested folder 2026-09-18 15:26:17 +05:30
rkcsh ace6ad1dbd fix: auto-detect working directory, install venv, configure ufw firewall, and sync Option B vitals in setup_service.sh 2026-09-18 15:21:23 +05:30
rkcsh f57e8b9916 fix: synchronize all latest dashboard, Option B, and API updates to fiveparaminte-main for Render deployment 2026-09-18 14:56:33 +05:30
rkcsh 26819ac893 fix: support both SQLite and PostgreSQL without check_same_thread crash 2026-09-18 14:14:32 +05:30
rkcsh a8075da9ee feat: retain last recorded patient vitals on dashboard (Option B) 2026-09-18 12:00:50 +05:30
rkcsh 1ebbd0e68a fix: timezone-aware timestamp parsing, favicon 404, and performance lag 2026-09-18 11:40:14 +05:30
rkcsh b35d196ba6 add real-time patient vital signs monitoring dashboard UI and backend integration 2026-09-18 10:37:49 +05:30
rkcsh 2342346d39 Strictly eliminate fake and stale vitals and enforce machine-to-MRN isolation 2026-09-17 19:36:21 +05:30
rkcsh 37445d2ae9 fix: reliable cloud sync and connection pooling for Render live dashboard 2026-09-17 16:41:52 +05:30
rkcsh 68bda2519f reorganize application code into fiveparaminte-main directory structure 2026-09-17 14:30:02 +05:30
rkcsh 1b1b382f3e fix: sync contec oscilloscope machine UI, dual temp T1/T2, and cache-control to fiveparaminte-main for Render deployment 2026-09-15 13:54:06 +05:30
rkcsh a0ab3c073e add dashboard monitoring interface and dual temperature tracking 2026-09-15 11:43:54 +05:30
rkcsh 60da868a3a feat: multi-point patient verification (MRN, Name, DOB) via URL parameters 2026-09-15 10:07:14 +05:30
rkcsh 5773656b01 feat: add /api/push-vitals endpoint for local-to-cloud live telemetry sync 2026-09-15 09:56:28 +05:30
rkcsh 4dc1b8e720 fix: sync all files to fiveparaminte-main including routers and static dirs 2026-09-15 09:43:23 +05:30
rkcsh 48a9288e93 deployed problem 2026-09-15 09:20:04 +05:30
27 changed files with 9034 additions and 759 deletions
+1
View File
@@ -1,6 +1,7 @@
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONPATH=/app
# Install dependencies
COPY requirements.txt .
+53 -14
View File
@@ -1,5 +1,7 @@
import httpx
import logging
import asyncio
import os
from typing import Optional
from schemas import NormalizedVitals
from config import settings
@@ -12,18 +14,17 @@ def get_http_client() -> httpx.AsyncClient:
global _http_client
if _http_client is None or _http_client.is_closed:
_http_client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=3.0, read=5.0, write=3.0, pool=3.0),
limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
timeout=httpx.Timeout(connect=5.0, read=5.0, write=5.0, pool=10.0),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=50)
)
return _http_client
async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
"""
Attempts to POST the normalized vitals to the configured target API
and to the Render cloud database.
and to the Render cloud database concurrently.
Returns True if at least one forward succeeds.
"""
import os
headers = {"Content-Type": "application/json"}
# Strip heavy wave sample arrays before REST transmission (cloud APIs only ingest vital signs)
@@ -31,12 +32,16 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
payload.pop("ecg_wave", None)
payload.pop("pleth_wave", None)
if not payload.get("patient_id"):
payload["patient_id"] = "UNKNOWN"
client = get_http_client()
# 1. Forward to primary target API (e.g. Injomo Dev Workflow)
async def _forward_primary() -> bool:
primary_url = settings.target_api_url
primary_success = False
if primary_url:
if not primary_url:
return False
try:
primary_headers = headers.copy()
if settings.api_token:
@@ -48,19 +53,18 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}")
primary_success = True
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to primary API: {type(e).__name__} ({err_msg})")
return False
# 2. Forward to Render Cloud Server
async def _forward_render() -> bool:
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
render_success = False
# Detect if we are already running on Render to prevent an infinite loop
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not is_on_render:
if is_on_render:
return True
try:
url_with_key = render_url
if settings.api_token:
@@ -73,9 +77,44 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}")
render_success = True
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to Render Cloud: {type(e).__name__} ({err_msg})")
return False
return primary_success or render_success
results = await asyncio.gather(_forward_primary(), _forward_render(), return_exceptions=True)
return any(r is True for r in results if not isinstance(r, Exception))
async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -> bool:
"""
Synchronizes patient demographics assignment to Render Cloud Server.
Ensures local and remote dashboards always show the exact same patient.
"""
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
client = get_http_client()
render_assign_url = "https://contectfiveparamonitor.onrender.com/api/patient/assign"
payload = {
"device_id": device_id,
"mrn": patient_data.get("mrn") or patient_data.get("patient_id") or "",
"name": patient_data.get("name") or patient_data.get("patient_name") or "Unknown",
"dob": patient_data.get("dob") or patient_data.get("patient_dob"),
"gender": patient_data.get("gender") or patient_data.get("patient_gender"),
"bed_id": patient_data.get("bed_id"),
"patient_type": patient_data.get("patient_type")
}
try:
url_with_key = render_assign_url
if settings.api_token:
url_with_key = f"{render_assign_url}?api_key={settings.api_token}"
resp = await client.post(url_with_key, json=payload, headers={"Content-Type": "application/json"})
resp.raise_for_status()
logger.info(f"Successfully synced patient {payload['mrn']} ({payload['name']}) to Render Cloud for {device_id}")
return True
except Exception as e:
logger.warning(f"Could not sync patient assignment to Render Cloud: {type(e).__name__} ({e})")
return False
+88
View File
@@ -79,3 +79,91 @@ async def retry_failed_transmissions():
logger.error(f"Error in retry background task: {e}")
await asyncio.sleep(settings.retry_interval_seconds)
async def sync_cloud_state():
"""
Periodically syncs all assigned patients and their latest non-null vitals to Render Cloud.
Ensures that whenever Render restarts or deploys with an empty DB, the cloud dashboard
is immediately populated with the same data as localhost.
"""
import os
if os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", ""):
return # Do not run sync loop if we are running inside Render
from models import Patient
from api_client import sync_patient_assignment_to_cloud, forward_vitals_to_api
from datetime import datetime, timezone
from sqlalchemy import or_
logger.info("Starting background task for periodic cloud state sync to Render...")
await asyncio.sleep(5)
while True:
try:
db = SessionLocal()
try:
patients = db.query(Patient).filter(
Patient.patient_id != "UNKNOWN",
Patient.mrn != "UNKNOWN"
).all()
for p in patients:
if not p.device_id:
continue
# 1. Sync patient demographics assignment to Render
p_data = {
"mrn": p.mrn or p.patient_id,
"name": p.name,
"gender": p.gender,
"dob": p.dob,
"bed_id": p.bed_id,
"patient_type": getattr(p, "patient_type", None)
}
await sync_patient_assignment_to_cloud(p_data, p.device_id)
# 2. Forward latest non-null vitals to Render (Option B)
from sqlalchemy import or_
from datetime import datetime, timezone
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
VitalReading.device_id == p.device_id
)
hr_r = db.query(VitalReading.heart_rate).filter(patient_filters, VitalReading.heart_rate != None).order_by(VitalReading.timestamp.desc()).first()
spo2_r = db.query(VitalReading.spo2).filter(patient_filters, VitalReading.spo2 != None).order_by(VitalReading.timestamp.desc()).first()
bp_r = db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp).filter(patient_filters, VitalReading.systolic_bp != None).order_by(VitalReading.timestamp.desc()).first()
temp_r = db.query(VitalReading.temperature).filter(patient_filters, VitalReading.temperature != None).order_by(VitalReading.timestamp.desc()).first()
temp2_r = db.query(VitalReading.temperature2).filter(patient_filters, VitalReading.temperature2 != None).order_by(VitalReading.timestamp.desc()).first()
temp_diff_r = db.query(VitalReading.temperature_diff).filter(patient_filters, VitalReading.temperature_diff != None).order_by(VitalReading.timestamp.desc()).first()
rr_r = db.query(VitalReading.respiratory_rate).filter(patient_filters, VitalReading.respiratory_rate != None).order_by(VitalReading.timestamp.desc()).first()
if hr_r or spo2_r or temp_r or bp_r:
merged = NormalizedVitals(
device_id=p.device_id,
patient_id=p.mrn or p.patient_id,
patient_name=p.name,
patient_gender=p.gender,
patient_dob=p.dob,
bed_id=p.bed_id,
patient_type=getattr(p, "patient_type", None),
timestamp=datetime.now(timezone.utc),
heart_rate=hr_r[0] if hr_r else None,
spo2=spo2_r[0] if spo2_r else None,
systolic_bp=bp_r[0] if bp_r else None,
diastolic_bp=bp_r[1] if bp_r else None,
map_bp=bp_r[2] if bp_r else None,
temperature=temp_r[0] if temp_r else None,
temperature2=temp2_r[0] if temp2_r else None,
temperature_diff=temp_diff_r[0] if temp_diff_r else None,
respiratory_rate=rr_r[0] if rr_r else None
)
await forward_vitals_to_api(merged)
finally:
db.close()
except Exception as e:
logger.error(f"Error in sync_cloud_state background task: {e}")
await asyncio.sleep(15)
+44 -54
View File
@@ -58,13 +58,14 @@ CONTEC_PARAM_MAP = {
# Temperature
"TEMP": "temperature",
"TEMP1": "temperature",
"TEMP2": "temperature",
"TEMP2": "temperature2",
"T1": "temperature",
"T2": "temperature",
"T2": "temperature2",
"BODY_TEMP": "temperature",
"TEMP-1": "temperature",
"TEMP-2": "temperature",
"TEMP-2": "temperature2",
"BODYTEMP": "temperature",
"TD": "temperature_diff",
}
# Sentinel value used by Contec to mark an invalid/disconnected parameter
@@ -164,6 +165,10 @@ def parse_contec_hl7_text(raw_text: str, client_ip: str) -> Optional[NormalizedV
if len(fields) > 8 and fields[8]:
vitals_dict["patient_gender"] = fields[8].strip().upper()
# PID-35: Patient Category (ADULT / PEDIATRIC / NEONATAL)
if len(fields) > 35 and fields[35]:
vitals_dict["patient_type"] = fields[35].strip()
elif seg_name == "PV1":
# PV1-3: Assigned Patient Location (Room^Bed^...)
if len(fields) > 3 and fields[3]:
@@ -173,6 +178,10 @@ def parse_contec_hl7_text(raw_text: str, client_ip: str) -> Optional[NormalizedV
if bed:
vitals_dict["bed_id"] = bed
# PV1-18: Patient Type
if len(fields) > 18 and fields[18]:
vitals_dict["patient_type"] = fields[18].strip()
elif seg_name == "OBX":
if len(fields) > 5:
obs_val = fields[5].strip()
@@ -278,6 +287,7 @@ def _parse_45_byte_packet(data: bytes, vitals_dict: dict) -> bool:
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
logger.debug("DEBUG_NIBP_PKT len=%d nibp_hex=%s", len(data), data[8:24].hex())
nibp_sys = read_u16_le(data, 15)
nibp_dia = read_u16_le(data, 17)
nibp_map = read_u16_le(data, 19)
@@ -304,7 +314,7 @@ def _parse_45_byte_packet(data: bytes, vitals_dict: dict) -> bool:
def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 56-byte packet (Subtype 23) — contains float32 Temperature values.
Parse 56-byte packet (Subtype 23) — contains float32 Temperature values for T1 and T2.
Verified layout:
[8-11] Temperature 1 (float32)
@@ -318,8 +328,6 @@ def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
found = False
temp1 = read_f32_le(data, 8)
temp2 = read_f32_le(data, 12)
@@ -327,24 +335,24 @@ def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
# Disconnected sentinel is 9999.0 or 3276.7 (0x7FFF)
return t is not None and is_valid_f32(t) and abs(t - 3276.7) > 1.0 and 20.0 < t < 45.0
if is_real_temp(temp1):
vitals_dict["temperature"] = round(temp1, 1)
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
elif is_real_temp(temp2):
vitals_dict["temperature"] = round(temp2, 1)
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
elif temp1 is not None or temp2 is not None:
# Sensor disconnected / off
vitals_dict["temperature"] = None
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
t1_val = round(temp1, 1) if is_real_temp(temp1) else None
t2_val = round(temp2, 1) if is_real_temp(temp2) else None
return found
vitals_dict["temperature"] = t1_val
vitals_dict["temperature2"] = t2_val
if t1_val is not None and t2_val is not None:
vitals_dict["temperature_diff"] = round(abs(t1_val - t2_val), 1)
else:
vitals_dict["temperature_diff"] = None
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
if "temperature2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature2")
if "temperature_diff" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature_diff")
return (t1_val is not None or t2_val is not None)
def _parse_spo2_waveform_packet(data: bytes, vitals_dict: dict) -> bool:
@@ -364,42 +372,26 @@ def _parse_spo2_waveform_packet(data: bytes, vitals_dict: dict) -> bool:
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
found = False
# Check SpO2 at standard offset 264
spo2_val = read_u16_le(data, 264)
# If 264 is disconnected sentinel, probe is NOT connected / off-finger
if spo2_val in (None, 65535, 32767, 255, 9999, 0):
vitals_dict["spo2"] = None
vitals_dict["heart_rate"] = None
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
return True
if 50 <= spo2_val <= 100:
if spo2_val is not None and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
found = True
# ECG HR is at offset 266 in the 286-byte SpO2 waveform packet footer.
# Offset 268 is the SpO2 HIGH ALARM LIMIT (always 100) — NOT pulse rate!
# So we read ECG HR at 266 only.
# Pulse Rate / ECG HR is at offset 266 in the footer
ecg_hr = read_u16_le(data, 266)
logger.info(
f"SpO2 pkt: SpO2[264]={spo2_val} ECG_HR[266]={ecg_hr} alarm_hi[268]={read_u16_le(data,268)} alarm_lo[270]={read_u16_le(data,270) if len(data)>=272 else 'N/A'}"
)
if ecg_hr is not None and ecg_hr not in (9999, 65535, 32767, 0) and 20 <= ecg_hr <= 300:
vitals_dict["heart_rate"] = float(ecg_hr)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
return found
logger.info(
f"SpO2 pkt: SpO2[264]={spo2_val} ECG_HR[266]={ecg_hr} alarm_hi[268]={read_u16_le(data,268)} alarm_lo[270]={read_u16_le(data,270) if len(data)>=272 else 'N/A'}"
)
return True
def _parse_286_byte_packet(data: bytes, vitals_dict: dict) -> bool:
@@ -435,13 +427,13 @@ def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
if "ecg_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("ecg_wave")
logger.debug("DEBUG_ECG_PKT len=%d tail_hex=%s", len(data), data[890:989].hex())
hr_val = read_u16_le(data, 904)
if hr_val is not None and hr_val not in (65535, 9999, 0) and 20 <= hr_val <= 300:
# Only use ECG HR if SpO2-probe PR has NOT already set heart_rate.
# SpO2 PR (from the pleth packet) matches the monitor display and wins.
if "heart_rate" not in vitals_dict.get("present_fields", []):
if hr_val is not None and hr_val not in (65535, 32767, 9999, 0) and 20 <= hr_val <= 300:
if vitals_dict.get("heart_rate") is None:
vitals_dict["heart_rate"] = float(hr_val)
vitals_dict.setdefault("present_fields", []).append("heart_rate")
if "heart_rate" not in vitals_dict.setdefault("present_fields", []):
vitals_dict["present_fields"].append("heart_rate")
return True
@@ -479,9 +471,7 @@ def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[Normal
elif pkt_len == 989:
if _parse_989_byte_packet(pkt_data, vitals_dict):
found_any = True
elif 100 < pkt_len < 600:
if _parse_spo2_waveform_packet(pkt_data, vitals_dict):
found_any = True
# Handled specifically by known packet sizes above
i += max(pkt_len, 4)
else:
+120 -31
View File
@@ -58,25 +58,58 @@ TELEMETRY_WORDS = {
}
def is_telemetry_word(s_lower: str) -> bool:
"""Check if string matches hardware telemetry or alarm words."""
if s_lower in TELEMETRY_WORDS:
return True
tokens = set(re.findall(r'[a-zA-Z0-9]+', s_lower))
alarm_tokens = {
'sensor', 'alarm', 'lead', 'leads', 'searching', 'pulse', 'cuff', 'probe',
'off', 'fall', 'leak', 'motion', 'weak', 'arr', 'asystole', 'vfib', 'vtac',
'pvc', 'brady', 'tachy', 'error', 'ecg', 'spo2', 'nibp', 'resp', 'temp',
'mv', 'bpm', 'mmhg'
}
if any(tok in alarm_tokens for tok in tokens):
return True
return False
def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
"""
Extracts real patient demographics from Contec CMS8500 / CMS7000 binary packets.
Extracts real patient demographics from Contec packets.
Only extracts clean, legitimate patient fields — strictly rejects sensor alarms,
waveform noise, and hardware telemetry.
"""
patient_type = None
data_upper = data.upper()
if b'PEDIATRIC' in data_upper or b'PED' in data_upper:
patient_type = 'Pediatric'
elif b'NEONATE' in data_upper or b'NEONATAL' in data_upper or b'NEO' in data_upper:
patient_type = 'Neonatal'
elif b'ADULT' in data_upper or b'ADU' in data_upper:
patient_type = 'Adult'
# Strictly reject binary waveform/measurement packets from generating false patient records.
# Patient MRN/Name extraction requires an authentic HL7 message with PID or MSH.
if b'PID|' not in data and b'MSH|' not in data:
if patient_type:
return {"patient_type": patient_type}
return None
clean = []
for m in re.finditer(rb'[a-zA-Z0-9_\-\.]{2,}', data):
# Match strings of printable alphanumeric and reasonable characters, allowing single spaces
for m in re.finditer(rb'[a-zA-Z0-9_\-\.]{2,}(?:\s+[a-zA-Z0-9_\-\.]{2,})*', data):
s = m.group(0).decode('latin1', errors='ignore').strip()
if not s or len(s) < 2:
continue
s_lower = s.lower()
if any(kw in s_lower for kw in TELEMETRY_WORDS):
if is_telemetry_word(s_lower):
continue
if len(set(s)) == 1:
continue
if set(s).issubset({'.', '-', '_', '/', '\\'}):
continue
if len(s) > 8 and re.match(r'^[A-Za-z]+$', s):
if len(s) > 6 and re.match(r'^[A-Za-z0-9]+$', s):
diffs = [abs(ord(s[i]) - ord(s[i-1])) for i in range(1, len(s))]
if sum(1 for d in diffs if d <= 2) / len(diffs) > 0.6:
continue
@@ -84,7 +117,7 @@ def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
continue
clean.append(s)
if not clean:
if not clean and not patient_type:
return None
name = None
@@ -95,9 +128,8 @@ def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
for s in clean:
s_up = s.upper()
s_low = s.lower()
if s_low in TELEMETRY_WORDS:
continue
# Check gender first
if s_up in ('MALE', 'FEMALE') and not gender:
gender = 'Male' if s_up == 'MALE' else 'Female'
continue
@@ -105,52 +137,75 @@ def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
gender = 'Male' if s_up == 'M' else 'Female'
continue
# Date of birth: 4-digit birth year between 1920 and 2024, or YYYY-MM-DD
if re.match(r'^(19\d\d|20[01]\d|202[0-4])$', s) and not dob:
if is_telemetry_word(s_low):
continue
# Date of birth: 4-digit birth year between 1920 and 2026, or YYYY-MM-DD
if re.match(r'^(19\d\d|20[01]\d|202[0-6])$', s) and not dob:
dob = s
continue
elif re.match(r'^\d{4}[-/]\d{2}[-/]\d{2}$', s) and not dob:
dob = s
continue
# Name candidate: strictly at least 3 letters, has a vowel, not telemetry
if re.search(r'^[A-Za-z\s\.\-]{3,30}$', s) and re.search(r'[aeiouyAEIOUY]', s) and not name:
if s_low not in TELEMETRY_WORDS:
# Name candidate: contains letters and a vowel, length between 3 and 40
if re.search(r'^[A-Za-z\s\.\-]{3,40}$', s) and re.search(r'[aeiouyAEIOUY]', s) and not name:
if not is_telemetry_word(s_low):
name = s
continue
# MRN candidate: numeric or alphanumeric ID, at least 2 chars, not a year, not telemetry
if not mrn and re.match(r'^[A-Za-z0-9\-]{2,15}$', s) and s_low not in TELEMETRY_WORDS:
# MRN candidate: numeric or alphanumeric ID, 2..15 chars, not a birth year
if not mrn and re.match(r'^[A-Za-z0-9\-]{2,15}$', s) and not is_telemetry_word(s_low):
if not (s.isdigit() and 1920 <= int(s) <= 2030):
if s.isdigit() and len(s) > 10:
continue
mrn = s
continue
# Only accept when BOTH real name and real MRN/DOB are present together
if name and (mrn or dob):
# Accept if valid MRN or Name was detected
if mrn or name:
clean_mrn = mrn or name
clean_name = name or f"Patient {clean_mrn}"
return {
"name": name,
"mrn": mrn or name,
"patient_id": mrn or name,
"name": clean_name,
"mrn": clean_mrn,
"patient_id": clean_mrn,
"dob": dob,
"gender": gender,
"patient_type": patient_type,
}
elif patient_type:
return {
"patient_type": patient_type,
}
return None
async def apply_auto_demographics(client_ip: str, demo: dict):
from config import settings
from api_client import sync_patient_assignment_to_cloud
model = settings.device_models.get(client_ip, settings.monitor_model)
device_id = f"{model}_{client_ip}"
name = demo.get("name")
mrn = demo.get("mrn")
dob = demo.get("dob")
gender = demo.get("gender")
patient_type = demo.get("patient_type")
current = device_patient_map.get(device_id, {})
if current.get("name") == name and current.get("mrn") == mrn and current.get("dob") == dob and current.get("gender") == gender:
# If only patient_type updated
if not mrn and not name:
if patient_type and current.get("patient_type") != patient_type:
current["patient_type"] = patient_type
device_patient_map[device_id] = current
return
logger.info(f"*** REAL PATIENT DEMOGRAPHICS DETECTED FROM MACHINE ({client_ip}): Name={name}, MRN={mrn}, DOB={dob}, Gender={gender} ***")
if (current.get("name") == name and current.get("mrn") == mrn and
current.get("dob") == dob and current.get("gender") == gender and
current.get("patient_type") == patient_type):
return
logger.info(f"*** REAL PATIENT DEMOGRAPHICS DETECTED FROM MACHINE ({client_ip}): Name={name}, MRN={mrn}, DOB={dob}, Gender={gender}, Type={patient_type} ***")
bed_id = "1" if "192.168.100.106" in client_ip else "2"
p_data = {
@@ -159,7 +214,8 @@ async def apply_auto_demographics(client_ip: str, demo: dict):
"name": name,
"gender": gender,
"dob": dob,
"bed_id": bed_id
"bed_id": bed_id,
"patient_type": patient_type or current.get("patient_type"),
}
device_patient_map[device_id] = p_data
@@ -218,11 +274,15 @@ async def apply_auto_demographics(client_ip: str, demo: dict):
"gender": gender,
"patient_dob": dob,
"dob": dob,
"bed_id": bed_id
"bed_id": bed_id,
"patient_type": p_data.get("patient_type"),
})
except Exception as e:
logger.debug(f"Failed to broadcast auto patient: {e}")
# Forward to Render Cloud Server so remote is kept in sync
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int):
"""
@@ -251,8 +311,13 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
logger.info(f"Contec client {client_ip}:{client_port} on port {port} sent EOF (0 bytes)")
break
if client_ip != "127.0.0.1":
logger.info(f"TCP READ from {client_ip}: {len(data)} bytes, hex: {data[:24].hex()}")
# Dynamically extract and apply patient demographics sent from the machine
try:
demo = extract_demographics_from_packet(data, client_ip)
if demo:
await apply_auto_demographics(client_ip, demo)
except Exception as ex:
logger.debug(f"Demographics extraction error: {ex}")
buffer += data
@@ -376,6 +441,7 @@ def get_assigned_patient(device_id: str):
"gender": patient.gender,
"dob": patient.dob,
"bed_id": patient.bed_id,
"patient_type": getattr(patient, "patient_type", None),
}
device_patient_map[device_id] = p_data
return p_data
@@ -396,6 +462,7 @@ def set_assigned_patient(device_id: str, patient_data: dict):
cached.patient_gender = patient_data.get("gender")
cached.patient_dob = patient_data.get("dob")
cached.bed_id = patient_data.get("bed_id")
cached.patient_type = patient_data.get("patient_type")
async def process_vitals(vitals):
@@ -416,6 +483,7 @@ async def process_vitals(vitals):
vitals.patient_gender = assigned.get("gender")
vitals.patient_dob = assigned.get("dob")
vitals.bed_id = assigned.get("bed_id")
vitals.patient_type = assigned.get("patient_type")
elif vitals.patient_name and vitals.patient_name != "Unknown":
# Monitor sent fresh demographics via HL7 PID segment!
device_patient_map[device_id] = {
@@ -425,6 +493,7 @@ async def process_vitals(vitals):
"gender": vitals.patient_gender,
"dob": vitals.patient_dob,
"bed_id": vitals.bed_id,
"patient_type": getattr(vitals, "patient_type", None),
}
# Check if incoming packet has fresh waveforms
@@ -453,21 +522,22 @@ async def process_vitals(vitals):
if value is not None:
active_fields.append(field)
# Update cache & record timestamps
# Update cache & record timestamps for present values
for field in active_fields:
val = getattr(vitals, field, None)
setattr(cached, field, val)
if val is not None:
setattr(cached, field, val)
device_field_timestamps[device_id][field] = now
else:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
# Clean up stale fields:
# Live vitals (HR, SpO2, Temp, RR) have a 60-second window because monitor cycles take ~24s
# NIBP (BP, MAP) has a 1-hour window because it is measured intermittently
# Live vitals (HR, SpO2, Temp, RR) have a 15-second window
# NIBP (BP, MAP) has a 15-minute window because it is measured intermittently
for field in list(device_field_timestamps[device_id].keys()):
last_updated = device_field_timestamps[device_id][field]
timeout = 3600.0 if field in ("systolic_bp", "diastolic_bp", "map_bp") else 60.0
timeout = 900.0 if field in ("systolic_bp", "diastolic_bp", "map_bp") else 15.0
if (now - last_updated).total_seconds() > timeout:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
@@ -506,6 +576,7 @@ async def process_vitals(vitals):
vitals_dict["gender"] = vitals.patient_gender
vitals_dict["dob"] = vitals.patient_dob
vitals_dict["bed_id"] = vitals.bed_id
vitals_dict["patient_type"] = getattr(vitals, "patient_type", None)
await ws_manager.broadcast_vitals(vitals_dict)
except Exception as e:
logger.debug(f"WebSocket broadcast failed: {e}")
@@ -569,6 +640,7 @@ async def _persist_and_forward_bg(vitals, should_persist):
gender=vitals.patient_gender,
dob=vitals.patient_dob,
bed_id=vitals.bed_id,
patient_type=getattr(vitals, "patient_type", None),
device_id=vitals.device_id
)
db.add(patient)
@@ -584,12 +656,27 @@ async def _persist_and_forward_bg(vitals, should_persist):
patient.dob = vitals.patient_dob
if vitals.bed_id:
patient.bed_id = vitals.bed_id
if getattr(vitals, "patient_type", None):
patient.patient_type = vitals.patient_type
if vitals.device_id:
patient.device_id = vitals.device_id
if patient:
if not vitals.patient_id or vitals.patient_id == "UNKNOWN":
vitals.patient_id = patient.mrn or patient.patient_id
if not vitals.patient_name or vitals.patient_name == "Unknown":
vitals.patient_name = patient.name
if not vitals.patient_gender and patient.gender:
vitals.patient_gender = patient.gender
if not vitals.patient_dob and patient.dob:
vitals.patient_dob = patient.dob
if not vitals.bed_id and patient.bed_id:
vitals.bed_id = patient.bed_id
if not getattr(vitals, "patient_type", None) and getattr(patient, "patient_type", None):
vitals.patient_type = patient.patient_type
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
patient_id=vitals.patient_id or (patient.mrn if patient else "UNKNOWN"),
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
@@ -598,6 +685,8 @@ async def _persist_and_forward_bg(vitals, should_persist):
map_bp=vitals.map_bp,
respiratory_rate=vitals.respiratory_rate,
temperature=vitals.temperature,
temperature2=vitals.temperature2,
temperature_diff=vitals.temperature_diff,
transmitted=False
)
db.add(reading)
+3 -1
View File
@@ -3,9 +3,11 @@ from sqlalchemy.pool import NullPool
from sqlalchemy.orm import sessionmaker, declarative_base
from config import settings
connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
engine = create_engine(
settings.database_url,
connect_args={"check_same_thread": False},
connect_args=connect_args,
poolclass=NullPool
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+1
View File
@@ -1,6 +1,7 @@
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONPATH=/app
# Install dependencies
COPY requirements.txt .
+75 -18
View File
@@ -1,63 +1,120 @@
import httpx
import logging
import asyncio
import os
from typing import Optional
from schemas import NormalizedVitals
from config import settings
logger = logging.getLogger(__name__)
_http_client: Optional[httpx.AsyncClient] = None
def get_http_client() -> httpx.AsyncClient:
global _http_client
if _http_client is None or _http_client.is_closed:
_http_client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=5.0, read=5.0, write=5.0, pool=10.0),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=50)
)
return _http_client
async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
"""
Attempts to POST the normalized vitals to the configured target API
and to the Render cloud database.
and to the Render cloud database concurrently.
Returns True if at least one forward succeeds.
"""
import os
headers = {"Content-Type": "application/json"}
# Strip heavy wave sample arrays before REST transmission (cloud APIs only ingest vital signs)
payload = vitals.model_dump(mode="json")
payload.pop("ecg_wave", None)
payload.pop("pleth_wave", None)
if not payload.get("patient_id"):
payload["patient_id"] = "UNKNOWN"
client = get_http_client()
# 1. Forward to primary target API (e.g. Injomo Dev Workflow)
async def _forward_primary() -> bool:
primary_url = settings.target_api_url
primary_success = False
if primary_url:
if not primary_url:
return False
try:
primary_headers = headers.copy()
if settings.api_token:
primary_headers["Authorization"] = f"Bearer {settings.api_token}"
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
primary_url,
json=vitals.model_dump(mode="json"),
json=payload,
headers=primary_headers
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}")
primary_success = True
return True
except Exception as e:
logger.error(f"Error forwarding to primary API: {e}")
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to primary API: {type(e).__name__} ({err_msg})")
return False
# 2. Forward to Render Cloud Server
async def _forward_render() -> bool:
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
render_success = False
# Detect if we are already running on Render to prevent an infinite loop
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not is_on_render:
if is_on_render:
return True
try:
url_with_key = render_url
if settings.api_token:
url_with_key = f"{render_url}?api_key={settings.api_token}"
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
url_with_key,
json=vitals.model_dump(mode="json"),
json=payload,
headers=headers
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}")
render_success = True
return True
except Exception as e:
logger.error(f"Error forwarding to Render Cloud: {e}")
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to Render Cloud: {type(e).__name__} ({err_msg})")
return False
return primary_success or render_success
results = await asyncio.gather(_forward_primary(), _forward_render(), return_exceptions=True)
return any(r is True for r in results if not isinstance(r, Exception))
async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -> bool:
"""
Synchronizes patient demographics assignment to Render Cloud Server.
Ensures local and remote dashboards always show the exact same patient.
"""
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
client = get_http_client()
render_assign_url = "https://contectfiveparamonitor.onrender.com/api/patient/assign"
payload = {
"device_id": device_id,
"mrn": patient_data.get("mrn") or patient_data.get("patient_id") or "",
"name": patient_data.get("name") or patient_data.get("patient_name") or "Unknown",
"dob": patient_data.get("dob") or patient_data.get("patient_dob"),
"gender": patient_data.get("gender") or patient_data.get("patient_gender"),
"bed_id": patient_data.get("bed_id"),
"patient_type": patient_data.get("patient_type")
}
try:
url_with_key = render_assign_url
if settings.api_token:
url_with_key = f"{render_assign_url}?api_key={settings.api_token}"
resp = await client.post(url_with_key, json=payload, headers={"Content-Type": "application/json"})
resp.raise_for_status()
logger.info(f"Successfully synced patient {payload['mrn']} ({payload['name']}) to Render Cloud for {device_id}")
return True
except Exception as e:
logger.warning(f"Could not sync patient assignment to Render Cloud: {type(e).__name__} ({e})")
return False
+140 -38
View File
@@ -17,47 +17,61 @@ async def retry_failed_transmissions():
while True:
try:
# 1. Fetch untransmitted readings in a quick DB session and close it immediately
db = SessionLocal()
readings_to_retry = []
try:
failed_readings = db.query(VitalReading).filter(VitalReading.transmitted == False).order_by(VitalReading.id.desc()).limit(20).all()
for r in failed_readings:
readings_to_retry.append({
"id": r.id,
"vitals": NormalizedVitals(
device_id=r.device_id,
patient_id=r.patient_id,
timestamp=r.timestamp,
heart_rate=r.heart_rate,
spo2=r.spo2,
systolic_bp=r.systolic_bp,
diastolic_bp=r.diastolic_bp,
map_bp=r.map_bp,
respiratory_rate=r.respiratory_rate,
temperature=r.temperature
)
})
finally:
db.close()
# 2. Forward outside of any DB transaction
successful_ids = []
failed_ids = []
consecutive_failures = 0
for item in readings_to_retry:
success = await forward_vitals_to_api(item["vitals"])
if success:
successful_ids.append(item["id"])
consecutive_failures = 0
else:
failed_ids.append(item["id"])
consecutive_failures += 1
# If external API is unresponsive, stop this batch immediately so we don't spam
if consecutive_failures >= 3:
logger.warning("External API appears unreachable during retry; pausing retry queue.")
break
# 3. Update status in a quick DB session
if successful_ids or failed_ids:
db = SessionLocal()
try:
# Get readings that haven't been transmitted
failed_readings = db.query(VitalReading).filter(VitalReading.transmitted == False).limit(50).all()
for reading in failed_readings:
vitals = NormalizedVitals(
device_id=reading.device_id,
patient_id=reading.patient_id,
timestamp=reading.timestamp,
heart_rate=reading.heart_rate,
spo2=reading.spo2,
systolic_bp=reading.systolic_bp,
diastolic_bp=reading.diastolic_bp,
map_bp=reading.map_bp,
respiratory_rate=reading.respiratory_rate,
temperature=reading.temperature
)
success = await forward_vitals_to_api(vitals)
if success:
reading.transmitted = True
log_entry = TransmissionLog(
reading_id=reading.id,
status="success",
response_code=200
)
db.add(log_entry)
logger.info(f"Successfully re-transmitted reading {reading.id}")
else:
# Log the failure but don't mark as transmitted
log_entry = TransmissionLog(
reading_id=reading.id,
status="failed",
error_message="Retry failed"
)
db.add(log_entry)
if failed_readings:
for rid in successful_ids:
db.query(VitalReading).filter(VitalReading.id == rid).update({"transmitted": True})
db.add(TransmissionLog(reading_id=rid, status="success", response_code=200))
for rid in failed_ids:
db.add(TransmissionLog(reading_id=rid, status="failed", error_message="Retry failed"))
db.commit()
except Exception as e:
logger.error(f"Failed updating retry transmission status: {e}")
db.rollback()
finally:
db.close()
@@ -65,3 +79,91 @@ async def retry_failed_transmissions():
logger.error(f"Error in retry background task: {e}")
await asyncio.sleep(settings.retry_interval_seconds)
async def sync_cloud_state():
"""
Periodically syncs all assigned patients and their latest non-null vitals to Render Cloud.
Ensures that whenever Render restarts or deploys with an empty DB, the cloud dashboard
is immediately populated with the same data as localhost.
"""
import os
if os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", ""):
return # Do not run sync loop if we are running inside Render
from models import Patient
from api_client import sync_patient_assignment_to_cloud, forward_vitals_to_api
from datetime import datetime, timezone
from sqlalchemy import or_
logger.info("Starting background task for periodic cloud state sync to Render...")
await asyncio.sleep(5)
while True:
try:
db = SessionLocal()
try:
patients = db.query(Patient).filter(
Patient.patient_id != "UNKNOWN",
Patient.mrn != "UNKNOWN"
).all()
for p in patients:
if not p.device_id:
continue
# 1. Sync patient demographics assignment to Render
p_data = {
"mrn": p.mrn or p.patient_id,
"name": p.name,
"gender": p.gender,
"dob": p.dob,
"bed_id": p.bed_id,
"patient_type": getattr(p, "patient_type", None)
}
await sync_patient_assignment_to_cloud(p_data, p.device_id)
# 2. Forward latest non-null vitals to Render (Option B)
from sqlalchemy import or_
from datetime import datetime, timezone
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
VitalReading.device_id == p.device_id
)
hr_r = db.query(VitalReading.heart_rate).filter(patient_filters, VitalReading.heart_rate != None).order_by(VitalReading.timestamp.desc()).first()
spo2_r = db.query(VitalReading.spo2).filter(patient_filters, VitalReading.spo2 != None).order_by(VitalReading.timestamp.desc()).first()
bp_r = db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp).filter(patient_filters, VitalReading.systolic_bp != None).order_by(VitalReading.timestamp.desc()).first()
temp_r = db.query(VitalReading.temperature).filter(patient_filters, VitalReading.temperature != None).order_by(VitalReading.timestamp.desc()).first()
temp2_r = db.query(VitalReading.temperature2).filter(patient_filters, VitalReading.temperature2 != None).order_by(VitalReading.timestamp.desc()).first()
temp_diff_r = db.query(VitalReading.temperature_diff).filter(patient_filters, VitalReading.temperature_diff != None).order_by(VitalReading.timestamp.desc()).first()
rr_r = db.query(VitalReading.respiratory_rate).filter(patient_filters, VitalReading.respiratory_rate != None).order_by(VitalReading.timestamp.desc()).first()
if hr_r or spo2_r or temp_r or bp_r:
merged = NormalizedVitals(
device_id=p.device_id,
patient_id=p.mrn or p.patient_id,
patient_name=p.name,
patient_gender=p.gender,
patient_dob=p.dob,
bed_id=p.bed_id,
patient_type=getattr(p, "patient_type", None),
timestamp=datetime.now(timezone.utc),
heart_rate=hr_r[0] if hr_r else None,
spo2=spo2_r[0] if spo2_r else None,
systolic_bp=bp_r[0] if bp_r else None,
diastolic_bp=bp_r[1] if bp_r else None,
map_bp=bp_r[2] if bp_r else None,
temperature=temp_r[0] if temp_r else None,
temperature2=temp2_r[0] if temp2_r else None,
temperature_diff=temp_diff_r[0] if temp_diff_r else None,
respiratory_rate=rr_r[0] if rr_r else None
)
await forward_vitals_to_api(merged)
finally:
db.close()
except Exception as e:
logger.error(f"Error in sync_cloud_state background task: {e}")
await asyncio.sleep(15)
+113 -188
View File
@@ -58,13 +58,14 @@ CONTEC_PARAM_MAP = {
# Temperature
"TEMP": "temperature",
"TEMP1": "temperature",
"TEMP2": "temperature",
"TEMP2": "temperature2",
"T1": "temperature",
"T2": "temperature",
"T2": "temperature2",
"BODY_TEMP": "temperature",
"TEMP-1": "temperature",
"TEMP-2": "temperature",
"TEMP-2": "temperature2",
"BODYTEMP": "temperature",
"TD": "temperature_diff",
}
# Sentinel value used by Contec to mark an invalid/disconnected parameter
@@ -164,6 +165,10 @@ def parse_contec_hl7_text(raw_text: str, client_ip: str) -> Optional[NormalizedV
if len(fields) > 8 and fields[8]:
vitals_dict["patient_gender"] = fields[8].strip().upper()
# PID-35: Patient Category (ADULT / PEDIATRIC / NEONATAL)
if len(fields) > 35 and fields[35]:
vitals_dict["patient_type"] = fields[35].strip()
elif seg_name == "PV1":
# PV1-3: Assigned Patient Location (Room^Bed^...)
if len(fields) > 3 and fields[3]:
@@ -173,6 +178,10 @@ def parse_contec_hl7_text(raw_text: str, client_ip: str) -> Optional[NormalizedV
if bed:
vitals_dict["bed_id"] = bed
# PV1-18: Patient Type
if len(fields) > 18 and fields[18]:
vitals_dict["patient_type"] = fields[18].strip()
elif seg_name == "OBX":
if len(fields) > 5:
obs_val = fields[5].strip()
@@ -247,7 +256,7 @@ def _extract_vitals_block(data: bytes, offset: int, client_ip: str) -> Optional[
vitals_dict["spo2"] = float(spo2)
found = True
# Heart rate: prefer ECG HR; fall back to SpO2-derived PR if ECG invalid
# Heart rate: ECG HR at offset+2 is primary; SpO2 PR at offset+4 is fallback.
if is_valid_u16(hr_ecg) and 20 <= hr_ecg <= 300:
vitals_dict["heart_rate"] = float(hr_ecg)
found = True
@@ -260,59 +269,52 @@ def _extract_vitals_block(data: bytes, offset: int, client_ip: str) -> Optional[
def _parse_45_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 45-byte packet (Subtype 22) — contains NIBP values and alarm limits.
Parse 45/47-byte packet (Subtype 22) — contains NIBP values and alarm limits.
Verified layout:
[0-1] Packet length LE u16 = 45
[2-3] Marker: 01 46
[0-1] Packet length LE u16 = 45 or 47
[2-3] Marker: 01 46 or 00 46
[4-7] Sub-header (00 00 16 00)
[8-9] Year LE u16
[10] Month, [11] Day, [12] Hour, [13] Minute, [14] Seconds
[15-16] NIBP Systolic (LE u16, 9999 = no measurement)
[17-18] NIBP Diastolic (LE u16, 9999 = no measurement)
[19-20] NIBP MAP (LE u16, 9999 = no measurement)
[15-16] NIBP Systolic (LE u16, 32767 / 9999 = no measurement / idle)
[17-18] NIBP Diastolic (LE u16, 32767 / 9999 = no measurement / idle)
[19-20] NIBP MAP (LE u16, 32767 / 9999 = no measurement / idle)
"""
if len(data) < 22:
if len(data) < 21:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
logger.debug("DEBUG_NIBP_PKT len=%d nibp_hex=%s", len(data), data[8:24].hex())
nibp_sys = read_u16_le(data, 15)
nibp_dia = read_u16_le(data, 17)
nibp_map = read_u16_le(data, 19)
found = False
sentinels = (9999, 32767, 65535, 0)
# Mark NIBP fields as present so cache can clear them if 9999
for f in ["systolic_bp", "diastolic_bp", "map_bp"]:
if f not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append(f)
if nibp_sys is not None and nibp_sys != 9999 and 40 < nibp_sys < 250:
if nibp_sys is not None and nibp_sys not in sentinels and 40 <= nibp_sys <= 260:
vitals_dict["systolic_bp"] = float(nibp_sys)
found = True
else:
vitals_dict["systolic_bp"] = None
if "systolic_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("systolic_bp")
if nibp_dia is not None and nibp_dia != 9999 and 20 < nibp_dia < 200:
if nibp_dia is not None and nibp_dia not in sentinels and 20 <= nibp_dia <= 200:
vitals_dict["diastolic_bp"] = float(nibp_dia)
found = True
else:
vitals_dict["diastolic_bp"] = None
if "diastolic_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("diastolic_bp")
if nibp_map is not None and nibp_map != 9999 and 20 < nibp_map < 250:
if nibp_map is not None and nibp_map not in sentinels and 20 <= nibp_map <= 250:
vitals_dict["map_bp"] = float(nibp_map)
found = True
else:
vitals_dict["map_bp"] = None
if "map_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("map_bp")
return found
return True
def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 56-byte packet (Subtype 23) — contains float32 Temperature values.
Parse 56-byte packet (Subtype 23) — contains float32 Temperature values for T1 and T2.
Verified layout:
[8-11] Temperature 1 (float32)
@@ -325,144 +327,83 @@ def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
vitals_dict.setdefault("temperature", None)
found = False
temp1 = read_f32_le(data, 8)
temp2 = read_f32_le(data, 12)
# 9999.0 is sentinel
if temp1 is not None and is_valid_f32(temp1) and 10.0 < temp1 < 50.0:
vitals_dict["temperature"] = round(temp1, 1)
found = True
elif temp2 is not None and is_valid_f32(temp2) and 10.0 < temp2 < 50.0:
vitals_dict["temperature"] = round(temp2, 1)
found = True
def is_real_temp(t):
# Disconnected sentinel is 9999.0 or 3276.7 (0x7FFF)
return t is not None and is_valid_f32(t) and abs(t - 3276.7) > 1.0 and 20.0 < t < 45.0
return found
t1_val = round(temp1, 1) if is_real_temp(temp1) else None
t2_val = round(temp2, 1) if is_real_temp(temp2) else None
vitals_dict["temperature"] = t1_val
vitals_dict["temperature2"] = t2_val
if t1_val is not None and t2_val is not None:
vitals_dict["temperature_diff"] = round(abs(t1_val - t2_val), 1)
else:
vitals_dict["temperature_diff"] = None
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
if "temperature2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature2")
if "temperature_diff" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature_diff")
return (t1_val is not None or t2_val is not None)
def _parse_spo2_waveform_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse SpO2 & Pleth waveform packet (Subtype 21 for CMS7000PLUS / CMS8500).
SpO2% is located at offset 264. Disconnected sentinel is 65535, 32767, or 255.
Offsets 270+ contain alarm limits (100, 90) and must NOT be parsed as vitals.
"""
if len(data) < 266:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
# Extract Pleth waveform samples (bytes 8..263)
vitals_dict["pleth_wave"] = list(data[8:min(264, len(data))])
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
# Check SpO2 at standard offset 264
spo2_val = read_u16_le(data, 264)
if spo2_val is not None and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
# Pulse Rate / ECG HR is at offset 266 in the footer
ecg_hr = read_u16_le(data, 266)
if ecg_hr is not None and ecg_hr not in (9999, 65535, 32767, 0) and 20 <= ecg_hr <= 300:
vitals_dict["heart_rate"] = float(ecg_hr)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
logger.info(
f"SpO2 pkt: SpO2[264]={spo2_val} ECG_HR[266]={ecg_hr} alarm_hi[268]={read_u16_le(data,268)} alarm_lo[270]={read_u16_le(data,270) if len(data)>=272 else 'N/A'}"
)
return True
def _parse_286_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 286-byte waveform packet (Subtype 21).
Verified layout:
[8..263] Pleth waveform samples (8-bit unsigned, 256 samples)
[264-265] SpO2% (LE u16) — live oxygen saturation percentage
255 / 65535 = sensor disconnected
[266-267] ECG HR via SpO2 PR (LE u16) — 9999 / 65535 = not available
"""
if len(data) < 270:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
vitals_dict.setdefault("spo2", None)
found = False
# Extract Pleth waveform samples (bytes 8..263)
raw_pleth = list(data[8:264])
# Only store if not all-zeroes or all-sentinel (flatline = disconnected)
if any(v not in (0x00, 0x7F, 0x3A, 0x92, 0xFF) for v in raw_pleth[:10]):
vitals_dict["pleth_wave"] = raw_pleth
else:
vitals_dict["pleth_wave"] = raw_pleth # Always send so frontend can show flatline too
# Offset 264: SpO2 percentage
spo2_val = read_u16_le(data, 264)
if spo2_val is not None and spo2_val != 65535 and spo2_val != 255 and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
found = True
# Offset 266: HR from SpO2 PR (only add to present fields if valid to prevent overriding ECG HR)
hr_val = read_u16_le(data, 266)
if hr_val is not None and hr_val != 65535 and hr_val != 9999 and 20 <= hr_val <= 300:
vitals_dict["heart_rate"] = float(hr_val)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
found = True
return found
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_288_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 288-byte waveform packet (Subtype 21 for CMS8500).
Layout:
[8..263] Pleth waveform samples (8-bit unsigned, 256 samples)
[264-265] SpO2% (LE u16) — live oxygen saturation percentage
[266-267] PR (LE u16) — pulse rate
"""
if len(data) < 270:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
vitals_dict.setdefault("spo2", None)
found = False
# Extract Pleth waveform samples (bytes 8..263)
vitals_dict["pleth_wave"] = list(data[8:264])
# Offset 264: SpO2 percentage
spo2_val = read_u16_le(data, 264)
if spo2_val is not None and spo2_val != 65535 and spo2_val != 255 and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
found = True
# Offset 266: PR from SpO2 PR
hr_val = read_u16_le(data, 266)
if hr_val is not None and hr_val != 65535 and hr_val != 9999 and 20 <= hr_val <= 300:
vitals_dict["heart_rate"] = float(hr_val)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
found = True
return found
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_341_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 341-byte waveform packet. Same tail layout as 286-byte.
"""
if len(data) < 270:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
vitals_dict.setdefault("spo2", None)
found = False
spo2_val = read_u16_le(data, 264)
if spo2_val is not None and spo2_val != 65535 and spo2_val != 255 and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
found = True
hr_val = read_u16_le(data, 266)
if hr_val is not None and hr_val != 65535 and hr_val != 9999 and 20 <= hr_val <= 300:
vitals_dict["heart_rate"] = float(hr_val)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
found = True
return found
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
@@ -479,38 +420,27 @@ def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
# Extract ECG waveform samples (bytes 8..903), downsample to ~224 points for smooth real-time rendering
raw_ecg = list(data[8:904])
vitals_dict["ecg_wave"] = raw_ecg[::4]
if "ecg_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("ecg_wave")
vitals_dict.setdefault("heart_rate", None)
found = False
# Extract ECG waveform samples (bytes 8..903), downsample to 200 points for efficiency
raw_ecg = list(data[8:904])
# Downsample by factor of 4 -> ~224 points (still high-resolution enough)
vitals_dict["ecg_wave"] = raw_ecg[::4]
logger.debug("DEBUG_ECG_PKT len=%d tail_hex=%s", len(data), data[890:989].hex())
hr_val = read_u16_le(data, 904)
if hr_val is not None and hr_val != 65535 and hr_val != 9999 and 20 <= hr_val <= 300:
if hr_val is not None and hr_val not in (65535, 32767, 9999, 0) and 20 <= hr_val <= 300:
if vitals_dict.get("heart_rate") is None:
vitals_dict["heart_rate"] = float(hr_val)
found = True
if "heart_rate" not in vitals_dict.setdefault("present_fields", []):
vitals_dict["present_fields"].append("heart_rate")
return found
return True
def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[NormalizedVitals]:
"""
Scan the data stream for Contec CMS7000PLUS sub-packets and parse them
deterministically based on packet length.
Packet types (verified by raw hex analysis):
45 bytes (Subtype 22): NIBP Sys/Dia/MAP (u16 LE starting at offset 15)
56 bytes (Subtype 23): Float32 Temperature (offset 8/12)
286 bytes (Subtype 21): SpO2% at offset 264
341 bytes (Subtype 21): SpO2% at offset 264
989 bytes (Subtype 20): ECG Heart Rate at offset 904
Scan the data stream for Contec CMS7000PLUS / CMS8500 sub-packets and parse them.
"""
model = settings.device_models.get(client_ip, settings.monitor_model)
vitals_dict = {
@@ -530,23 +460,18 @@ def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[Normal
pkt_data = data[i:i + pkt_len]
if pkt_len in (45, 47):
_parse_45_byte_packet(pkt_data, vitals_dict)
if _parse_45_byte_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len == 56:
_parse_56_byte_packet(pkt_data, vitals_dict)
if _parse_56_byte_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len == 286:
_parse_286_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 288:
_parse_288_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 341:
_parse_341_byte_packet(pkt_data, vitals_dict)
elif pkt_len in (286, 288, 341):
if _parse_spo2_waveform_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len == 989:
_parse_989_byte_packet(pkt_data, vitals_dict)
if _parse_989_byte_packet(pkt_data, vitals_dict):
found_any = True
# Handled specifically by known packet sizes above
i += max(pkt_len, 4)
else:
+435 -65
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
import re
from datetime import datetime, timezone
from contec_parser import parse_contec_data
@@ -35,7 +36,10 @@ def generate_ack(received_msg: str) -> str:
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
ack_msh = f"MSH|^~\\&|{rec_app}|{rec_fac}|{sending_app}|{sending_fac}|{now_str}||ACK^R01|{msg_control_id}|P|2.3.1\r"
ack_msh = (
f"MSH|^~\\&|{rec_app}|{rec_fac}|{sending_app}|{sending_fac}"
f"|{now_str}||ACK|{now_str}|P|2.3\r"
)
ack_msa = f"MSA|AA|{msg_control_id}\r"
return ack_msh + ack_msa
@@ -44,6 +48,242 @@ def generate_ack(received_msg: str) -> str:
return ""
TELEMETRY_WORDS = {
'sensor', 'off', 'lead', 'leads', 'fall', 'searching', 'pulse',
'le', 'le1', 'le2', 'le3', 'le4', 'le5', 'alarm', 'probe',
'cuff', 'error', 'leak', 'motion', 'weak', 'signal', 'learning', 'arr',
'asystole', 'vfib', 'vtac', 'pvc', 'brady', 'tachy', 'adult', 'pediatric',
'neonate', 'unknown', 'none', 'clear', 'default', 'demo', 't1', 't2',
'spo2', 'ecg', 'nibp', 'resp', 'temp', 'hr', 'pr', 'map', 'dia', 'sys'
}
def is_telemetry_word(s_lower: str) -> bool:
"""Check if string matches hardware telemetry or alarm words."""
if s_lower in TELEMETRY_WORDS:
return True
tokens = set(re.findall(r'[a-zA-Z0-9]+', s_lower))
alarm_tokens = {
'sensor', 'alarm', 'lead', 'leads', 'searching', 'pulse', 'cuff', 'probe',
'off', 'fall', 'leak', 'motion', 'weak', 'arr', 'asystole', 'vfib', 'vtac',
'pvc', 'brady', 'tachy', 'error', 'ecg', 'spo2', 'nibp', 'resp', 'temp',
'mv', 'bpm', 'mmhg'
}
if any(tok in alarm_tokens for tok in tokens):
return True
return False
def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
"""
Extracts real patient demographics from Contec packets.
Only extracts clean, legitimate patient fields — strictly rejects sensor alarms,
waveform noise, and hardware telemetry.
"""
patient_type = None
data_upper = data.upper()
if b'PEDIATRIC' in data_upper or b'PED' in data_upper:
patient_type = 'Pediatric'
elif b'NEONATE' in data_upper or b'NEONATAL' in data_upper or b'NEO' in data_upper:
patient_type = 'Neonatal'
elif b'ADULT' in data_upper or b'ADU' in data_upper:
patient_type = 'Adult'
# Strictly reject binary waveform/measurement packets from generating false patient records.
# Patient MRN/Name extraction requires an authentic HL7 message with PID or MSH.
if b'PID|' not in data and b'MSH|' not in data:
if patient_type:
return {"patient_type": patient_type}
return None
clean = []
# Match strings of printable alphanumeric and reasonable characters, allowing single spaces
for m in re.finditer(rb'[a-zA-Z0-9_\-\.]{2,}(?:\s+[a-zA-Z0-9_\-\.]{2,})*', data):
s = m.group(0).decode('latin1', errors='ignore').strip()
if not s or len(s) < 2:
continue
s_lower = s.lower()
if is_telemetry_word(s_lower):
continue
if len(set(s)) == 1:
continue
if set(s).issubset({'.', '-', '_', '/', '\\'}):
continue
if len(s) > 6 and re.match(r'^[A-Za-z0-9]+$', s):
diffs = [abs(ord(s[i]) - ord(s[i-1])) for i in range(1, len(s))]
if sum(1 for d in diffs if d <= 2) / len(diffs) > 0.6:
continue
if re.match(r'^\.\d+$', s):
continue
clean.append(s)
if not clean and not patient_type:
return None
name = None
mrn = None
dob = None
gender = None
for s in clean:
s_up = s.upper()
s_low = s.lower()
# Check gender first
if s_up in ('MALE', 'FEMALE') and not gender:
gender = 'Male' if s_up == 'MALE' else 'Female'
continue
elif s_up in ('M', 'F') and not gender:
gender = 'Male' if s_up == 'M' else 'Female'
continue
if is_telemetry_word(s_low):
continue
# Date of birth: 4-digit birth year between 1920 and 2026, or YYYY-MM-DD
if re.match(r'^(19\d\d|20[01]\d|202[0-6])$', s) and not dob:
dob = s
continue
elif re.match(r'^\d{4}[-/]\d{2}[-/]\d{2}$', s) and not dob:
dob = s
continue
# Name candidate: contains letters and a vowel, length between 3 and 40
if re.search(r'^[A-Za-z\s\.\-]{3,40}$', s) and re.search(r'[aeiouyAEIOUY]', s) and not name:
if not is_telemetry_word(s_low):
name = s
continue
# MRN candidate: numeric or alphanumeric ID, 2..15 chars, not a birth year
if not mrn and re.match(r'^[A-Za-z0-9\-]{2,15}$', s) and not is_telemetry_word(s_low):
if not (s.isdigit() and 1920 <= int(s) <= 2030):
if s.isdigit() and len(s) > 10:
continue
mrn = s
continue
# Accept if valid MRN or Name was detected
if mrn or name:
clean_mrn = mrn or name
clean_name = name or f"Patient {clean_mrn}"
return {
"name": clean_name,
"mrn": clean_mrn,
"patient_id": clean_mrn,
"dob": dob,
"gender": gender,
"patient_type": patient_type,
}
elif patient_type:
return {
"patient_type": patient_type,
}
return None
async def apply_auto_demographics(client_ip: str, demo: dict):
from config import settings
from api_client import sync_patient_assignment_to_cloud
model = settings.device_models.get(client_ip, settings.monitor_model)
device_id = f"{model}_{client_ip}"
name = demo.get("name")
mrn = demo.get("mrn")
dob = demo.get("dob")
gender = demo.get("gender")
patient_type = demo.get("patient_type")
current = device_patient_map.get(device_id, {})
# If only patient_type updated
if not mrn and not name:
if patient_type and current.get("patient_type") != patient_type:
current["patient_type"] = patient_type
device_patient_map[device_id] = current
return
if (current.get("name") == name and current.get("mrn") == mrn and
current.get("dob") == dob and current.get("gender") == gender and
current.get("patient_type") == patient_type):
return
logger.info(f"*** REAL PATIENT DEMOGRAPHICS DETECTED FROM MACHINE ({client_ip}): Name={name}, MRN={mrn}, DOB={dob}, Gender={gender}, Type={patient_type} ***")
bed_id = "1" if "192.168.100.106" in client_ip else "2"
p_data = {
"patient_id": mrn,
"mrn": mrn,
"name": name,
"gender": gender,
"dob": dob,
"bed_id": bed_id,
"patient_type": patient_type or current.get("patient_type"),
}
device_patient_map[device_id] = p_data
if device_id in device_cache:
cached = device_cache[device_id]
cached.patient_id = mrn
cached.patient_name = name
cached.patient_gender = gender
cached.patient_dob = dob
cached.bed_id = bed_id
db = SessionLocal()
try:
db.query(Patient).filter(Patient.device_id == device_id).update({"device_id": None})
patient = db.query(Patient).filter(Patient.mrn == mrn).first()
if not patient:
patient = Patient(
patient_id=mrn,
mrn=mrn,
name=name or "Unknown",
gender=gender,
dob=dob,
bed_id=bed_id,
device_id=device_id
)
db.add(patient)
else:
patient.patient_id = mrn
patient.mrn = mrn
if name:
patient.name = name
if gender:
patient.gender = gender
if dob:
patient.dob = dob
patient.bed_id = bed_id
patient.device_id = device_id
db.commit()
except Exception as e:
logger.error(f"Error persisting auto patient: {e}")
db.rollback()
finally:
db.close()
try:
from routers.dashboard import get_ws_manager
ws_manager = get_ws_manager()
await ws_manager.broadcast_vitals({
"event": "patient_assigned",
"device_id": device_id,
"patient_id": mrn,
"mrn": mrn,
"patient_name": name,
"name": name,
"patient_gender": gender,
"gender": gender,
"patient_dob": dob,
"dob": dob,
"bed_id": bed_id,
"patient_type": p_data.get("patient_type"),
})
except Exception as e:
logger.debug(f"Failed to broadcast auto patient: {e}")
# Forward to Render Cloud Server so remote is kept in sync
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int):
"""
TCP server handler for Contec CMS7000PLUS client connections.
@@ -52,6 +292,13 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
client_ip, client_port = writer.get_extra_info('peername')
logger.info(f"Accepted Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
sock = writer.get_extra_info('socket')
if sock:
try:
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except Exception:
pass
display = get_terminal_display()
display.connection_opened(client_ip, port)
@@ -64,8 +311,15 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
logger.info(f"Contec client {client_ip}:{client_port} on port {port} sent EOF (0 bytes)")
break
# Dynamically extract and apply patient demographics sent from the machine
try:
demo = extract_demographics_from_packet(data, client_ip)
if demo:
await apply_auto_demographics(client_ip, demo)
except Exception as ex:
logger.debug(f"Demographics extraction error: {ex}")
buffer += data
logger.debug(f"Received {len(data)} bytes from {client_ip}:{client_port} on port {port} (buf={len(buffer)})")
# ---------------------------------------------------------------
# MLLP / HL7 text path (typically port 511)
@@ -166,6 +420,50 @@ device_cache = {}
device_field_timestamps = {} # {device_id: {field_name: datetime}}
last_db_write = {}
# In-memory mapping of device_id -> patient demographic dict:
# {"mrn": "...", "name": "...", "dob": "...", "gender": "...", "bed_id": "..."}
device_patient_map = {}
def get_assigned_patient(device_id: str):
"""Retrieve patient demographics assigned to device_id (cached or from DB)."""
if device_id in device_patient_map:
return device_patient_map[device_id]
db = SessionLocal()
try:
patient = db.query(Patient).filter(Patient.device_id == device_id).first()
if patient:
p_data = {
"patient_id": patient.mrn or patient.patient_id,
"mrn": patient.mrn or patient.patient_id,
"name": patient.name or "Unknown",
"gender": patient.gender,
"dob": patient.dob,
"bed_id": patient.bed_id,
"patient_type": getattr(patient, "patient_type", None),
}
device_patient_map[device_id] = p_data
return p_data
except Exception as e:
logger.debug(f"Error querying patient for device {device_id}: {e}")
finally:
db.close()
return None
def set_assigned_patient(device_id: str, patient_data: dict):
"""Update in-memory mapping and active device cache for device_id."""
device_patient_map[device_id] = patient_data
if device_id in device_cache:
cached = device_cache[device_id]
cached.patient_id = patient_data.get("mrn") or patient_data.get("patient_id") or "UNKNOWN"
cached.patient_name = patient_data.get("name")
cached.patient_gender = patient_data.get("gender")
cached.patient_dob = patient_data.get("dob")
cached.bed_id = patient_data.get("bed_id")
cached.patient_type = patient_data.get("patient_type")
async def process_vitals(vitals):
"""
@@ -177,6 +475,27 @@ async def process_vitals(vitals):
if device_id not in device_field_timestamps:
device_field_timestamps[device_id] = {}
# Check for assigned patient demographics
assigned = get_assigned_patient(device_id)
if (vitals.patient_id == "UNKNOWN" or not vitals.patient_name or vitals.patient_name == "Unknown") and assigned:
vitals.patient_id = assigned.get("mrn") or assigned.get("patient_id") or "UNKNOWN"
vitals.patient_name = assigned.get("name")
vitals.patient_gender = assigned.get("gender")
vitals.patient_dob = assigned.get("dob")
vitals.bed_id = assigned.get("bed_id")
vitals.patient_type = assigned.get("patient_type")
elif vitals.patient_name and vitals.patient_name != "Unknown":
# Monitor sent fresh demographics via HL7 PID segment!
device_patient_map[device_id] = {
"patient_id": vitals.patient_id,
"mrn": vitals.patient_id,
"name": vitals.patient_name,
"gender": vitals.patient_gender,
"dob": vitals.patient_dob,
"bed_id": vitals.bed_id,
"patient_type": getattr(vitals, "patient_type", None),
}
# Check if incoming packet has fresh waveforms
has_fresh_waveforms = (vitals.ecg_wave is not None) or (vitals.pleth_wave is not None)
@@ -203,20 +522,37 @@ async def process_vitals(vitals):
if value is not None:
active_fields.append(field)
# Update cache & record timestamps
# Update cache & record timestamps for present values
for field in active_fields:
val = getattr(vitals, field)
val = getattr(vitals, field, None)
if val is not None:
setattr(cached, field, val)
device_field_timestamps[device_id][field] = now
else:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
# Clean up stale fields (older than 15 seconds)
# Clean up stale fields:
# Live vitals (HR, SpO2, Temp, RR) have a 15-second window
# NIBP (BP, MAP) has a 15-minute window because it is measured intermittently
for field in list(device_field_timestamps[device_id].keys()):
last_updated = device_field_timestamps[device_id][field]
if (now - last_updated).total_seconds() > 15.0:
timeout = 900.0 if field in ("systolic_bp", "diastolic_bp", "map_bp") else 15.0
if (now - last_updated).total_seconds() > timeout:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
cached.timestamp = vitals.timestamp
if vitals.patient_id and vitals.patient_id != "UNKNOWN":
cached.patient_id = vitals.patient_id
if vitals.patient_name:
cached.patient_name = vitals.patient_name
if vitals.patient_gender:
cached.patient_gender = vitals.patient_gender
if vitals.patient_dob:
cached.patient_dob = vitals.patient_dob
if vitals.bed_id:
cached.bed_id = vitals.bed_id
vitals = cached
# 2. Update console display banner
@@ -224,7 +560,7 @@ async def process_vitals(vitals):
display.update_vitals(vitals)
logger.info(
f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id}: "
f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id} ({vitals.patient_name}): "
f"HR={vitals.heart_rate}, SpO2={vitals.spo2}, "
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp}, "
f"RR={vitals.respiratory_rate}, Temp={vitals.temperature}"
@@ -234,23 +570,41 @@ async def process_vitals(vitals):
try:
from routers.dashboard import get_ws_manager
ws_manager = get_ws_manager()
await ws_manager.broadcast_vitals(vitals.model_dump(mode="json"))
vitals_dict = vitals.model_dump(mode="json")
vitals_dict["mrn"] = vitals.patient_id
vitals_dict["name"] = vitals.patient_name
vitals_dict["gender"] = vitals.patient_gender
vitals_dict["dob"] = vitals.patient_dob
vitals_dict["bed_id"] = vitals.bed_id
vitals_dict["patient_type"] = getattr(vitals, "patient_type", None)
await ws_manager.broadcast_vitals(vitals_dict)
except Exception as e:
logger.debug(f"WebSocket broadcast failed: {e}")
# 4. Throttled Database & REST forwarding (at most once every 5 seconds)
# Note: If we have fresh waveforms, we bypass throttle for REST forwarding to keep cloud graphs real-time
# 4. Asynchronous background persistence and cloud forwarding (completely non-blocking, throttled to 1s)
now = datetime.now(timezone.utc)
should_write_db = False
if device_id not in last_db_write or (now - last_db_write[device_id]).total_seconds() >= 5.0:
should_write_db = True
should_persist = False
if device_id not in last_db_write or (now - last_db_write[device_id]).total_seconds() >= 1.0:
should_persist = True
last_db_write[device_id] = now
if should_write_db:
# Save to SQLite database and REST forward
if should_persist:
asyncio.create_task(_persist_and_forward_bg(vitals, should_persist))
# Clear waveforms from cache to prevent repetition in subsequent non-waveform sub-packets
if device_id in device_cache:
device_cache[device_id].ecg_wave = None
device_cache[device_id].pleth_wave = None
async def _persist_and_forward_bg(vitals, should_persist):
"""Background worker for DB persistence and external cloud forwarding."""
if not should_persist:
return
reading_id = None
if should_persist:
db = SessionLocal()
try:
# 1. Device tracking
device = db.query(Device).filter(Device.device_id == vitals.device_id).first()
if not device:
device = Device(
@@ -265,20 +619,20 @@ async def process_vitals(vitals):
device.status = "active"
device.last_seen = datetime.now(timezone.utc)
# 2. Patient tracking — store/update demographics from HL7
if vitals.patient_id == "UNKNOWN":
last_known_reading = (
db.query(VitalReading)
.filter(VitalReading.device_id == vitals.device_id)
.filter(VitalReading.patient_id != "UNKNOWN")
.order_by(VitalReading.timestamp.desc())
.first()
)
if last_known_reading:
vitals.patient_id = last_known_reading.patient_id
patient = None
if vitals.patient_id and vitals.patient_id != "UNKNOWN":
patient = db.query(Patient).filter(
(Patient.mrn == vitals.patient_id) | (Patient.patient_id == vitals.patient_id)
).first()
if not patient and vitals.device_id:
patient = db.query(Patient).filter(
Patient.device_id == vitals.device_id,
Patient.patient_id != "UNKNOWN"
).first()
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
if not patient:
if vitals.patient_id and vitals.patient_id != "UNKNOWN":
patient = Patient(
patient_id=vitals.patient_id,
mrn=vitals.patient_id,
@@ -286,11 +640,15 @@ async def process_vitals(vitals):
gender=vitals.patient_gender,
dob=vitals.patient_dob,
bed_id=vitals.bed_id,
patient_type=getattr(vitals, "patient_type", None),
device_id=vitals.device_id
)
db.add(patient)
else:
# Refresh demographics whenever monitor sends updated PID fields
if vitals.patient_name:
if vitals.patient_id and vitals.patient_id != "UNKNOWN":
patient.patient_id = vitals.patient_id
patient.mrn = vitals.patient_id
if vitals.patient_name and vitals.patient_name != "Unknown":
patient.name = vitals.patient_name
if vitals.patient_gender:
patient.gender = vitals.patient_gender
@@ -298,12 +656,27 @@ async def process_vitals(vitals):
patient.dob = vitals.patient_dob
if vitals.bed_id:
patient.bed_id = vitals.bed_id
if getattr(vitals, "patient_type", None):
patient.patient_type = vitals.patient_type
if vitals.device_id:
patient.device_id = vitals.device_id
if patient:
if not vitals.patient_id or vitals.patient_id == "UNKNOWN":
vitals.patient_id = patient.mrn or patient.patient_id
if not vitals.patient_name or vitals.patient_name == "Unknown":
vitals.patient_name = patient.name
if not vitals.patient_gender and patient.gender:
vitals.patient_gender = patient.gender
if not vitals.patient_dob and patient.dob:
vitals.patient_dob = patient.dob
if not vitals.bed_id and patient.bed_id:
vitals.bed_id = patient.bed_id
if not getattr(vitals, "patient_type", None) and getattr(patient, "patient_type", None):
vitals.patient_type = patient.patient_type
# 3. Create Reading Entry
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
patient_id=vitals.patient_id or (patient.mrn if patient else "UNKNOWN"),
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
@@ -312,45 +685,42 @@ async def process_vitals(vitals):
map_bp=vitals.map_bp,
respiratory_rate=vitals.respiratory_rate,
temperature=vitals.temperature,
temperature2=vitals.temperature2,
temperature_diff=vitals.temperature_diff,
transmitted=False
)
db.add(reading)
db.commit()
db.refresh(reading)
# 4. REST forwarding
success = await forward_vitals_to_api(vitals)
if success:
reading.transmitted = True
log_entry = TransmissionLog(
reading_id=reading.id,
status="success",
response_code=200
)
else:
log_entry = TransmissionLog(
reading_id=reading.id,
status="failed",
error_message="Immediate REST forward failed"
)
db.add(log_entry)
db.commit()
reading_id = reading.id
except Exception as e:
logger.error(f"Database operation failed: {e}")
logger.error(f"Background database operation failed: {e}")
db.rollback()
finally:
db.close()
elif has_fresh_waveforms:
# Bypass throttle for REST forwarding to keep cloud graphs real-time
try:
await forward_vitals_to_api(vitals)
except Exception as e:
logger.error(f"Real-time REST forwarding failed: {e}")
# Clear waveforms from cache to prevent repetition in subsequent non-waveform sub-packets
if device_id in device_cache:
device_cache[device_id].ecg_wave = None
device_cache[device_id].pleth_wave = None
# DB connection is now completely closed. Awaiting HTTP requests will never exhaust DB connections!
success = False
if should_persist:
try:
success = await forward_vitals_to_api(vitals)
except Exception as e:
logger.warning(f"Background REST forward failed: {type(e).__name__}: {e}")
# Record transmission log in a separate, brief connection
if reading_id is not None:
db = SessionLocal()
try:
log_entry = TransmissionLog(
reading_id=reading_id,
status="success" if success else "failed",
response_code=200 if success else 500,
error_message=None if success else "Immediate REST forward failed"
)
db.add(log_entry)
db.commit()
except Exception as e:
logger.error(f"Failed to record transmission log: {e}")
db.rollback()
finally:
db.close()
+6 -1
View File
@@ -1,9 +1,14 @@
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
from sqlalchemy.orm import sessionmaker, declarative_base
from config import settings
connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
engine = create_engine(
settings.database_url, connect_args={"check_same_thread": False}
settings.database_url,
connect_args=connect_args,
poolclass=NullPool
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+60 -3
View File
@@ -11,7 +11,7 @@ import uvicorn
from config import settings
from database import engine, Base
from contec_server import handle_contec_client
from background_tasks import retry_failed_transmissions
from background_tasks import retry_failed_transmissions, sync_cloud_state
from routers import dashboard
from routers.dashboard import get_ws_manager
@@ -62,6 +62,22 @@ def _get_local_ips():
@asynccontextmanager
async def lifespan(app: FastAPI):
# Protect against Windows IOCP Proactor crashes (WinError 64 / 121 / 10054 on accept)
import sys
if sys.platform == "win32":
try:
loop = asyncio.get_running_loop()
def silence_windows_iocp_errors(loop, context):
msg = str(context.get("message", ""))
exc = context.get("exception")
winerr = getattr(exc, "winerror", None)
if winerr in (64, 121, 10054) or "Accept failed on a socket" in msg or "The specified network name is no longer available" in msg:
return
loop.default_exception_handler(context)
loop.set_exception_handler(silence_windows_iocp_errors)
except Exception:
pass
servers = []
# === Start Contec CMS7000PLUS TCP Servers ===
@@ -105,7 +121,8 @@ async def lifespan(app: FastAPI):
print("\n" + "=" * 65)
print(f" Contec {model} Patient Monitor - Vital Signs Forwarder")
print("=" * 65)
print(f" DASHBOARD: http://localhost:{port}/api/dashboard")
print(f" DASHBOARD: http://localhost:{port}/dashboard")
print(f" http://localhost:{port}/api/dashboard")
print(f" API Docs: http://localhost:{port}/docs")
print(f" Contec Ports: {settings.contec_ports}")
print(f" Monitor Model: {model}")
@@ -121,14 +138,40 @@ async def lifespan(app: FastAPI):
print(f" Set Server IP = one of the IPs above, Port = {settings.contec_ports[0] if settings.contec_ports else 511}")
print("=" * 65 + "\n")
# Startup: Start the background retry task
# Startup: Start background retry and cloud state sync tasks
retry_task = asyncio.create_task(retry_failed_transmissions())
cloud_sync_task = asyncio.create_task(sync_cloud_state())
# Startup: Start bore tunnel for remote monitors if available and not already running
bore_proc = None
bore_exe = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bore.exe")
if os.path.exists(bore_exe):
import subprocess
try:
check_out = subprocess.check_output('tasklist /FI "IMAGENAME eq bore.exe"', shell=True).decode()
if "bore.exe" not in check_out:
bore_proc = subprocess.Popen(
[bore_exe, "local", "511", "--to", "159.223.110.159", "--port", "61501"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
logger.info(" [OK] Bore tunnel active for remote hotspot monitors (159.223.110.159:61501)")
else:
logger.info(" [INFO] Bore tunnel is already running")
except Exception as e:
logger.debug(f"Could not check/start bore tunnel: {e}")
yield
# Shutdown
logger.info("Shutting down servers...")
if bore_proc:
try:
bore_proc.terminate()
except Exception:
pass
retry_task.cancel()
cloud_sync_task.cancel()
for server in servers:
server.close()
for server in servers:
@@ -156,10 +199,24 @@ app.add_middleware(
# Include dashboard routes
app.include_router(dashboard.router, prefix="/api")
# Also serve dashboard at root and /dashboard directly for convenience
from fastapi.responses import HTMLResponse
@app.get("/", response_class=HTMLResponse)
@app.get("/dashboard", response_class=HTMLResponse)
def serve_dashboard_root():
return dashboard.serve_dashboard()
# Serve static folder
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
from fastapi.responses import Response
@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
return Response(status_code=204, media_type="image/x-icon")
# === WebSocket endpoint for live vitals ===
@app.websocket("/ws/vitals")
+4
View File
@@ -19,6 +19,8 @@ class Patient(Base):
gender = Column(String, nullable=True) # M/F/U from PID-8
dob = Column(String, nullable=True) # YYYYMMDD from PID-7
bed_id = Column(String, nullable=True) # bed from PV1-3
patient_type = Column(String, nullable=True) # Adult / Pediatric / Neonatal
device_id = Column(String, nullable=True, index=True) # assigned machine / device_id
class VitalReading(Base):
__tablename__ = "vital_readings"
@@ -33,6 +35,8 @@ class VitalReading(Base):
map_bp = Column(Float, nullable=True)
respiratory_rate = Column(Float, nullable=True)
temperature = Column(Float, nullable=True)
temperature2 = Column(Float, nullable=True)
temperature_diff = Column(Float, nullable=True)
transmitted = Column(Boolean, default=False, index=True)
class TransmissionLog(Base):
+2
View File
@@ -5,6 +5,8 @@ services:
buildCommand: pip install -r requirements.txt
startCommand: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT
envVars:
- key: PYTHONPATH
value: /opt/render/project/src
- key: PORT
value: 10000
- key: PYTHON_VERSION
+1
View File
@@ -0,0 +1 @@
# Python package marker
+696
View File
@@ -0,0 +1,696 @@
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 datetime import datetime, timezone, timedelta
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
patient_type: 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(),
headers={
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
)
except FileNotFoundError:
return HTMLResponse(content="<h1>Dashboard not found</h1><p>static/dashboard.html is missing</p>", 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_criteria = []
if mrn:
mrn_clean = mrn.strip()
patient_criteria.append(or_(Patient.mrn == mrn_clean, Patient.patient_id == mrn_clean))
if name:
name_clean = name.strip()
patient_criteria.append(Patient.name.ilike(f"%{name_clean}%"))
if dob:
dob_clean = dob.strip().replace("-", "").replace("/", "")
patient_criteria.append(or_(
Patient.dob.ilike(f"%{dob.strip()}%"),
Patient.dob.ilike(f"%{dob_clean}%"),
Patient.dob.ilike(f"%{dob.strip()[:4]}%") if len(dob.strip()) >= 4 else False
))
if patient_criteria:
from sqlalchemy import and_
matching_patients = db.query(Patient).filter(and_(*patient_criteria)).all()
if not matching_patients:
return []
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 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))
else:
return []
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.
Strictly enforce live data timeouts:
- Continuous parameters (HR, SpO2, RR, Temp): max 25 seconds age.
- Intermittent parameters (NIBP): max 15 minutes age.
- Never backfill from stale historical readings.
"""
patients = db.query(Patient).filter(Patient.patient_id != "UNKNOWN", Patient.mrn != "UNKNOWN").all()
results = []
now = datetime.now(timezone.utc)
live_cutoff = now - timedelta(seconds=25)
nibp_cutoff = now - timedelta(minutes=15)
for p in patients:
merged_vitals = None
# Primary source: in-memory live cache if running locally with active machine stream
try:
from contec_server import device_cache, device_field_timestamps
if p.device_id and p.device_id in device_cache:
cached = device_cache[p.device_id]
field_ts = device_field_timestamps.get(p.device_id, {})
cached_time = cached.timestamp
if cached_time:
if cached_time.tzinfo is None:
cached_time = cached_time.replace(tzinfo=timezone.utc)
if (now - cached_time).total_seconds() <= 25:
merged_vitals = {
"heart_rate": cached.heart_rate if (field_ts.get("heart_rate") and (now - field_ts["heart_rate"]).total_seconds() <= 25) else None,
"spo2": cached.spo2 if (field_ts.get("spo2") and (now - field_ts["spo2"]).total_seconds() <= 25) else None,
"respiratory_rate": cached.respiratory_rate if (field_ts.get("respiratory_rate") and (now - field_ts["respiratory_rate"]).total_seconds() <= 25) else None,
"temperature": cached.temperature if (field_ts.get("temperature") and (now - field_ts["temperature"]).total_seconds() <= 25) else None,
"temperature2": getattr(cached, 'temperature2', None) if (field_ts.get("temperature2") and (now - field_ts["temperature2"]).total_seconds() <= 25) else None,
"temperature_diff": getattr(cached, 'temperature_diff', None) if (field_ts.get("temperature_diff") and (now - field_ts["temperature_diff"]).total_seconds() <= 25) else None,
"systolic_bp": cached.systolic_bp if (field_ts.get("systolic_bp") and (now - field_ts["systolic_bp"]).total_seconds() <= 900) else None,
"diastolic_bp": cached.diastolic_bp if (field_ts.get("diastolic_bp") and (now - field_ts["diastolic_bp"]).total_seconds() <= 900) else None,
"map_bp": cached.map_bp if (field_ts.get("map_bp") and (now - field_ts["map_bp"]).total_seconds() <= 900) else None,
"timestamp": cached_time.isoformat()
}
except Exception as e:
logger.debug(f"Could not read from device_cache: {e}")
# Fallback or cloud/Render mode: query DB
if merged_vitals is None:
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
(VitalReading.device_id == p.device_id) if p.device_id else False
)
# 1. Try to find readings within the live cutoff window (last 25 seconds)
recent_readings = (
db.query(VitalReading)
.filter(patient_filters, VitalReading.timestamp >= live_cutoff)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
# 2. Option B: If no live reading in last 25s, retain latest recorded readings from DB
if not recent_readings:
recent_readings = (
db.query(VitalReading)
.filter(patient_filters)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
if recent_readings:
ts_iso = recent_readings[0].timestamp.isoformat() if recent_readings[0].timestamp else None
if ts_iso and not ts_iso.endswith('Z') and '+' not in ts_iso:
ts_iso += 'Z'
merged_vitals = {
"heart_rate": None,
"spo2": None,
"systolic_bp": None,
"diastolic_bp": None,
"map_bp": None,
"respiratory_rate": None,
"temperature": None,
"temperature2": None,
"temperature_diff": None,
"timestamp": ts_iso,
}
for r in reversed(recent_readings):
if r.heart_rate is not None: merged_vitals["heart_rate"] = r.heart_rate
if r.spo2 is not None: merged_vitals["spo2"] = r.spo2
if r.respiratory_rate is not None: merged_vitals["respiratory_rate"] = r.respiratory_rate
if r.temperature is not None: merged_vitals["temperature"] = r.temperature
if getattr(r, 'temperature2', None) is not None: merged_vitals["temperature2"] = r.temperature2
if getattr(r, 'temperature_diff', None) is not None: merged_vitals["temperature_diff"] = r.temperature_diff
if r.systolic_bp is not None: merged_vitals["systolic_bp"] = r.systolic_bp
if r.diastolic_bp is not None: merged_vitals["diastolic_bp"] = r.diastolic_bp
if r.map_bp is not None: merged_vitals["map_bp"] = r.map_bp
# Check NIBP separately if not already populated
if not merged_vitals or merged_vitals.get("systolic_bp") is None:
r_bp = (
db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp, VitalReading.timestamp)
.filter(patient_filters, VitalReading.systolic_bp != None)
.order_by(VitalReading.timestamp.desc())
.first()
)
if r_bp:
if merged_vitals is None:
ts_bp = r_bp[3].isoformat() if r_bp[3] else None
if ts_bp and not ts_bp.endswith('Z') and '+' not in ts_bp:
ts_bp += 'Z'
merged_vitals = {
"heart_rate": None, "spo2": None, "respiratory_rate": None,
"temperature": None, "temperature2": None, "temperature_diff": None,
"timestamp": ts_bp
}
merged_vitals["systolic_bp"] = r_bp[0]
merged_vitals["diastolic_bp"] = r_bp[1]
merged_vitals["map_bp"] = r_bp[2]
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": merged_vitals
})
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
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud({"mrn": "UNKNOWN", "name": "Unassigned Patient"}, device_id))
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,
patient_type=req.patient_type,
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
if req.patient_type is not None:
patient.patient_type = req.patient_type
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,
"patient_type": patient.patient_type,
}
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,
"patient_type": patient.patient_type,
})
except Exception as e:
logger.debug(f"Broadcast of patient_assigned failed: {e}")
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
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.settimeout(0.5)
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)
vitals_dict["patient_type"] = getattr(vitals, "patient_type", 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)}")
+3
View File
@@ -14,6 +14,8 @@ class NormalizedVitals(BaseModel):
map_bp: Optional[float] = None
respiratory_rate: Optional[float] = None
temperature: Optional[float] = None
temperature2: Optional[float] = None
temperature_diff: Optional[float] = None
present_fields: Optional[list] = None
# Raw waveform sample arrays (8-bit unsigned, 0-255)
ecg_wave: Optional[List[int]] = None # ECG waveform samples from 989-byte packet
@@ -23,6 +25,7 @@ class NormalizedVitals(BaseModel):
patient_gender: Optional[str] = None # PID-8: M / F / U
patient_dob: Optional[str] = None # PID-7: YYYYMMDD
bed_id: Optional[str] = None # PV1-3: bed / room ID
patient_type: Optional[str] = None # Patient category: Adult / Pediatric / Neonatal
class DeviceStatus(BaseModel):
device_id: str
+37 -5
View File
@@ -5,11 +5,41 @@ set -e
SERVICE_NAME="contec-forwarder"
SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service"
WORKING_DIR="/home/prathiyuman/Prathiyuman/ContecMonitor/fiveparaminte-main"
# Auto-detect script directory
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if [ -f "${SCRIPT_DIR}/main.py" ]; then
WORKING_DIR="${SCRIPT_DIR}"
else
WORKING_DIR="${PWD}"
fi
echo "=== Detected Working Directory: ${WORKING_DIR} ==="
# Ensure virtual environment exists
if [ ! -f "${WORKING_DIR}/venv/bin/python" ]; then
echo "=== Virtual environment not found. Creating venv... ==="
python3 -m venv "${WORKING_DIR}/venv" || sudo apt-get update && sudo apt-get install -y python3-venv python3-pip && python3 -m venv "${WORKING_DIR}/venv"
fi
PYTHON_EXEC="${WORKING_DIR}/venv/bin/python"
SCRIPT_PATH="${WORKING_DIR}/main.py"
echo "=== Creating systemd service file ==="
# Ensure dependencies are installed
if [ -f "${WORKING_DIR}/requirements.txt" ]; then
echo "=== Installing/verifying dependencies ==="
"${WORKING_DIR}/venv/bin/pip" install --upgrade pip
"${WORKING_DIR}/venv/bin/pip" install -r "${WORKING_DIR}/requirements.txt"
fi
# Open firewall ports for Contec monitor (Port 511) and Web API (Port 8006)
if command -v ufw >/dev/null 2>&1; then
echo "=== Configuring firewall (UFW) for Port 511 and 8006 ==="
sudo ufw allow 511/tcp comment "Contec Patient Monitor CMS TCP" || true
sudo ufw allow 8006/tcp comment "Contec Forwarder HTTP API" || true
fi
echo "=== Creating systemd service file: ${SERVICE_FILE} ==="
sudo bash -c "cat > ${SERVICE_FILE}" <<EOF
[Unit]
Description=Contec Patient Monitor Vital Signs Forwarder
@@ -20,6 +50,7 @@ Type=simple
User=root
WorkingDirectory=${WORKING_DIR}
Environment=PORT=8006
Environment=PYTHONPATH=${WORKING_DIR}
ExecStart=${PYTHON_EXEC} ${SCRIPT_PATH}
Restart=always
RestartSec=5
@@ -36,7 +67,7 @@ sudo systemctl daemon-reload
echo "=== Enabling ${SERVICE_NAME} to start on boot ==="
sudo systemctl enable ${SERVICE_NAME}
echo "=== Starting ${SERVICE_NAME} service ==="
echo "=== Restarting ${SERVICE_NAME} service ==="
sudo systemctl restart ${SERVICE_NAME}
echo "=== Service Status ==="
@@ -45,6 +76,7 @@ sudo systemctl status ${SERVICE_NAME} --no-pager
echo ""
echo "=========================================================="
echo "SUCCESS: Contec Forwarder is now running in the background!"
echo "To check live service logs, run:"
echo " sudo journalctl -u contec-forwarder -f"
echo "Port 511 is open and listening for the Contec physical monitor."
echo "To check live logs, run:"
echo " sudo journalctl -u ${SERVICE_NAME} -f"
echo "=========================================================="
File diff suppressed because it is too large Load Diff
+27 -2
View File
@@ -11,7 +11,7 @@ import uvicorn
from config import settings
from database import engine, Base
from contec_server import handle_contec_client
from background_tasks import retry_failed_transmissions
from background_tasks import retry_failed_transmissions, sync_cloud_state
from routers import dashboard
from routers.dashboard import get_ws_manager
@@ -62,6 +62,22 @@ def _get_local_ips():
@asynccontextmanager
async def lifespan(app: FastAPI):
# Protect against Windows IOCP Proactor crashes (WinError 64 / 121 / 10054 on accept)
import sys
if sys.platform == "win32":
try:
loop = asyncio.get_running_loop()
def silence_windows_iocp_errors(loop, context):
msg = str(context.get("message", ""))
exc = context.get("exception")
winerr = getattr(exc, "winerror", None)
if winerr in (64, 121, 10054) or "Accept failed on a socket" in msg or "The specified network name is no longer available" in msg:
return
loop.default_exception_handler(context)
loop.set_exception_handler(silence_windows_iocp_errors)
except Exception:
pass
servers = []
# === Start Contec CMS7000PLUS TCP Servers ===
@@ -122,8 +138,9 @@ async def lifespan(app: FastAPI):
print(f" Set Server IP = one of the IPs above, Port = {settings.contec_ports[0] if settings.contec_ports else 511}")
print("=" * 65 + "\n")
# Startup: Start the background retry task
# Startup: Start background retry and cloud state sync tasks
retry_task = asyncio.create_task(retry_failed_transmissions())
cloud_sync_task = asyncio.create_task(sync_cloud_state())
# Startup: Start bore tunnel for remote monitors if available and not already running
bore_proc = None
@@ -154,6 +171,7 @@ async def lifespan(app: FastAPI):
except Exception:
pass
retry_task.cancel()
cloud_sync_task.cancel()
for server in servers:
server.close()
for server in servers:
@@ -192,6 +210,13 @@ def serve_dashboard_root():
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
from fastapi.responses import Response
@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
return Response(status_code=204, media_type="image/x-icon")
# === WebSocket endpoint for live vitals ===
@app.websocket("/ws/vitals")
+3
View File
@@ -19,6 +19,7 @@ class Patient(Base):
gender = Column(String, nullable=True) # M/F/U from PID-8
dob = Column(String, nullable=True) # YYYYMMDD from PID-7
bed_id = Column(String, nullable=True) # bed from PV1-3
patient_type = Column(String, nullable=True) # Adult / Pediatric / Neonatal
device_id = Column(String, nullable=True, index=True) # assigned machine / device_id
class VitalReading(Base):
@@ -34,6 +35,8 @@ class VitalReading(Base):
map_bp = Column(Float, nullable=True)
respiratory_rate = Column(Float, nullable=True)
temperature = Column(Float, nullable=True)
temperature2 = Column(Float, nullable=True)
temperature_diff = Column(Float, nullable=True)
transmitted = Column(Boolean, default=False, index=True)
class TransmissionLog(Base):
+2
View File
@@ -5,6 +5,8 @@ services:
buildCommand: pip install -r requirements.txt
startCommand: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT
envVars:
- key: PYTHONPATH
value: /opt/render/project/src
- key: PORT
value: 10000
- key: PYTHON_VERSION
+252 -26
View File
@@ -11,6 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisco
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from sqlalchemy import or_
from datetime import datetime, timezone, timedelta
from database import get_db
from models import Device, Patient, VitalReading, TransmissionLog
@@ -24,6 +25,7 @@ class PatientAssignRequest(BaseModel):
dob: Optional[str] = None
gender: Optional[str] = None
bed_id: Optional[str] = None
patient_type: Optional[str] = None
# === WebSocket Manager for real-time vitals push ===
class VitalsWebSocketManager:
@@ -70,7 +72,14 @@ def serve_dashboard():
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())
return HTMLResponse(
content=f.read(),
headers={
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
)
except FileNotFoundError:
return HTMLResponse(content="<h1>Dashboard not found</h1><p>static/dashboard.html is missing</p>", status_code=404)
@@ -112,26 +121,30 @@ def get_latest_readings(
if device_id:
query = query.filter(VitalReading.device_id == device_id)
patient_filters = []
patient_criteria = []
if mrn:
mrn_clean = mrn.strip()
patient_filters.append(Patient.mrn == mrn_clean)
patient_filters.append(Patient.patient_id == mrn_clean)
patient_criteria.append(or_(Patient.mrn == mrn_clean, Patient.patient_id == mrn_clean))
if name:
name_clean = name.strip()
patient_filters.append(Patient.name.ilike(f"%{name_clean}%"))
patient_criteria.append(Patient.name.ilike(f"%{name_clean}%"))
if dob:
dob_clean = dob.strip()
patient_filters.append(Patient.dob.ilike(f"%{dob_clean}%"))
dob_clean = dob.strip().replace("-", "").replace("/", "")
patient_criteria.append(or_(
Patient.dob.ilike(f"%{dob.strip()}%"),
Patient.dob.ilike(f"%{dob_clean}%"),
Patient.dob.ilike(f"%{dob.strip()[:4]}%") if len(dob.strip()) >= 4 else False
))
if patient_filters:
matching_patients = db.query(Patient).filter(or_(*patient_filters)).all()
if patient_criteria:
from sqlalchemy import and_
matching_patients = db.query(Patient).filter(and_(*patient_criteria)).all()
if not matching_patients:
return []
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:
@@ -139,6 +152,8 @@ def get_latest_readings(
if conds:
query = query.filter(or_(*conds))
else:
return []
readings = query.order_by(VitalReading.timestamp.desc()).limit(limit).all()
return readings
@@ -147,23 +162,120 @@ def get_latest_readings(
def list_patients(db: Session = Depends(get_db)):
"""
List all patients with their assigned device and latest vitals readings.
Strictly enforce live data timeouts:
- Continuous parameters (HR, SpO2, RR, Temp): max 25 seconds age.
- Intermittent parameters (NIBP): max 15 minutes age.
- Never backfill from stale historical readings.
"""
patients = db.query(Patient).filter(Patient.patient_id != "UNKNOWN", Patient.mrn != "UNKNOWN").all()
results = []
now = datetime.now(timezone.utc)
live_cutoff = now - timedelta(seconds=25)
nibp_cutoff = now - timedelta(minutes=15)
for p in patients:
# Find latest vitals for this patient or this patient's assigned device
latest_reading = (
db.query(VitalReading)
.filter(
or_(
merged_vitals = None
# Primary source: in-memory live cache if running locally with active machine stream
try:
from contec_server import device_cache, device_field_timestamps
if p.device_id and p.device_id in device_cache:
cached = device_cache[p.device_id]
field_ts = device_field_timestamps.get(p.device_id, {})
cached_time = cached.timestamp
if cached_time:
if cached_time.tzinfo is None:
cached_time = cached_time.replace(tzinfo=timezone.utc)
if (now - cached_time).total_seconds() <= 25:
merged_vitals = {
"heart_rate": cached.heart_rate if (field_ts.get("heart_rate") and (now - field_ts["heart_rate"]).total_seconds() <= 25) else None,
"spo2": cached.spo2 if (field_ts.get("spo2") and (now - field_ts["spo2"]).total_seconds() <= 25) else None,
"respiratory_rate": cached.respiratory_rate if (field_ts.get("respiratory_rate") and (now - field_ts["respiratory_rate"]).total_seconds() <= 25) else None,
"temperature": cached.temperature if (field_ts.get("temperature") and (now - field_ts["temperature"]).total_seconds() <= 25) else None,
"temperature2": getattr(cached, 'temperature2', None) if (field_ts.get("temperature2") and (now - field_ts["temperature2"]).total_seconds() <= 25) else None,
"temperature_diff": getattr(cached, 'temperature_diff', None) if (field_ts.get("temperature_diff") and (now - field_ts["temperature_diff"]).total_seconds() <= 25) else None,
"systolic_bp": cached.systolic_bp if (field_ts.get("systolic_bp") and (now - field_ts["systolic_bp"]).total_seconds() <= 900) else None,
"diastolic_bp": cached.diastolic_bp if (field_ts.get("diastolic_bp") and (now - field_ts["diastolic_bp"]).total_seconds() <= 900) else None,
"map_bp": cached.map_bp if (field_ts.get("map_bp") and (now - field_ts["map_bp"]).total_seconds() <= 900) else None,
"timestamp": cached_time.isoformat()
}
except Exception as e:
logger.debug(f"Could not read from device_cache: {e}")
# Fallback or cloud/Render mode: query DB
if merged_vitals is None:
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
(VitalReading.device_id == p.device_id) if p.device_id else False
)
# 1. Try to find readings within the live cutoff window (last 25 seconds)
recent_readings = (
db.query(VitalReading)
.filter(patient_filters, VitalReading.timestamp >= live_cutoff)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
# 2. Option B: If no live reading in last 25s, retain latest recorded readings from DB
if not recent_readings:
recent_readings = (
db.query(VitalReading)
.filter(patient_filters)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
if recent_readings:
ts_iso = recent_readings[0].timestamp.isoformat() if recent_readings[0].timestamp else None
if ts_iso and not ts_iso.endswith('Z') and '+' not in ts_iso:
ts_iso += 'Z'
merged_vitals = {
"heart_rate": None,
"spo2": None,
"systolic_bp": None,
"diastolic_bp": None,
"map_bp": None,
"respiratory_rate": None,
"temperature": None,
"temperature2": None,
"temperature_diff": None,
"timestamp": ts_iso,
}
for r in reversed(recent_readings):
if r.heart_rate is not None: merged_vitals["heart_rate"] = r.heart_rate
if r.spo2 is not None: merged_vitals["spo2"] = r.spo2
if r.respiratory_rate is not None: merged_vitals["respiratory_rate"] = r.respiratory_rate
if r.temperature is not None: merged_vitals["temperature"] = r.temperature
if getattr(r, 'temperature2', None) is not None: merged_vitals["temperature2"] = r.temperature2
if getattr(r, 'temperature_diff', None) is not None: merged_vitals["temperature_diff"] = r.temperature_diff
if r.systolic_bp is not None: merged_vitals["systolic_bp"] = r.systolic_bp
if r.diastolic_bp is not None: merged_vitals["diastolic_bp"] = r.diastolic_bp
if r.map_bp is not None: merged_vitals["map_bp"] = r.map_bp
# Check NIBP separately if not already populated
if not merged_vitals or merged_vitals.get("systolic_bp") is None:
r_bp = (
db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp, VitalReading.timestamp)
.filter(patient_filters, VitalReading.systolic_bp != None)
.order_by(VitalReading.timestamp.desc())
.first()
)
if r_bp:
if merged_vitals is None:
ts_bp = r_bp[3].isoformat() if r_bp[3] else None
if ts_bp and not ts_bp.endswith('Z') and '+' not in ts_bp:
ts_bp += 'Z'
merged_vitals = {
"heart_rate": None, "spo2": None, "respiratory_rate": None,
"temperature": None, "temperature2": None, "temperature_diff": None,
"timestamp": ts_bp
}
merged_vitals["systolic_bp"] = r_bp[0]
merged_vitals["diastolic_bp"] = r_bp[1]
merged_vitals["map_bp"] = r_bp[2]
device = db.query(Device).filter(Device.device_id == p.device_id).first() if p.device_id else None
@@ -178,16 +290,7 @@ def list_patients(db: Session = Depends(get_db)):
"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
"latest_vitals": merged_vitals
})
return results
@@ -235,6 +338,8 @@ async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db
})
except Exception:
pass
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud({"mrn": "UNKNOWN", "name": "Unassigned Patient"}, device_id))
return {
"status": "success",
"message": f"Device {device_id} unassigned successfully"
@@ -256,6 +361,7 @@ async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db
gender=req.gender,
dob=req.dob,
bed_id=req.bed_id,
patient_type=req.patient_type,
device_id=device_id
)
db.add(patient)
@@ -270,6 +376,8 @@ async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db
patient.dob = req.dob
if req.bed_id is not None:
patient.bed_id = req.bed_id
if req.patient_type is not None:
patient.patient_type = req.patient_type
patient.device_id = device_id
# 3. Associate recent readings for this device that were marked "UNKNOWN"
@@ -289,6 +397,7 @@ async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db
"gender": patient.gender,
"dob": patient.dob,
"bed_id": patient.bed_id,
"patient_type": patient.patient_type,
}
set_assigned_patient(device_id, p_data)
@@ -306,10 +415,14 @@ async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db
"patient_dob": patient.dob,
"dob": patient.dob,
"bed_id": patient.bed_id,
"patient_type": patient.patient_type,
})
except Exception as e:
logger.debug(f"Broadcast of patient_assigned failed: {e}")
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
return {
"status": "success",
"message": f"Patient {patient.name} ({mrn}) assigned to device {device_id}",
@@ -445,6 +558,7 @@ def get_network_info():
primary_ip = None
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0.5)
s.connect(('8.8.8.8', 80))
primary_ip = s.getsockname()[0]
s.close()
@@ -468,3 +582,115 @@ 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 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)
vitals_dict["patient_type"] = getattr(vitals, "patient_type", 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)}")
+3
View File
@@ -14,6 +14,8 @@ class NormalizedVitals(BaseModel):
map_bp: Optional[float] = None
respiratory_rate: Optional[float] = None
temperature: Optional[float] = None
temperature2: Optional[float] = None
temperature_diff: Optional[float] = None
present_fields: Optional[list] = None
# Raw waveform sample arrays (8-bit unsigned, 0-255)
ecg_wave: Optional[List[int]] = None # ECG waveform samples from 989-byte packet
@@ -23,6 +25,7 @@ class NormalizedVitals(BaseModel):
patient_gender: Optional[str] = None # PID-8: M / F / U
patient_dob: Optional[str] = None # PID-7: YYYYMMDD
bed_id: Optional[str] = None # PV1-3: bed / room ID
patient_type: Optional[str] = None # Patient category: Adult / Pediatric / Neonatal
class DeviceStatus(BaseModel):
device_id: str
+37 -5
View File
@@ -5,11 +5,41 @@ set -e
SERVICE_NAME="contec-forwarder"
SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service"
WORKING_DIR="/home/prathiyuman/Prathiyuman/ContecMonitor/fiveparaminte-main"
# Auto-detect script directory
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if [ -f "${SCRIPT_DIR}/main.py" ]; then
WORKING_DIR="${SCRIPT_DIR}"
else
WORKING_DIR="${PWD}"
fi
echo "=== Detected Working Directory: ${WORKING_DIR} ==="
# Ensure virtual environment exists
if [ ! -f "${WORKING_DIR}/venv/bin/python" ]; then
echo "=== Virtual environment not found. Creating venv... ==="
python3 -m venv "${WORKING_DIR}/venv" || sudo apt-get update && sudo apt-get install -y python3-venv python3-pip && python3 -m venv "${WORKING_DIR}/venv"
fi
PYTHON_EXEC="${WORKING_DIR}/venv/bin/python"
SCRIPT_PATH="${WORKING_DIR}/main.py"
echo "=== Creating systemd service file ==="
# Ensure dependencies are installed
if [ -f "${WORKING_DIR}/requirements.txt" ]; then
echo "=== Installing/verifying dependencies ==="
"${WORKING_DIR}/venv/bin/pip" install --upgrade pip
"${WORKING_DIR}/venv/bin/pip" install -r "${WORKING_DIR}/requirements.txt"
fi
# Open firewall ports for Contec monitor (Port 511) and Web API (Port 8006)
if command -v ufw >/dev/null 2>&1; then
echo "=== Configuring firewall (UFW) for Port 511 and 8006 ==="
sudo ufw allow 511/tcp comment "Contec Patient Monitor CMS TCP" || true
sudo ufw allow 8006/tcp comment "Contec Forwarder HTTP API" || true
fi
echo "=== Creating systemd service file: ${SERVICE_FILE} ==="
sudo bash -c "cat > ${SERVICE_FILE}" <<EOF
[Unit]
Description=Contec Patient Monitor Vital Signs Forwarder
@@ -20,6 +50,7 @@ Type=simple
User=root
WorkingDirectory=${WORKING_DIR}
Environment=PORT=8006
Environment=PYTHONPATH=${WORKING_DIR}
ExecStart=${PYTHON_EXEC} ${SCRIPT_PATH}
Restart=always
RestartSec=5
@@ -36,7 +67,7 @@ sudo systemctl daemon-reload
echo "=== Enabling ${SERVICE_NAME} to start on boot ==="
sudo systemctl enable ${SERVICE_NAME}
echo "=== Starting ${SERVICE_NAME} service ==="
echo "=== Restarting ${SERVICE_NAME} service ==="
sudo systemctl restart ${SERVICE_NAME}
echo "=== Service Status ==="
@@ -45,6 +76,7 @@ sudo systemctl status ${SERVICE_NAME} --no-pager
echo ""
echo "=========================================================="
echo "SUCCESS: Contec Forwarder is now running in the background!"
echo "To check live service logs, run:"
echo " sudo journalctl -u contec-forwarder -f"
echo "Port 511 is open and listening for the Contec physical monitor."
echo "To check live logs, run:"
echo " sudo journalctl -u ${SERVICE_NAME} -f"
echo "=========================================================="
+1936 -236
View File
File diff suppressed because it is too large Load Diff