From 5d418c8eb0665c2d5c3655454c4550b0c98eb3c3 Mon Sep 17 00:00:00 2001 From: prathiyuman Date: Mon, 13 Jul 2026 15:50:53 +0530 Subject: [PATCH] Fix CMS8500 SpO2 offset in 288-byte packet --- contec_parser.py | 553 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 553 insertions(+) create mode 100644 contec_parser.py diff --git a/contec_parser.py b/contec_parser.py new file mode 100644 index 0000000..966ebec --- /dev/null +++ b/contec_parser.py @@ -0,0 +1,553 @@ +import logging +from datetime import datetime, timezone +from typing import Optional, List +import struct + +from schemas import NormalizedVitals +from config import settings + +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 and val != 255 and val != 65535 + + +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 + model = settings.device_models.get(client_ip, settings.monitor_model) + msh_fields = segments[0].split("|") + device_id = msh_fields[2] if len(msh_fields) > 2 and msh_fields[2] else f"{model}_{client_ip}" + + patient_id = "UNKNOWN" + vitals_dict = { + "device_id": device_id, + "patient_id": patient_id, + "timestamp": datetime.now(timezone.utc), + "ip_address": client_ip, + } + + 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_val = fields[5].strip() + + if not obs_val or obs_val == "---": + continue + + val_float = safe_float(obs_val) + if val_float is None: + continue + + # Split OBX-3 (Observation Identifier) by '^' and check all parts for parameter matches + obs_parts = [p.strip().upper() for p in fields[3].split("^") if p.strip()] + + matched = False + for param_key, mapped_field in CONTEC_PARAM_MAP.items(): + for part in obs_parts: + if param_key == part or param_key in part: + vitals_dict[mapped_field] = val_float + found_any = True + logger.info(f"Parsed Contec HL7 Vital: {mapped_field} = {val_float} (from OBX-3: {fields[3]})") + matched = True + break + if matched: + 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) + """ + model = settings.device_models.get(client_ip, settings.monitor_model) + vitals_dict = { + "device_id": f"{model}_{client_ip}", + "patient_id": "UNKNOWN", + "timestamp": datetime.now(timezone.utc), + "ip_address": client_ip, + } + 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_45_byte_packet(data: bytes, vitals_dict: dict) -> bool: + """ + Parse 45-byte packet (Subtype 22) — contains NIBP values and alarm limits. + + Verified layout: + [0-1] Packet length LE u16 = 45 + [2-3] Marker: 01 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) + """ + if len(data) < 22: + return False + + if "present_fields" not in vitals_dict: + vitals_dict["present_fields"] = [] + + nibp_sys = read_u16_le(data, 15) + nibp_dia = read_u16_le(data, 17) + nibp_map = read_u16_le(data, 19) + + found = False + + # 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: + vitals_dict["systolic_bp"] = float(nibp_sys) + found = True + else: + vitals_dict["systolic_bp"] = None + + if nibp_dia is not None and nibp_dia != 9999 and 20 < nibp_dia < 200: + vitals_dict["diastolic_bp"] = float(nibp_dia) + found = True + else: + vitals_dict["diastolic_bp"] = None + + if nibp_map is not None and nibp_map != 9999 and 20 < nibp_map < 250: + vitals_dict["map_bp"] = float(nibp_map) + found = True + else: + vitals_dict["map_bp"] = None + + return found + + +def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool: + """ + Parse 56-byte packet (Subtype 23) — contains float32 Temperature values. + + Verified layout: + [8-11] Temperature 1 (float32) + [12-15] Temperature 2 (float32) + [20-23] Temp 1 Alarm High Limit (float32) + [24-27] Temp 1 Alarm Low Limit (float32) + """ + if len(data) < 16: + return False + + 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 + + return found + + +def _parse_286_byte_packet(data: bytes, vitals_dict: dict) -> bool: + """ + Parse 286-byte waveform packet (Subtype 21). + + Verified layout at tail: + [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") + vitals_dict.setdefault("spo2", None) + + found = False + + # 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 + + +def _parse_288_byte_packet(data: bytes, vitals_dict: dict) -> bool: + """ + Parse 288-byte waveform packet (Subtype 21 for CMS8500). + + Layout: + [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") + vitals_dict.setdefault("spo2", None) + + found = False + + # 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 + + +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 + + +def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool: + """ + Parse 989-byte waveform packet (Subtype 20). + + Verified layout: + [904-905] ECG Heart Rate (LE u16) + 65535 or 9999 = invalid/disconnected + """ + if len(data) < 906: + return False + + 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") + vitals_dict.setdefault("heart_rate", None) + + found = False + + 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: + vitals_dict["heart_rate"] = float(hr_val) + 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 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 + """ + model = settings.device_models.get(client_ip, settings.monitor_model) + vitals_dict = { + "device_id": f"{model}_{client_ip}", + "patient_id": "UNKNOWN", + "timestamp": datetime.now(timezone.utc), + "ip_address": client_ip, + } + found_any = False + + i = 0 + while i < len(data) - 8: + if (data[i + 2] == 0x04 or data[i + 2] == 0x01 or data[i + 2] == 0x00) and data[i + 3] == 0x46: + pkt_len = struct.unpack_from(' 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 + else: + logger.warning(f"HL7 message detected but failed to parse. Content: {repr(text)}") + except Exception as e: + logger.warning(f"Failed decoding/parsing HL7: {e}") + + # 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