727 lines
28 KiB
Python
727 lines
28 KiB
Python
import asyncio
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
|
|
from contec_parser import parse_contec_data
|
|
from database import SessionLocal
|
|
from models import Device, Patient, VitalReading, TransmissionLog
|
|
from api_client import forward_vitals_to_api
|
|
from terminal_display import get_terminal_display
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Constants for MLLP framing (used on port 511 for Contec HL7 text)
|
|
VT = b'\x0b'
|
|
FS_CR = b'\x1c\x0d'
|
|
|
|
|
|
def generate_ack(received_msg: str) -> str:
|
|
"""Generates a simple HL7 ACK message to satisfy the monitor's TCP client."""
|
|
try:
|
|
segments = received_msg.split('\r')
|
|
msh = segments[0] if segments else ""
|
|
if not msh.startswith("MSH|"):
|
|
return ""
|
|
|
|
fields = msh.split('|')
|
|
if len(fields) < 10:
|
|
return ""
|
|
|
|
sending_app = fields[2]
|
|
sending_fac = fields[3]
|
|
rec_app = fields[4]
|
|
rec_fac = fields[5]
|
|
msg_control_id = fields[9]
|
|
|
|
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
|
|
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
|
|
except Exception as e:
|
|
logger.debug(f"Could not generate ACK: {e}")
|
|
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.
|
|
Directs incoming traffic to the unified parse_contec_data handler.
|
|
"""
|
|
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)
|
|
|
|
buffer = b""
|
|
|
|
try:
|
|
while True:
|
|
data = await reader.read(8192)
|
|
if not data:
|
|
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
|
|
|
|
# ---------------------------------------------------------------
|
|
# MLLP / HL7 text path (typically port 511)
|
|
# ---------------------------------------------------------------
|
|
if VT in buffer and FS_CR in buffer and b"MSH|" in buffer:
|
|
while True:
|
|
start_idx = buffer.find(VT)
|
|
end_idx = buffer.find(FS_CR)
|
|
if start_idx != -1 and end_idx != -1 and start_idx < end_idx:
|
|
raw_hl7 = buffer[start_idx + 1:end_idx].decode('utf-8', errors='ignore')
|
|
buffer = buffer[end_idx + 2:]
|
|
logger.info(f"Contec MLLP message detected on port {port}")
|
|
vitals = parse_contec_data(raw_hl7.encode('utf-8'), client_ip)
|
|
if vitals:
|
|
await process_vitals(vitals)
|
|
ack_msg = generate_ack(raw_hl7)
|
|
if ack_msg:
|
|
writer.write(VT + ack_msg.encode('utf-8') + FS_CR)
|
|
await writer.drain()
|
|
else:
|
|
break
|
|
continue # Go back to reading more data
|
|
|
|
# ---------------------------------------------------------------
|
|
# Binary Contec proprietary protocol (port 514)
|
|
# Packets are framed as: [len_lo][len_hi][04][46] ... payload ...
|
|
# The monitor sends each packet as a complete TCP write,
|
|
# but we buffer anyway in case of TCP segmentation.
|
|
# ---------------------------------------------------------------
|
|
consumed = 0
|
|
while consumed < len(buffer) - 4:
|
|
b0 = buffer[consumed]
|
|
b1 = buffer[consumed + 1]
|
|
b2 = buffer[consumed + 2]
|
|
b3 = buffer[consumed + 3]
|
|
|
|
if (b2 == 0x04 or b2 == 0x01 or b2 == 0x00) and b3 == 0x46:
|
|
pkt_len = b0 | (b1 << 8)
|
|
pkt_end = consumed + pkt_len
|
|
|
|
if pkt_len < 5 or pkt_len > 8192:
|
|
# Bad length — skip one byte and keep scanning
|
|
consumed += 1
|
|
continue
|
|
|
|
if pkt_end > len(buffer):
|
|
# Don't have the full packet yet — wait for more data
|
|
break
|
|
|
|
# Parse the complete packet
|
|
pkt = buffer[consumed:pkt_end]
|
|
vitals = parse_contec_data(pkt, client_ip)
|
|
if vitals:
|
|
logger.info(
|
|
f"[PORT {port}] Vitals from {client_ip}: "
|
|
f"SpO2={vitals.spo2}%, HR={vitals.heart_rate}bpm "
|
|
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp} "
|
|
f"Temp={vitals.temperature}C"
|
|
)
|
|
await process_vitals(vitals)
|
|
|
|
consumed = pkt_end
|
|
|
|
elif (b2 == 0x04 or b2 == 0x01) and b3 == 0x47:
|
|
# End-of-frame marker — skip 4 bytes
|
|
consumed += 4
|
|
|
|
else:
|
|
# Not a recognised frame start — advance one byte
|
|
consumed += 1
|
|
|
|
# Discard consumed bytes from front of buffer
|
|
buffer = buffer[consumed:]
|
|
|
|
# Safety valve
|
|
if len(buffer) > 32768:
|
|
logger.warning(f"Buffer too large ({len(buffer)} bytes) on port {port}, resetting")
|
|
buffer = b""
|
|
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except ConnectionResetError:
|
|
logger.info(f"Contec CMS7000PLUS connection reset from {client_ip} on port {port}")
|
|
except Exception as e:
|
|
logger.error(f"Error in Contec client handler (port {port}): {e}")
|
|
finally:
|
|
logger.info(f"Closing Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
|
|
display.connection_closed(client_ip, port)
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# Cache to hold merged vitals per device to prevent fragmentation
|
|
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):
|
|
"""
|
|
Log, database, WebSocket broadcast, and REST forward parsed vitals.
|
|
"""
|
|
device_id = vitals.device_id
|
|
now = datetime.now(timezone.utc)
|
|
|
|
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)
|
|
|
|
# 1. Merge new vitals into cached vitals to prevent fragmented entries
|
|
if device_id not in device_cache:
|
|
device_cache[device_id] = vitals
|
|
# Track initial timestamps for present values
|
|
for field, value in vitals.model_dump().items():
|
|
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
|
|
continue
|
|
if value is not None:
|
|
device_field_timestamps[device_id][field] = now
|
|
else:
|
|
cached = device_cache[device_id]
|
|
|
|
# Determine active fields in incoming packet
|
|
active_fields = []
|
|
if vitals.present_fields is not None:
|
|
active_fields = vitals.present_fields
|
|
else:
|
|
for field, value in vitals.model_dump().items():
|
|
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
|
|
continue
|
|
if value is not None:
|
|
active_fields.append(field)
|
|
|
|
# Update cache & record timestamps for present values
|
|
for field in active_fields:
|
|
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:
|
|
# 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 = 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
|
|
display = get_terminal_display()
|
|
display.update_vitals(vitals)
|
|
|
|
logger.info(
|
|
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}"
|
|
)
|
|
|
|
# 3. Broadcast to live WebSockets dashboard in real-time
|
|
try:
|
|
from routers.dashboard import get_ws_manager
|
|
ws_manager = get_ws_manager()
|
|
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. Asynchronous background persistence and cloud forwarding (completely non-blocking, throttled to 1s)
|
|
now = datetime.now(timezone.utc)
|
|
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_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:
|
|
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 "unknown",
|
|
status="active"
|
|
)
|
|
db.add(device)
|
|
else:
|
|
if vitals.ip_address:
|
|
device.ip_address = vitals.ip_address
|
|
device.status = "active"
|
|
device.last_seen = datetime.now(timezone.utc)
|
|
|
|
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()
|
|
|
|
if not patient:
|
|
if vitals.patient_id and vitals.patient_id != "UNKNOWN":
|
|
patient = Patient(
|
|
patient_id=vitals.patient_id,
|
|
mrn=vitals.patient_id,
|
|
name=vitals.patient_name or "Unknown",
|
|
gender=vitals.patient_gender,
|
|
dob=vitals.patient_dob,
|
|
bed_id=vitals.bed_id,
|
|
patient_type=getattr(vitals, "patient_type", None),
|
|
device_id=vitals.device_id
|
|
)
|
|
db.add(patient)
|
|
else:
|
|
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
|
|
if vitals.patient_dob:
|
|
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 or (patient.mrn if patient else "UNKNOWN"),
|
|
timestamp=vitals.timestamp,
|
|
heart_rate=vitals.heart_rate,
|
|
spo2=vitals.spo2,
|
|
systolic_bp=vitals.systolic_bp,
|
|
diastolic_bp=vitals.diastolic_bp,
|
|
map_bp=vitals.map_bp,
|
|
respiratory_rate=vitals.respiratory_rate,
|
|
temperature=vitals.temperature,
|
|
temperature2=vitals.temperature2,
|
|
temperature_diff=vitals.temperature_diff,
|
|
transmitted=False
|
|
)
|
|
db.add(reading)
|
|
db.commit()
|
|
db.refresh(reading)
|
|
reading_id = reading.id
|
|
except Exception as e:
|
|
logger.error(f"Background database operation failed: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
|
|
# 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()
|