aruncsh/ContecMonitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: sync all files to fiveparaminte-main including routers and static dirs

Browse Source
This commit is contained in:
rkcsh
2026-09-15 09:43:23 +05:30
parent 48a9288e93
commit 4dc1b8e720
12 changed files with 4197 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fiveparaminte-main/contec_parser.py
+117 -182
View File
@@ -247,7 +247,7 @@ def _extract_vitals_block(data: bytes, offset: int, client_ip: str) -> Optional[
vitals_dict["spo2"] = float(spo2)
found = True
# Heart rate: prefer ECG HR; fall back to SpO2-derived PR if ECG invalid
# Heart rate: ECG HR at offset+2 is primary; SpO2 PR at offset+4 is fallback.
if is_valid_u16(hr_ecg) and 20 <= hr_ecg <= 300:
vitals_dict["heart_rate"] = float(hr_ecg)
found = True
@@ -260,19 +260,19 @@ def _extract_vitals_block(data: bytes, offset: int, client_ip: str) -> Optional[
def _parse_45_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 45-byte packet (Subtype 22) — contains NIBP values and alarm limits.
Parse 45/47-byte packet (Subtype 22) — contains NIBP values and alarm limits.
Verified layout:
[0-1] Packet length LE u16 = 45
[2-3] Marker: 01 46
[0-1] Packet length LE u16 = 45 or 47
[2-3] Marker: 01 46 or 00 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)
[15-16] NIBP Systolic (LE u16, 32767 / 9999 = no measurement / idle)
[17-18] NIBP Diastolic (LE u16, 32767 / 9999 = no measurement / idle)
[19-20] NIBP MAP (LE u16, 32767 / 9999 = no measurement / idle)
"""
if len(data) < 22:
if len(data) < 21:
return False
if "present_fields" not in vitals_dict:
@@ -282,32 +282,24 @@ def _parse_45_byte_packet(data: bytes, vitals_dict: dict) -> bool:
nibp_dia = read_u16_le(data, 17)
nibp_map = read_u16_le(data, 19)
found = False
sentinels = (9999, 32767, 65535, 0)
# 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:
if nibp_sys is not None and nibp_sys not in sentinels and 40 <= nibp_sys <= 260:
vitals_dict["systolic_bp"] = float(nibp_sys)
found = True
else:
vitals_dict["systolic_bp"] = None
if "systolic_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("systolic_bp")
if nibp_dia is not None and nibp_dia != 9999 and 20 < nibp_dia < 200:
if nibp_dia is not None and nibp_dia not in sentinels and 20 <= nibp_dia <= 200:
vitals_dict["diastolic_bp"] = float(nibp_dia)
found = True
else:
vitals_dict["diastolic_bp"] = None
if "diastolic_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("diastolic_bp")
if nibp_map is not None and nibp_map != 9999 and 20 < nibp_map < 250:
if nibp_map is not None and nibp_map not in sentinels and 20 <= nibp_map <= 250:
vitals_dict["map_bp"] = float(nibp_map)
found = True
else:
vitals_dict["map_bp"] = None
if "map_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("map_bp")
return found
return True
def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
@@ -325,144 +317,101 @@ def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
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:
def is_real_temp(t):
# Disconnected sentinel is 9999.0 or 3276.7 (0x7FFF)
return t is not None and is_valid_f32(t) and abs(t - 3276.7) > 1.0 and 20.0 < t < 45.0
if is_real_temp(temp1):
vitals_dict["temperature"] = round(temp1, 1)
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
elif temp2 is not None and is_valid_f32(temp2) and 10.0 < temp2 < 50.0:
elif is_real_temp(temp2):
vitals_dict["temperature"] = round(temp2, 1)
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
elif temp1 is not None or temp2 is not None:
# Sensor disconnected / off
vitals_dict["temperature"] = None
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
found = True
return found
def _parse_spo2_waveform_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse SpO2 & Pleth waveform packet (Subtype 21 for CMS7000PLUS / CMS8500).
SpO2% is located at offset 264. Disconnected sentinel is 65535, 32767, or 255.
Offsets 270+ contain alarm limits (100, 90) and must NOT be parsed as vitals.
"""
if len(data) < 266:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
# Extract Pleth waveform samples (bytes 8..263)
vitals_dict["pleth_wave"] = list(data[8:min(264, len(data))])
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
found = False
# Check SpO2 at standard offset 264
spo2_val = read_u16_le(data, 264)
# If 264 is disconnected sentinel, probe is NOT connected / off-finger
if spo2_val in (None, 65535, 32767, 255, 9999, 0):
vitals_dict["spo2"] = None
vitals_dict["heart_rate"] = None
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
return True
if 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
found = True
# ECG HR is at offset 266 in the 286-byte SpO2 waveform packet footer.
# Offset 268 is the SpO2 HIGH ALARM LIMIT (always 100) — NOT pulse rate!
# So we read ECG HR at 266 only.
ecg_hr = read_u16_le(data, 266)
logger.info(
f"SpO2 pkt: SpO2[264]={spo2_val} ECG_HR[266]={ecg_hr} alarm_hi[268]={read_u16_le(data,268)} alarm_lo[270]={read_u16_le(data,270) if len(data)>=272 else 'N/A'}"
)
if ecg_hr is not None and ecg_hr not in (9999, 65535, 32767, 0) and 20 <= ecg_hr <= 300:
vitals_dict["heart_rate"] = float(ecg_hr)
if "heart_rate" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("heart_rate")
return found
def _parse_286_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 286-byte waveform packet (Subtype 21).
Verified layout:
[8..263] Pleth waveform samples (8-bit unsigned, 256 samples)
[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")
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
vitals_dict.setdefault("spo2", None)
found = False
# Extract Pleth waveform samples (bytes 8..263)
raw_pleth = list(data[8:264])
# Only store if not all-zeroes or all-sentinel (flatline = disconnected)
if any(v not in (0x00, 0x7F, 0x3A, 0x92, 0xFF) for v in raw_pleth[:10]):
vitals_dict["pleth_wave"] = raw_pleth
else:
vitals_dict["pleth_wave"] = raw_pleth # Always send so frontend can show flatline too
# 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
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_288_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 288-byte waveform packet (Subtype 21 for CMS8500).
Layout:
[8..263] Pleth waveform samples (8-bit unsigned, 256 samples)
[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")
if "pleth_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("pleth_wave")
vitals_dict.setdefault("spo2", None)
found = False
# Extract Pleth waveform samples (bytes 8..263)
vitals_dict["pleth_wave"] = list(data[8:264])
# 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
return _parse_spo2_waveform_packet(data, vitals_dict)
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
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
@@ -479,38 +428,27 @@ def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
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")
# Extract ECG waveform samples (bytes 8..903), downsample to ~224 points for smooth real-time rendering
raw_ecg = list(data[8:904])
vitals_dict["ecg_wave"] = raw_ecg[::4]
if "ecg_wave" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("ecg_wave")
vitals_dict.setdefault("heart_rate", None)
found = False
# Extract ECG waveform samples (bytes 8..903), downsample to 200 points for efficiency
raw_ecg = list(data[8:904])
# Downsample by factor of 4 -> ~224 points (still high-resolution enough)
vitals_dict["ecg_wave"] = raw_ecg[::4]
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
if hr_val is not None and hr_val not in (65535, 9999, 0) and 20 <= hr_val <= 300:
# Only use ECG HR if SpO2-probe PR has NOT already set heart_rate.
# SpO2 PR (from the pleth packet) matches the monitor display and wins.
if "heart_rate" not in vitals_dict.get("present_fields", []):
vitals_dict["heart_rate"] = float(hr_val)
vitals_dict.setdefault("present_fields", []).append("heart_rate")
return found
return True
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
Scan the data stream for Contec CMS7000PLUS / CMS8500 sub-packets and parse them.
"""
model = settings.device_models.get(client_ip, settings.monitor_model)
vitals_dict = {
@@ -530,23 +468,20 @@ def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[Normal
pkt_data = data[i:i + pkt_len]
if pkt_len in (45, 47):
_parse_45_byte_packet(pkt_data, vitals_dict)
found_any = True
if _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
if _parse_56_byte_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len in (286, 288, 341):
if _parse_spo2_waveform_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len == 989:
_parse_989_byte_packet(pkt_data, vitals_dict)
found_any = True
if _parse_989_byte_packet(pkt_data, vitals_dict):
found_any = True
elif 100 < pkt_len < 600:
if _parse_spo2_waveform_packet(pkt_data, vitals_dict):
found_any = True
i += max(pkt_len, 4)
else: