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 # 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 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 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 reading = VitalReading( device_id=vitals.device_id, patient_id=vitals.patient_id, timestamp=vitals.timestamp, heart_rate=vitals.heart_rate, spo2=vitals.spo2, systolic_bp=vitals.systolic_bp, diastolic_bp=vitals.diastolic_bp, map_bp=vitals.map_bp, respiratory_rate=vitals.respiratory_rate, temperature=vitals.temperature, 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()