import logging from datetime import datetime, timezone from typing import Optional, List import struct from schemas import NormalizedVitals logger = logging.getLogger(__name__) # Contec CMS7000PLUS / CMS8000 / CMS9200PLUS OBX identifiers mapping to schema fields CONTEC_PARAM_MAP = { # Heart Rate / Pulse Rate "HR": "heart_rate", "HEART_RATE": "heart_rate", "ECG_HR": "heart_rate", "PR": "heart_rate", "PULSE_RATE": "heart_rate", "SPO2_PR": "heart_rate", "ECGHR": "heart_rate", "ECG-HR": "heart_rate", "PULSERATE": "heart_rate", # SpO2 "SPO2": "spo2", "SAO2": "spo2", "SPO2_PCT": "spo2", "OXYGEN_SAT": "spo2", "SPO2_VAL": "spo2", "O2SAT": "spo2", # Blood Pressure (NIBP) "NIBP_SYS": "systolic_bp", "NIBP_DIA": "diastolic_bp", "NIBP_MEAN": "map_bp", "NIBP_MAP": "map_bp", "SYS": "systolic_bp", "DIA": "diastolic_bp", "MAP": "map_bp", "NBP_S": "systolic_bp", "NBP_D": "diastolic_bp", "NBP_M": "map_bp", "NIBP-S": "systolic_bp", "NIBP-D": "diastolic_bp", "NIBP-M": "map_bp", "BP_SYS": "systolic_bp", "BP_DIA": "diastolic_bp", "BP_MEAN": "map_bp", # Respiratory Rate "RESP": "respiratory_rate", "RESP_RATE": "respiratory_rate", "RR": "respiratory_rate", "BR": "respiratory_rate", "RESPRATE": "respiratory_rate", "RESP-RATE": "respiratory_rate", # Temperature "TEMP": "temperature", "TEMP1": "temperature", "TEMP2": "temperature", "T1": "temperature", "T2": "temperature", "BODY_TEMP": "temperature", "TEMP-1": "temperature", "TEMP-2": "temperature", "BODYTEMP": "temperature", } # Sentinel value used by Contec to mark an invalid/disconnected parameter INVALID_SENTINEL_U16 = 9999 # 0x270F INVALID_SENTINEL_F32 = 9999.0 def safe_float(value: str) -> Optional[float]: try: return float(value) except (ValueError, TypeError): return None def is_valid_u16(val: Optional[int]) -> bool: return val is not None and val != INVALID_SENTINEL_U16 def is_valid_f32(val: Optional[float]) -> bool: return val is not None and abs(val - INVALID_SENTINEL_F32) > 1.0 def read_u16_le(buf: bytes, offset: int) -> Optional[int]: if offset + 2 <= len(buf): return struct.unpack_from(' Optional[float]: if offset + 4 <= len(buf): return struct.unpack_from(' Optional[NormalizedVitals]: """ Lightweight, native parser for Contec HL7 messages. Splits segments and fields manually, avoiding third-party library dependencies. """ try: # Strip MLLP framing if present (VT=0x0b, FS=0x1c, CR=0x0d) clean_text = raw_text.strip() if clean_text.startswith("\x0b"): clean_text = clean_text[1:] if "\x1c\x0d" in clean_text: clean_text = clean_text.split("\x1c\x0d")[0] # Split into segments by CR or LF segments = [seg.strip() for seg in clean_text.replace("\n", "\r").split("\r") if seg.strip()] if not segments or not segments[0].startswith("MSH"): return None # Parse MSH segment msh_fields = segments[0].split("|") device_id = msh_fields[2] if len(msh_fields) > 2 and msh_fields[2] else f"CMS7000_{client_ip}" patient_id = "UNKNOWN" vitals_dict = { "device_id": device_id, "patient_id": patient_id, "timestamp": datetime.now(timezone.utc), } found_any = False # Parse PID and OBX segments for seg in segments[1:]: fields = seg.split("|") if not fields: continue seg_name = fields[0] if seg_name == "PID": if len(fields) > 3 and fields[3]: patient_id = fields[3].split("^")[0] vitals_dict["patient_id"] = patient_id elif seg_name == "OBX": if len(fields) > 5: obs_id = fields[3].split("^")[0].strip().upper() obs_val = fields[5].strip() if not obs_val or obs_val == "---": continue val_float = safe_float(obs_val) if val_float is None: continue for param_key, mapped_field in CONTEC_PARAM_MAP.items(): if obs_id == param_key or param_key in obs_id: vitals_dict[mapped_field] = val_float found_any = True logger.info(f"Parsed Contec HL7 Vital: {mapped_field} = {val_float} (from {obs_id})") break return NormalizedVitals(**vitals_dict) if found_any else None except Exception as e: logger.error(f"Error parsing Contec HL7 text: {e}") return None def _extract_vitals_block(data: bytes, offset: int, client_ip: str) -> Optional[NormalizedVitals]: """ Extract vitals from a Contec vitals block starting at the given offset. Known Contec CMS7000PLUS binary vitals block layout (confirmed from packet captures): PACKET TYPE B (header bytes [xx][00][04][46] or [xx][yy][04][46]): ----------------------------------------------------------------------- Bytes 8-11 : float32 LE → NIBP Systolic (9999.0 = invalid) Bytes 12-15 : float32 LE → NIBP Diastolic (9999.0 = invalid) Bytes 16-19 : float32 LE → NIBP MAP (9999.0 = invalid) Bytes 20-23 : float32 LE → Temperature 1 (°C, 9999.0 = invalid) Bytes 24-27 : float32 LE → Temperature 2 (°C, 9999.0 = invalid) Vitals summary block (at data[offset] position, found by scanning for SpO2+HR pattern): Offset +0 : uint16 LE → SpO2 (%) (9999 = invalid) Offset +2 : uint16 LE → Heart Rate (bpm)(9999 = invalid, ECG not connected) Offset +4 : uint16 LE → Pulse Rate (bpm)(from SpO2 probe, also 9999 if invalid) Offset +6 : uint16 LE → (SpO2 alarm/threshold) Offset +8 : uint16 LE → (mode flags) Offset +10 : uint16 LE → NIBP Systolic (9999 = invalid) Offset +12 : uint16 LE → NIBP Diastolic (9999 = invalid) """ vitals_dict = { "device_id": f"CMS7000PLUS_{client_ip}", "patient_id": "UNKNOWN", "timestamp": datetime.now(timezone.utc), } found = False spo2 = read_u16_le(data, offset) hr_ecg = read_u16_le(data, offset + 2) hr_spo2 = read_u16_le(data, offset + 4) # SpO2 value: valid range 50–100 if spo2 is not None and 50 <= spo2 <= 100: vitals_dict["spo2"] = float(spo2) found = True # Heart rate: prefer ECG HR; fall back to SpO2-derived PR if ECG invalid if is_valid_u16(hr_ecg) and 20 <= hr_ecg <= 300: vitals_dict["heart_rate"] = float(hr_ecg) found = True elif is_valid_u16(hr_spo2) and 20 <= hr_spo2 <= 300: vitals_dict["heart_rate"] = float(hr_spo2) found = True return NormalizedVitals(**vitals_dict) if found else None def _parse_float_vitals(data: bytes, base_offset: int, vitals_dict: dict) -> bool: """ Parse float32 vitals (NIBP, Temperature) from packet starting at base_offset. Returns True if any valid vitals found. """ found = False nibp_sys = read_f32_le(data, base_offset) nibp_dia = read_f32_le(data, base_offset + 4) nibp_map = read_f32_le(data, base_offset + 8) temp1 = read_f32_le(data, base_offset + 12) temp2 = read_f32_le(data, base_offset + 16) if nibp_sys is not None and is_valid_f32(nibp_sys) and 50.0 < nibp_sys < 300.0: vitals_dict["systolic_bp"] = round(nibp_sys, 1) found = True if nibp_dia is not None and is_valid_f32(nibp_dia) and 20.0 < nibp_dia < 200.0: vitals_dict["diastolic_bp"] = round(nibp_dia, 1) found = True if nibp_map is not None and is_valid_f32(nibp_map) and 20.0 < nibp_map < 250.0: vitals_dict["map_bp"] = round(nibp_map, 1) found = True if temp1 is not None and is_valid_f32(temp1) and 30.0 < temp1 < 45.0: vitals_dict["temperature"] = round(temp1, 1) found = True elif temp2 is not None and is_valid_f32(temp2) and 30.0 < temp2 < 45.0: vitals_dict["temperature"] = round(temp2, 1) found = True return found def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[NormalizedVitals]: """ Scan the data stream for Contec CMS7000PLUS sub-packets. Contec packets follow this framing pattern: [len_lo] [len_hi] [04] [46] [sub_type] [00] [sub_len_lo] [00] ...payload... OR the data is a concatenation of multiple sub-packets separated by [xx][xx][04][47] (end marker?) The key is to find [04][46] marker bytes and parse the packet. """ vitals_dict = { "device_id": f"CMS7000PLUS_{client_ip}", "patient_id": "UNKNOWN", "timestamp": datetime.now(timezone.utc), } found_any = False # Scan through the buffer looking for [04 46] packet markers i = 0 while i < len(data) - 8: # Look for the [04][46] marker which is the packet type indicator if data[i + 2] == 0x04 and data[i + 3] == 0x46: pkt_len = struct.unpack_from(' 50: # The summary block is consistently found around offset 322 in 397-byte packets # Relative to packet start: around pkt_len - 75 summary_offset = i + max(4, pkt_len - 80) if summary_offset + 10 < len(data): # Scan in the last ~100 bytes of the packet for SpO2 pattern scan_end = min(i + pkt_len, len(data) - 4) scan_start = max(i + 4, scan_end - 100) for j in range(scan_start, scan_end - 4, 2): spo2 = read_u16_le(data, j) hr = read_u16_le(data, j + 2) pr = read_u16_le(data, j + 4) # Check if this looks like [SpO2][HR/sentinel][PR] spo2_ok = spo2 is not None and 50 <= spo2 <= 100 # HR can be 9999 (invalid/ECG disconnected) or valid 20-300 hr_or_sentinel = hr is not None and (hr == INVALID_SENTINEL_U16 or 20 <= hr <= 300) pr_ok = pr is not None and (pr == INVALID_SENTINEL_U16 or 20 <= pr <= 300) if spo2_ok and hr_or_sentinel and pr_ok: vitals_dict["spo2"] = float(spo2) found_any = True # Use ECG HR if valid, else use SpO2 PR if hr is not None and hr != INVALID_SENTINEL_U16 and 20 <= hr <= 300: vitals_dict["heart_rate"] = float(hr) elif pr is not None and pr != INVALID_SENTINEL_U16 and 20 <= pr <= 300: vitals_dict["heart_rate"] = float(pr) logger.info( f"Parsed vitals block at offset {j}: " f"SpO2={spo2}, HR={hr}, PR={pr}" ) break # Move past this packet i += max(pkt_len, 4) else: i += 1 return NormalizedVitals(**vitals_dict) if found_any else None def parse_contec_binary_packet(data: bytes, client_ip: str) -> Optional[NormalizedVitals]: """ Decodes the Contec CMS7000PLUS proprietary binary protocol. The monitor sends two alternating TCP packets: - Packet A (991 bytes): ECG waveform + SpO2 waveform + vitals footer - Packet B (397 bytes): Resp waveform + vitals summary + param block Both have the structure: [len_lo][len_hi][04][46][subtype][00][sublen][00]...[payload]... """ try: return _find_vitals_block_in_stream(data, client_ip) except Exception as e: logger.error(f"Error parsing Contec binary packet: {e}") return None def parse_contec_data(raw_data: bytes, client_ip: str) -> Optional[NormalizedVitals]: """ Unified entry point for parsing data from a Contec CMS7000PLUS monitor. Attempts HL7 text parsing first, falling back to proprietary binary parsing. """ if not raw_data: return None # Attempt 1: HL7 Text Decoding try: text = raw_data.decode("utf-8", errors="ignore").strip() if "MSH" in text or "\x0bMSH" in text: vitals = parse_contec_hl7_text(text, client_ip) if vitals: return vitals except Exception: pass # Attempt 2: Binary Parsing (Contec proprietary protocol) vitals = parse_contec_binary_packet(raw_data, client_ip) if vitals: return vitals # Logging fallback logger.warning( f"Data packet from {client_ip} unrecognized. " f"Len: {len(raw_data)}, Hex head: {raw_data[:32].hex()}" ) return None