import asyncio import logging 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}|{now_str}||ACK^R01|{msg_control_id}|P|2.3.1\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 "" 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}") 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 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) # --------------------------------------------------------------- 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 = {} 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] = {} # 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 field in active_fields: val = getattr(vitals, field) setattr(cached, field, val) device_field_timestamps[device_id][field] = now # Clean up stale fields (older than 15 seconds) 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: setattr(cached, field, None) device_field_timestamps[device_id].pop(field, None) cached.timestamp = vitals.timestamp 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}: " 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() await ws_manager.broadcast_vitals(vitals.model_dump(mode="json")) except Exception as e: logger.debug(f"WebSocket broadcast failed: {e}") # 4. Throttled Database & REST forwarding (at most once every 5 seconds) 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 last_db_write[device_id] = now if should_write_db: # Save to SQLite database and REST forward db = SessionLocal() try: # 1. Device tracking device = db.query(Device).filter(Device.device_id == vitals.device_id).first() if not device: device = Device( device_id=vitals.device_id, ip_address=vitals.ip_address or "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) # 2. Patient tracking patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first() if not patient: patient = Patient(patient_id=vitals.patient_id, name="Unknown") db.add(patient) # 3. Create Reading Entry 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, 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() except Exception as e: logger.error(f"Database operation failed: {e}") db.rollback() finally: db.close()