aruncsh/ContecMonitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
801def2dc4cc519ace4ef9632f702e058c0f381f
ContecMonitor/contec_parser.py
T

554 lines
19 KiB
Python
Raw Normal View History

Fix CMS8500 SpO2 offset in 288-byte packet
2026-07-13 15:50:53 +05:30
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('<H', buf, offset)[0]
return None
def read_f32_le(buf: bytes, offset: int) -> Optional[float]:
if offset + 4 <= len(buf):
return struct.unpack_from('<f', buf, offset)[0]
return None
def parse_contec_hl7_text(raw_text: str, client_ip: str) -> 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('<H', data, i)[0]
if 30 < pkt_len < 1050 and i + pkt_len <= len(data):
pkt_data = data[i:i + pkt_len]
if pkt_len in (45, 47):
_parse_45_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 56:
_parse_56_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 286:
_parse_286_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 288:
_parse_288_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 341:
_parse_341_byte_packet(pkt_data, vitals_dict)
found_any = True
elif pkt_len == 989:
_parse_989_byte_packet(pkt_data, vitals_dict)
found_any = True
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
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
Reference in New Issue Copy Permalink