From 4dc1b8e7208f2ca53de6863bf75fce02ab704aef Mon Sep 17 00:00:00 2001 From: Arun Date: Tue, 15 Sep 2026 09:43:23 +0530 Subject: [PATCH] fix: sync all files to fiveparaminte-main including routers and static dirs --- fiveparaminte-main/Dockerfile | 1 + fiveparaminte-main/api_client.py | 58 +- fiveparaminte-main/background_tasks.py | 92 +- fiveparaminte-main/contec_parser.py | 299 +-- fiveparaminte-main/contec_server.py | 433 ++- fiveparaminte-main/database.py | 5 +- fiveparaminte-main/main.py | 34 +- fiveparaminte-main/models.py | 1 + fiveparaminte-main/render.yaml | 2 + fiveparaminte-main/routers/__init__.py | 1 + fiveparaminte-main/routers/dashboard.py | 470 ++++ fiveparaminte-main/static/dashboard.html | 3120 ++++++++++++++++++++++ 12 files changed, 4197 insertions(+), 319 deletions(-) create mode 100644 fiveparaminte-main/routers/__init__.py create mode 100644 fiveparaminte-main/routers/dashboard.py create mode 100644 fiveparaminte-main/static/dashboard.html diff --git a/fiveparaminte-main/Dockerfile b/fiveparaminte-main/Dockerfile index 1d6b037..15dd4be 100644 --- a/fiveparaminte-main/Dockerfile +++ b/fiveparaminte-main/Dockerfile @@ -1,6 +1,7 @@ FROM python:3.12-slim WORKDIR /app +ENV PYTHONPATH=/app # Install dependencies COPY requirements.txt . diff --git a/fiveparaminte-main/api_client.py b/fiveparaminte-main/api_client.py index 7882a8a..12fd250 100644 --- a/fiveparaminte-main/api_client.py +++ b/fiveparaminte-main/api_client.py @@ -6,6 +6,17 @@ from config import settings logger = logging.getLogger(__name__) +_http_client: Optional[httpx.AsyncClient] = None + +def get_http_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None or _http_client.is_closed: + _http_client = httpx.AsyncClient( + timeout=httpx.Timeout(connect=3.0, read=5.0, write=3.0, pool=3.0), + limits=httpx.Limits(max_keepalive_connections=5, max_connections=10) + ) + return _http_client + async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool: """ Attempts to POST the normalized vitals to the configured target API @@ -15,6 +26,13 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool: import os headers = {"Content-Type": "application/json"} + # Strip heavy wave sample arrays before REST transmission (cloud APIs only ingest vital signs) + payload = vitals.model_dump(mode="json") + payload.pop("ecg_wave", None) + payload.pop("pleth_wave", None) + + client = get_http_client() + # 1. Forward to primary target API (e.g. Injomo Dev Workflow) primary_url = settings.target_api_url primary_success = False @@ -23,17 +41,17 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool: primary_headers = headers.copy() if settings.api_token: primary_headers["Authorization"] = f"Bearer {settings.api_token}" - async with httpx.AsyncClient(timeout=10.0) as client: - response = await client.post( - primary_url, - json=vitals.model_dump(mode="json"), - headers=primary_headers - ) - response.raise_for_status() - logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}") - primary_success = True + response = await client.post( + primary_url, + json=payload, + headers=primary_headers + ) + response.raise_for_status() + logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}") + primary_success = True except Exception as e: - logger.error(f"Error forwarding to primary API: {e}") + err_msg = str(e).strip() or "timeout / connection reset" + logger.warning(f"Error forwarding to primary API: {type(e).__name__} ({err_msg})") # 2. Forward to Render Cloud Server render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals" @@ -48,16 +66,16 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool: if settings.api_token: url_with_key = f"{render_url}?api_key={settings.api_token}" - async with httpx.AsyncClient(timeout=10.0) as client: - response = await client.post( - url_with_key, - json=vitals.model_dump(mode="json"), - headers=headers - ) - response.raise_for_status() - logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}") - render_success = True + response = await client.post( + url_with_key, + json=payload, + headers=headers + ) + response.raise_for_status() + logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}") + render_success = True except Exception as e: - logger.error(f"Error forwarding to Render Cloud: {e}") + err_msg = str(e).strip() or "timeout / connection reset" + logger.warning(f"Error forwarding to Render Cloud: {type(e).__name__} ({err_msg})") return primary_success or render_success diff --git a/fiveparaminte-main/background_tasks.py b/fiveparaminte-main/background_tasks.py index 6130a38..f27b5c4 100644 --- a/fiveparaminte-main/background_tasks.py +++ b/fiveparaminte-main/background_tasks.py @@ -17,50 +17,64 @@ async def retry_failed_transmissions(): while True: try: + # 1. Fetch untransmitted readings in a quick DB session and close it immediately db = SessionLocal() + readings_to_retry = [] try: - # Get readings that haven't been transmitted - failed_readings = db.query(VitalReading).filter(VitalReading.transmitted == False).limit(50).all() - - for reading in failed_readings: - vitals = NormalizedVitals( - device_id=reading.device_id, - patient_id=reading.patient_id, - timestamp=reading.timestamp, - heart_rate=reading.heart_rate, - spo2=reading.spo2, - systolic_bp=reading.systolic_bp, - diastolic_bp=reading.diastolic_bp, - map_bp=reading.map_bp, - respiratory_rate=reading.respiratory_rate, - temperature=reading.temperature - ) - - success = await forward_vitals_to_api(vitals) - - if success: - reading.transmitted = True - log_entry = TransmissionLog( - reading_id=reading.id, - status="success", - response_code=200 + failed_readings = db.query(VitalReading).filter(VitalReading.transmitted == False).order_by(VitalReading.id.desc()).limit(20).all() + for r in failed_readings: + readings_to_retry.append({ + "id": r.id, + "vitals": NormalizedVitals( + device_id=r.device_id, + patient_id=r.patient_id, + timestamp=r.timestamp, + heart_rate=r.heart_rate, + spo2=r.spo2, + systolic_bp=r.systolic_bp, + diastolic_bp=r.diastolic_bp, + map_bp=r.map_bp, + respiratory_rate=r.respiratory_rate, + temperature=r.temperature ) - db.add(log_entry) - logger.info(f"Successfully re-transmitted reading {reading.id}") - else: - # Log the failure but don't mark as transmitted - log_entry = TransmissionLog( - reading_id=reading.id, - status="failed", - error_message="Retry failed" - ) - db.add(log_entry) - - if failed_readings: - db.commit() + }) finally: db.close() - + + # 2. Forward outside of any DB transaction + successful_ids = [] + failed_ids = [] + consecutive_failures = 0 + + for item in readings_to_retry: + success = await forward_vitals_to_api(item["vitals"]) + if success: + successful_ids.append(item["id"]) + consecutive_failures = 0 + else: + failed_ids.append(item["id"]) + consecutive_failures += 1 + # If external API is unresponsive, stop this batch immediately so we don't spam + if consecutive_failures >= 3: + logger.warning("External API appears unreachable during retry; pausing retry queue.") + break + + # 3. Update status in a quick DB session + if successful_ids or failed_ids: + db = SessionLocal() + try: + for rid in successful_ids: + db.query(VitalReading).filter(VitalReading.id == rid).update({"transmitted": True}) + db.add(TransmissionLog(reading_id=rid, status="success", response_code=200)) + for rid in failed_ids: + db.add(TransmissionLog(reading_id=rid, status="failed", error_message="Retry failed")) + db.commit() + except Exception as e: + logger.error(f"Failed updating retry transmission status: {e}") + db.rollback() + finally: + db.close() + except Exception as e: logger.error(f"Error in retry background task: {e}") diff --git a/fiveparaminte-main/contec_parser.py b/fiveparaminte-main/contec_parser.py index 95b7ad2..4116255 100644 --- a/fiveparaminte-main/contec_parser.py +++ b/fiveparaminte-main/contec_parser.py @@ -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: diff --git a/fiveparaminte-main/contec_server.py b/fiveparaminte-main/contec_server.py index 12f18b6..44372c0 100644 --- a/fiveparaminte-main/contec_server.py +++ b/fiveparaminte-main/contec_server.py @@ -1,5 +1,6 @@ import asyncio import logging +import re from datetime import datetime, timezone from contec_parser import parse_contec_data @@ -34,16 +35,195 @@ def generate_ack(received_msg: str) -> str: 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_msh = ( + f"MSH|^~\\&|{rec_app}|{rec_fac}|{sending_app}|{sending_fac}" + f"|{now_str}||ACK|{now_str}|P|2.3\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 "" +TELEMETRY_WORDS = { + 'sensor', 'off', 'lead', 'leads', 'fall', 'searching', 'pulse', + 'le', 'le1', 'le2', 'le3', 'le4', 'le5', 'alarm', 'probe', + 'cuff', 'error', 'leak', 'motion', 'weak', 'signal', 'learning', 'arr', + 'asystole', 'vfib', 'vtac', 'pvc', 'brady', 'tachy', 'adult', 'pediatric', + 'neonate', 'unknown', 'none', 'clear', 'default', 'demo', 't1', 't2', + 'spo2', 'ecg', 'nibp', 'resp', 'temp', 'hr', 'pr', 'map', 'dia', 'sys' +} + + +def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict: + """ + Extracts real patient demographics from Contec CMS8500 / CMS7000 binary packets. + Only extracts clean, legitimate patient fields — strictly rejects sensor alarms, + waveform noise, and hardware telemetry. + """ + clean = [] + for m in re.finditer(rb'[a-zA-Z0-9_\-\.]{2,}', data): + s = m.group(0).decode('latin1', errors='ignore').strip() + if not s or len(s) < 2: + continue + s_lower = s.lower() + if any(kw in s_lower for kw in TELEMETRY_WORDS): + continue + if len(set(s)) == 1: + continue + if set(s).issubset({'.', '-', '_', '/', '\\'}): + continue + if len(s) > 8 and re.match(r'^[A-Za-z]+$', s): + diffs = [abs(ord(s[i]) - ord(s[i-1])) for i in range(1, len(s))] + if sum(1 for d in diffs if d <= 2) / len(diffs) > 0.6: + continue + if re.match(r'^\.\d+$', s): + continue + clean.append(s) + + if not clean: + return None + + name = None + mrn = None + dob = None + gender = None + + for s in clean: + s_up = s.upper() + s_low = s.lower() + if s_low in TELEMETRY_WORDS: + continue + + if s_up in ('MALE', 'FEMALE') and not gender: + gender = 'Male' if s_up == 'MALE' else 'Female' + continue + elif s_up in ('M', 'F') and not gender: + gender = 'Male' if s_up == 'M' else 'Female' + continue + + # Date of birth: 4-digit birth year between 1920 and 2024, or YYYY-MM-DD + if re.match(r'^(19\d\d|20[01]\d|202[0-4])$', s) and not dob: + dob = s + continue + elif re.match(r'^\d{4}[-/]\d{2}[-/]\d{2}$', s) and not dob: + dob = s + continue + + # Name candidate: strictly at least 3 letters, has a vowel, not telemetry + if re.search(r'^[A-Za-z\s\.\-]{3,30}$', s) and re.search(r'[aeiouyAEIOUY]', s) and not name: + if s_low not in TELEMETRY_WORDS: + name = s + continue + + # MRN candidate: numeric or alphanumeric ID, at least 2 chars, not a year, not telemetry + if not mrn and re.match(r'^[A-Za-z0-9\-]{2,15}$', s) and s_low not in TELEMETRY_WORDS: + if not (s.isdigit() and 1920 <= int(s) <= 2030): + mrn = s + continue + + # Only accept when BOTH real name and real MRN/DOB are present together + if name and (mrn or dob): + return { + "name": name, + "mrn": mrn or name, + "patient_id": mrn or name, + "dob": dob, + "gender": gender, + } + return None + + +async def apply_auto_demographics(client_ip: str, demo: dict): + from config import settings + model = settings.device_models.get(client_ip, settings.monitor_model) + device_id = f"{model}_{client_ip}" + name = demo.get("name") + mrn = demo.get("mrn") + dob = demo.get("dob") + gender = demo.get("gender") + + current = device_patient_map.get(device_id, {}) + if current.get("name") == name and current.get("mrn") == mrn and current.get("dob") == dob and current.get("gender") == gender: + return + + logger.info(f"*** REAL PATIENT DEMOGRAPHICS DETECTED FROM MACHINE ({client_ip}): Name={name}, MRN={mrn}, DOB={dob}, Gender={gender} ***") + + bed_id = "1" if "192.168.100.106" in client_ip else "2" + p_data = { + "patient_id": mrn, + "mrn": mrn, + "name": name, + "gender": gender, + "dob": dob, + "bed_id": bed_id + } + device_patient_map[device_id] = p_data + + if device_id in device_cache: + cached = device_cache[device_id] + cached.patient_id = mrn + cached.patient_name = name + cached.patient_gender = gender + cached.patient_dob = dob + cached.bed_id = bed_id + + db = SessionLocal() + try: + db.query(Patient).filter(Patient.device_id == device_id).update({"device_id": None}) + patient = db.query(Patient).filter(Patient.mrn == mrn).first() + if not patient: + patient = Patient( + patient_id=mrn, + mrn=mrn, + name=name or "Unknown", + gender=gender, + dob=dob, + bed_id=bed_id, + device_id=device_id + ) + db.add(patient) + else: + patient.patient_id = mrn + patient.mrn = mrn + if name: + patient.name = name + if gender: + patient.gender = gender + if dob: + patient.dob = dob + patient.bed_id = bed_id + patient.device_id = device_id + db.commit() + except Exception as e: + logger.error(f"Error persisting auto patient: {e}") + db.rollback() + finally: + db.close() + + try: + from routers.dashboard import get_ws_manager + ws_manager = get_ws_manager() + await ws_manager.broadcast_vitals({ + "event": "patient_assigned", + "device_id": device_id, + "patient_id": mrn, + "mrn": mrn, + "patient_name": name, + "name": name, + "patient_gender": gender, + "gender": gender, + "patient_dob": dob, + "dob": dob, + "bed_id": bed_id + }) + except Exception as e: + logger.debug(f"Failed to broadcast auto patient: {e}") + + async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int): """ TCP server handler for Contec CMS7000PLUS client connections. @@ -52,6 +232,13 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str client_ip, client_port = writer.get_extra_info('peername') logger.info(f"Accepted Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}") + sock = writer.get_extra_info('socket') + if sock: + try: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except Exception: + pass + display = get_terminal_display() display.connection_opened(client_ip, port) @@ -64,8 +251,10 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str logger.info(f"Contec client {client_ip}:{client_port} on port {port} sent EOF (0 bytes)") break + if client_ip != "127.0.0.1": + logger.info(f"TCP READ from {client_ip}: {len(data)} bytes, hex: {data[:24].hex()}") + 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) @@ -166,6 +355,48 @@ device_cache = {} device_field_timestamps = {} # {device_id: {field_name: datetime}} last_db_write = {} +# In-memory mapping of device_id -> patient demographic dict: +# {"mrn": "...", "name": "...", "dob": "...", "gender": "...", "bed_id": "..."} +device_patient_map = {} + + +def get_assigned_patient(device_id: str): + """Retrieve patient demographics assigned to device_id (cached or from DB).""" + if device_id in device_patient_map: + return device_patient_map[device_id] + + db = SessionLocal() + try: + patient = db.query(Patient).filter(Patient.device_id == device_id).first() + if patient: + p_data = { + "patient_id": patient.mrn or patient.patient_id, + "mrn": patient.mrn or patient.patient_id, + "name": patient.name or "Unknown", + "gender": patient.gender, + "dob": patient.dob, + "bed_id": patient.bed_id, + } + device_patient_map[device_id] = p_data + return p_data + except Exception as e: + logger.debug(f"Error querying patient for device {device_id}: {e}") + finally: + db.close() + return None + + +def set_assigned_patient(device_id: str, patient_data: dict): + """Update in-memory mapping and active device cache for device_id.""" + device_patient_map[device_id] = patient_data + if device_id in device_cache: + cached = device_cache[device_id] + cached.patient_id = patient_data.get("mrn") or patient_data.get("patient_id") or "UNKNOWN" + cached.patient_name = patient_data.get("name") + cached.patient_gender = patient_data.get("gender") + cached.patient_dob = patient_data.get("dob") + cached.bed_id = patient_data.get("bed_id") + async def process_vitals(vitals): """ @@ -177,6 +408,25 @@ async def process_vitals(vitals): if device_id not in device_field_timestamps: device_field_timestamps[device_id] = {} + # Check for assigned patient demographics + assigned = get_assigned_patient(device_id) + if (vitals.patient_id == "UNKNOWN" or not vitals.patient_name or vitals.patient_name == "Unknown") and assigned: + vitals.patient_id = assigned.get("mrn") or assigned.get("patient_id") or "UNKNOWN" + vitals.patient_name = assigned.get("name") + vitals.patient_gender = assigned.get("gender") + vitals.patient_dob = assigned.get("dob") + vitals.bed_id = assigned.get("bed_id") + elif vitals.patient_name and vitals.patient_name != "Unknown": + # Monitor sent fresh demographics via HL7 PID segment! + device_patient_map[device_id] = { + "patient_id": vitals.patient_id, + "mrn": vitals.patient_id, + "name": vitals.patient_name, + "gender": vitals.patient_gender, + "dob": vitals.patient_dob, + "bed_id": vitals.bed_id, + } + # Check if incoming packet has fresh waveforms has_fresh_waveforms = (vitals.ecg_wave is not None) or (vitals.pleth_wave is not None) @@ -205,18 +455,34 @@ async def process_vitals(vitals): # Update cache & record timestamps for field in active_fields: - val = getattr(vitals, field) + val = getattr(vitals, field, None) setattr(cached, field, val) - device_field_timestamps[device_id][field] = now + if val is not None: + device_field_timestamps[device_id][field] = now + else: + device_field_timestamps[device_id].pop(field, None) - # Clean up stale fields (older than 15 seconds) + # Clean up stale fields: + # Live vitals (HR, SpO2, Temp, RR) have a 60-second window because monitor cycles take ~24s + # NIBP (BP, MAP) has a 1-hour window because it is measured intermittently 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: + timeout = 3600.0 if field in ("systolic_bp", "diastolic_bp", "map_bp") else 60.0 + if (now - last_updated).total_seconds() > timeout: setattr(cached, field, None) device_field_timestamps[device_id].pop(field, None) cached.timestamp = vitals.timestamp + if vitals.patient_id and vitals.patient_id != "UNKNOWN": + cached.patient_id = vitals.patient_id + if vitals.patient_name: + cached.patient_name = vitals.patient_name + if vitals.patient_gender: + cached.patient_gender = vitals.patient_gender + if vitals.patient_dob: + cached.patient_dob = vitals.patient_dob + if vitals.bed_id: + cached.bed_id = vitals.bed_id vitals = cached # 2. Update console display banner @@ -224,7 +490,7 @@ async def process_vitals(vitals): display.update_vitals(vitals) logger.info( - f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id}: " + f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id} ({vitals.patient_name}): " f"HR={vitals.heart_rate}, SpO2={vitals.spo2}, " f"BP={vitals.systolic_bp}/{vitals.diastolic_bp}, " f"RR={vitals.respiratory_rate}, Temp={vitals.temperature}" @@ -234,23 +500,40 @@ async def process_vitals(vitals): try: from routers.dashboard import get_ws_manager ws_manager = get_ws_manager() - await ws_manager.broadcast_vitals(vitals.model_dump(mode="json")) + vitals_dict = vitals.model_dump(mode="json") + vitals_dict["mrn"] = vitals.patient_id + vitals_dict["name"] = vitals.patient_name + vitals_dict["gender"] = vitals.patient_gender + vitals_dict["dob"] = vitals.patient_dob + vitals_dict["bed_id"] = vitals.bed_id + await ws_manager.broadcast_vitals(vitals_dict) except Exception as e: logger.debug(f"WebSocket broadcast failed: {e}") - # 4. Throttled Database & REST forwarding (at most once every 5 seconds) - # Note: If we have fresh waveforms, we bypass throttle for REST forwarding to keep cloud graphs real-time + # 4. Asynchronous background persistence and cloud forwarding (completely non-blocking, throttled to 1s) 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 + should_persist = False + if device_id not in last_db_write or (now - last_db_write[device_id]).total_seconds() >= 1.0: + should_persist = True last_db_write[device_id] = now - if should_write_db: - # Save to SQLite database and REST forward + if should_persist: + asyncio.create_task(_persist_and_forward_bg(vitals, should_persist)) + + # Clear waveforms from cache to prevent repetition in subsequent non-waveform sub-packets + if device_id in device_cache: + device_cache[device_id].ecg_wave = None + device_cache[device_id].pleth_wave = None + + +async def _persist_and_forward_bg(vitals, should_persist): + """Background worker for DB persistence and external cloud forwarding.""" + if not should_persist: + return + reading_id = None + if should_persist: db = SessionLocal() try: - # 1. Device tracking device = db.query(Device).filter(Device.device_id == vitals.device_id).first() if not device: device = Device( @@ -264,33 +547,36 @@ async def process_vitals(vitals): device.ip_address = vitals.ip_address device.status = "active" device.last_seen = datetime.now(timezone.utc) - - # 2. Patient tracking — store/update demographics from HL7 - if vitals.patient_id == "UNKNOWN": - last_known_reading = ( - db.query(VitalReading) - .filter(VitalReading.device_id == vitals.device_id) - .filter(VitalReading.patient_id != "UNKNOWN") - .order_by(VitalReading.timestamp.desc()) - .first() - ) - if last_known_reading: - vitals.patient_id = last_known_reading.patient_id - patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first() + patient = None + if vitals.patient_id and vitals.patient_id != "UNKNOWN": + patient = db.query(Patient).filter( + (Patient.mrn == vitals.patient_id) | (Patient.patient_id == vitals.patient_id) + ).first() + + if not patient and vitals.device_id: + patient = db.query(Patient).filter( + Patient.device_id == vitals.device_id, + Patient.patient_id != "UNKNOWN" + ).first() + if not patient: - patient = Patient( - patient_id=vitals.patient_id, - mrn=vitals.patient_id, - name=vitals.patient_name or "Unknown", - gender=vitals.patient_gender, - dob=vitals.patient_dob, - bed_id=vitals.bed_id, - ) - db.add(patient) + if vitals.patient_id and vitals.patient_id != "UNKNOWN": + patient = Patient( + patient_id=vitals.patient_id, + mrn=vitals.patient_id, + name=vitals.patient_name or "Unknown", + gender=vitals.patient_gender, + dob=vitals.patient_dob, + bed_id=vitals.bed_id, + device_id=vitals.device_id + ) + db.add(patient) else: - # Refresh demographics whenever monitor sends updated PID fields - if vitals.patient_name: + if vitals.patient_id and vitals.patient_id != "UNKNOWN": + patient.patient_id = vitals.patient_id + patient.mrn = vitals.patient_id + if vitals.patient_name and vitals.patient_name != "Unknown": patient.name = vitals.patient_name if vitals.patient_gender: patient.gender = vitals.patient_gender @@ -298,9 +584,9 @@ async def process_vitals(vitals): patient.dob = vitals.patient_dob if vitals.bed_id: patient.bed_id = vitals.bed_id + if vitals.device_id: + patient.device_id = vitals.device_id - - # 3. Create Reading Entry reading = VitalReading( device_id=vitals.device_id, patient_id=vitals.patient_id, @@ -317,40 +603,35 @@ async def process_vitals(vitals): 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() - + reading_id = reading.id except Exception as e: - logger.error(f"Database operation failed: {e}") + logger.error(f"Background database operation failed: {e}") db.rollback() finally: db.close() - elif has_fresh_waveforms: - # Bypass throttle for REST forwarding to keep cloud graphs real-time - try: - await forward_vitals_to_api(vitals) - except Exception as e: - logger.error(f"Real-time REST forwarding failed: {e}") - # Clear waveforms from cache to prevent repetition in subsequent non-waveform sub-packets - if device_id in device_cache: - device_cache[device_id].ecg_wave = None - device_cache[device_id].pleth_wave = None + # DB connection is now completely closed. Awaiting HTTP requests will never exhaust DB connections! + success = False + if should_persist: + try: + success = await forward_vitals_to_api(vitals) + except Exception as e: + logger.warning(f"Background REST forward failed: {type(e).__name__}: {e}") + + # Record transmission log in a separate, brief connection + if reading_id is not None: + db = SessionLocal() + try: + log_entry = TransmissionLog( + reading_id=reading_id, + status="success" if success else "failed", + response_code=200 if success else 500, + error_message=None if success else "Immediate REST forward failed" + ) + db.add(log_entry) + db.commit() + except Exception as e: + logger.error(f"Failed to record transmission log: {e}") + db.rollback() + finally: + db.close() diff --git a/fiveparaminte-main/database.py b/fiveparaminte-main/database.py index cbde1b6..770383a 100644 --- a/fiveparaminte-main/database.py +++ b/fiveparaminte-main/database.py @@ -1,9 +1,12 @@ from sqlalchemy import create_engine +from sqlalchemy.pool import NullPool from sqlalchemy.orm import sessionmaker, declarative_base from config import settings engine = create_engine( - settings.database_url, connect_args={"check_same_thread": False} + settings.database_url, + connect_args={"check_same_thread": False}, + poolclass=NullPool ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) diff --git a/fiveparaminte-main/main.py b/fiveparaminte-main/main.py index 617f971..8683ee7 100644 --- a/fiveparaminte-main/main.py +++ b/fiveparaminte-main/main.py @@ -105,7 +105,8 @@ async def lifespan(app: FastAPI): print("\n" + "=" * 65) print(f" Contec {model} Patient Monitor - Vital Signs Forwarder") print("=" * 65) - print(f" DASHBOARD: http://localhost:{port}/api/dashboard") + print(f" DASHBOARD: http://localhost:{port}/dashboard") + print(f" http://localhost:{port}/api/dashboard") print(f" API Docs: http://localhost:{port}/docs") print(f" Contec Ports: {settings.contec_ports}") print(f" Monitor Model: {model}") @@ -124,10 +125,34 @@ async def lifespan(app: FastAPI): # Startup: Start the background retry task retry_task = asyncio.create_task(retry_failed_transmissions()) + # Startup: Start bore tunnel for remote monitors if available and not already running + bore_proc = None + bore_exe = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bore.exe") + if os.path.exists(bore_exe): + import subprocess + try: + check_out = subprocess.check_output('tasklist /FI "IMAGENAME eq bore.exe"', shell=True).decode() + if "bore.exe" not in check_out: + bore_proc = subprocess.Popen( + [bore_exe, "local", "511", "--to", "159.223.110.159", "--port", "61501"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL + ) + logger.info(" [OK] Bore tunnel active for remote hotspot monitors (159.223.110.159:61501)") + else: + logger.info(" [INFO] Bore tunnel is already running") + except Exception as e: + logger.debug(f"Could not check/start bore tunnel: {e}") + yield # Shutdown logger.info("Shutting down servers...") + if bore_proc: + try: + bore_proc.terminate() + except Exception: + pass retry_task.cancel() for server in servers: server.close() @@ -156,6 +181,13 @@ app.add_middleware( # Include dashboard routes app.include_router(dashboard.router, prefix="/api") +# Also serve dashboard at root and /dashboard directly for convenience +from fastapi.responses import HTMLResponse +@app.get("/", response_class=HTMLResponse) +@app.get("/dashboard", response_class=HTMLResponse) +def serve_dashboard_root(): + return dashboard.serve_dashboard() + # Serve static folder from fastapi.staticfiles import StaticFiles app.mount("/static", StaticFiles(directory="static"), name="static") diff --git a/fiveparaminte-main/models.py b/fiveparaminte-main/models.py index 9194976..a4d02e2 100644 --- a/fiveparaminte-main/models.py +++ b/fiveparaminte-main/models.py @@ -19,6 +19,7 @@ class Patient(Base): gender = Column(String, nullable=True) # M/F/U from PID-8 dob = Column(String, nullable=True) # YYYYMMDD from PID-7 bed_id = Column(String, nullable=True) # bed from PV1-3 + device_id = Column(String, nullable=True, index=True) # assigned machine / device_id class VitalReading(Base): __tablename__ = "vital_readings" diff --git a/fiveparaminte-main/render.yaml b/fiveparaminte-main/render.yaml index c2862c9..93362c8 100644 --- a/fiveparaminte-main/render.yaml +++ b/fiveparaminte-main/render.yaml @@ -5,6 +5,8 @@ services: buildCommand: pip install -r requirements.txt startCommand: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT envVars: + - key: PYTHONPATH + value: /opt/render/project/src - key: PORT value: 10000 - key: PYTHON_VERSION diff --git a/fiveparaminte-main/routers/__init__.py b/fiveparaminte-main/routers/__init__.py new file mode 100644 index 0000000..5f24585 --- /dev/null +++ b/fiveparaminte-main/routers/__init__.py @@ -0,0 +1 @@ +# Python package marker diff --git a/fiveparaminte-main/routers/dashboard.py b/fiveparaminte-main/routers/dashboard.py new file mode 100644 index 0000000..5908179 --- /dev/null +++ b/fiveparaminte-main/routers/dashboard.py @@ -0,0 +1,470 @@ +import asyncio +import json +import logging +import socket +from typing import List + +from pydantic import BaseModel +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect +from fastapi.responses import HTMLResponse +from sqlalchemy.orm import Session +from sqlalchemy import or_ +from database import get_db +from models import Device, Patient, VitalReading, TransmissionLog + +logger = logging.getLogger(__name__) +router = APIRouter() + +class PatientAssignRequest(BaseModel): + device_id: str + mrn: str + name: Optional[str] = None + dob: Optional[str] = None + gender: Optional[str] = None + bed_id: Optional[str] = None + +# === WebSocket Manager for real-time vitals push === +class VitalsWebSocketManager: + """Manages WebSocket connections for live vitals streaming.""" + + def __init__(self): + self.active_connections: List[WebSocket] = [] + + async def connect(self, websocket: WebSocket): + await websocket.accept() + self.active_connections.append(websocket) + logger.info(f"WebSocket client connected ({len(self.active_connections)} active)") + + def disconnect(self, websocket: WebSocket): + if websocket in self.active_connections: + self.active_connections.remove(websocket) + logger.info(f"WebSocket client disconnected ({len(self.active_connections)} active)") + + async def broadcast_vitals(self, vitals_data: dict): + """Send vitals to all connected WebSocket clients.""" + dead = [] + for connection in self.active_connections: + try: + await connection.send_json(vitals_data) + except Exception: + dead.append(connection) + for d in dead: + self.disconnect(d) + + +# Global singleton +ws_manager = VitalsWebSocketManager() + + +def get_ws_manager() -> VitalsWebSocketManager: + return ws_manager + + +# === Dashboard HTML Route === +@router.get("/dashboard", response_class=HTMLResponse) +def serve_dashboard(): + """Serve the vitals monitoring dashboard.""" + import os + dashboard_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static", "dashboard.html") + try: + with open(dashboard_path, "r", encoding="utf-8") as f: + return HTMLResponse(content=f.read()) + except FileNotFoundError: + return HTMLResponse(content="

Dashboard not found

static/dashboard.html is missing

", status_code=404) + + +# === API Endpoints === +@router.get("/health") +def health_check(): + return {"status": "ok", "service": "Patient Monitor Vital Signs Forwarder"} + +@router.get("/status") +def get_system_status(db: Session = Depends(get_db)): + total_devices = db.query(Device).count() + active_devices = db.query(Device).filter(Device.status == "active").count() + total_readings = db.query(VitalReading).count() + pending_transmissions = db.query(VitalReading).filter(VitalReading.transmitted == False).count() + + return { + "total_devices": total_devices, + "active_devices": active_devices, + "total_readings_stored": total_readings, + "pending_transmissions": pending_transmissions + } + +@router.get("/devices") +def list_devices(db: Session = Depends(get_db)): + devices = db.query(Device).all() + return devices + +@router.get("/latest-readings") +def get_latest_readings( + limit: int = 10, + mrn: Optional[str] = None, + name: Optional[str] = None, + dob: Optional[str] = None, + device_id: Optional[str] = None, + db: Session = Depends(get_db) +): + query = db.query(VitalReading) + if device_id: + query = query.filter(VitalReading.device_id == device_id) + + patient_filters = [] + if mrn: + mrn_clean = mrn.strip() + patient_filters.append(Patient.mrn == mrn_clean) + patient_filters.append(Patient.patient_id == mrn_clean) + if name: + name_clean = name.strip() + patient_filters.append(Patient.name.ilike(f"%{name_clean}%")) + if dob: + dob_clean = dob.strip() + patient_filters.append(Patient.dob.ilike(f"%{dob_clean}%")) + + if patient_filters: + matching_patients = db.query(Patient).filter(or_(*patient_filters)).all() + matching_dev_ids = [p.device_id for p in matching_patients if p.device_id] + matching_mrns = [p.mrn for p in matching_patients if p.mrn] + [p.patient_id for p in matching_patients if p.patient_id] + + conds = [] + if mrn: + conds.append(VitalReading.patient_id == mrn.strip()) + if matching_mrns: + conds.append(VitalReading.patient_id.in_(matching_mrns)) + if matching_dev_ids: + conds.append(VitalReading.device_id.in_(matching_dev_ids)) + + if conds: + query = query.filter(or_(*conds)) + + readings = query.order_by(VitalReading.timestamp.desc()).limit(limit).all() + return readings + +@router.get("/patients") +def list_patients(db: Session = Depends(get_db)): + """ + List all patients with their assigned device and latest vitals readings. + """ + patients = db.query(Patient).filter(Patient.patient_id != "UNKNOWN", Patient.mrn != "UNKNOWN").all() + results = [] + for p in patients: + # Find latest vitals for this patient or this patient's assigned device + latest_reading = ( + db.query(VitalReading) + .filter( + or_( + VitalReading.patient_id == p.mrn, + VitalReading.patient_id == p.patient_id, + (VitalReading.device_id == p.device_id) if p.device_id else False + ) + ) + .order_by(VitalReading.timestamp.desc()) + .first() + ) + + device = db.query(Device).filter(Device.device_id == p.device_id).first() if p.device_id else None + + results.append({ + "id": p.id, + "patient_id": p.patient_id, + "mrn": p.mrn or p.patient_id, + "name": p.name or "Unknown", + "gender": p.gender, + "dob": p.dob, + "bed_id": p.bed_id, + "device_id": p.device_id, + "device_status": device.status if device else "unassigned", + "device_ip": device.ip_address if device else None, + "latest_vitals": { + "heart_rate": latest_reading.heart_rate if latest_reading else None, + "spo2": latest_reading.spo2 if latest_reading else None, + "systolic_bp": latest_reading.systolic_bp if latest_reading else None, + "diastolic_bp": latest_reading.diastolic_bp if latest_reading else None, + "map_bp": latest_reading.map_bp if latest_reading else None, + "respiratory_rate": latest_reading.respiratory_rate if latest_reading else None, + "temperature": latest_reading.temperature if latest_reading else None, + "timestamp": latest_reading.timestamp.isoformat() if (latest_reading and latest_reading.timestamp) else None, + } if latest_reading else None + }) + return results + + +@router.post("/patient/assign") +@router.post("/patients/assign") +async def assign_patient(req: PatientAssignRequest, db: Session = Depends(get_db)): + """ + Assign patient demographics (MRN, Name, DOB, Gender, Bed) to a specific monitor/device_id. + """ + from contec_server import set_assigned_patient + + device_id = req.device_id.strip() + mrn = req.mrn.strip() if req.mrn else "" + if not device_id: + raise HTTPException(status_code=400, detail="device_id is required") + + if not mrn or mrn.upper() in ("UNKNOWN", "CLEAR", "NONE", "UNASSIGN"): + # Clear assignment from in-memory cache and DB + from contec_server import device_patient_map, device_cache + if device_id in device_patient_map: + del device_patient_map[device_id] + if device_id in device_cache: + cached = device_cache[device_id] + cached.patient_id = "UNKNOWN" + cached.patient_name = None + cached.patient_gender = None + cached.patient_dob = None + cached.bed_id = None + db.query(Patient).filter(Patient.device_id == device_id).update({"device_id": None}) + db.commit() + try: + await ws_manager.broadcast_vitals({ + "event": "patient_assigned", + "device_id": device_id, + "patient_id": "UNKNOWN", + "mrn": "UNKNOWN", + "patient_name": None, + "name": None, + "patient_gender": None, + "gender": None, + "patient_dob": None, + "dob": None, + "bed_id": None + }) + except Exception: + pass + return { + "status": "success", + "message": f"Device {device_id} unassigned successfully" + } + + # 1. Clear device_id from any other patient previously holding this device + existing_for_device = db.query(Patient).filter(Patient.device_id == device_id).all() + for p in existing_for_device: + if p.mrn != mrn and p.patient_id != mrn: + p.device_id = None + + # 2. Find or create patient by MRN + patient = db.query(Patient).filter(or_(Patient.mrn == mrn, Patient.patient_id == mrn)).first() + if not patient: + patient = Patient( + patient_id=mrn, + mrn=mrn, + name=req.name or "Unknown", + gender=req.gender, + dob=req.dob, + bed_id=req.bed_id, + device_id=device_id + ) + db.add(patient) + else: + patient.patient_id = mrn + patient.mrn = mrn + if req.name is not None: + patient.name = req.name + if req.gender is not None: + patient.gender = req.gender + if req.dob is not None: + patient.dob = req.dob + if req.bed_id is not None: + patient.bed_id = req.bed_id + patient.device_id = device_id + + # 3. Associate recent readings for this device that were marked "UNKNOWN" + db.query(VitalReading).filter( + VitalReading.device_id == device_id, + VitalReading.patient_id == "UNKNOWN" + ).update({"patient_id": mrn}) + + db.commit() + db.refresh(patient) + + # 4. Update in-memory server mapping + p_data = { + "patient_id": patient.mrn, + "mrn": patient.mrn, + "name": patient.name, + "gender": patient.gender, + "dob": patient.dob, + "bed_id": patient.bed_id, + } + set_assigned_patient(device_id, p_data) + + # 5. Broadcast to connected WebSocket clients + try: + await ws_manager.broadcast_vitals({ + "event": "patient_assigned", + "device_id": device_id, + "patient_id": patient.mrn, + "mrn": patient.mrn, + "patient_name": patient.name, + "name": patient.name, + "patient_gender": patient.gender, + "gender": patient.gender, + "patient_dob": patient.dob, + "dob": patient.dob, + "bed_id": patient.bed_id, + }) + except Exception as e: + logger.debug(f"Broadcast of patient_assigned failed: {e}") + + return { + "status": "success", + "message": f"Patient {patient.name} ({mrn}) assigned to device {device_id}", + "patient": p_data + } + + +@router.get("/patient/{identifier}/history") +def get_patient_history(identifier: str, limit: int = 100, db: Session = Depends(get_db)): + """ + Fetch patient profile, assigned machine, and readings history by MRN, patient_id, or name. + """ + patient = db.query(Patient).filter( + or_( + Patient.mrn == identifier, + Patient.patient_id == identifier, + Patient.name.ilike(f"%{identifier}%") + ) + ).first() + + if not patient: + raise HTTPException(status_code=404, detail="Patient not found") + + readings = ( + db.query(VitalReading) + .filter(or_(VitalReading.patient_id == patient.mrn, VitalReading.patient_id == patient.patient_id)) + .order_by(VitalReading.timestamp.desc()) + .limit(limit) + .all() + ) + + device = db.query(Device).filter(Device.device_id == patient.device_id).first() if patient.device_id else None + + return { + "patient": { + "mrn": patient.mrn or patient.patient_id, + "name": patient.name, + "gender": patient.gender, + "dob": patient.dob, + "bed_id": patient.bed_id, + "device_id": patient.device_id, + }, + "device": { + "device_id": device.device_id if device else None, + "ip_address": device.ip_address if device else None, + "status": device.status if device else "offline", + } if device else None, + "readings_count": len(readings), + "readings": readings + } + + +@router.get("/readings") +def get_readings( + device_id: Optional[str] = None, + patient_id: Optional[str] = None, + mrn: Optional[str] = None, + name: Optional[str] = None, + gender: Optional[str] = None, + dob: Optional[str] = None, + limit: int = 50, + offset: int = 0, + db: Session = Depends(get_db) +): + """ + Fetch vital readings filtered by patient MRN, Name, DOB, Gender, or Device ID. + """ + query = db.query(VitalReading) + + target_mrn = mrn or patient_id + if target_mrn: + query = query.filter(VitalReading.patient_id == target_mrn) + + # If filtered by patient demographics (name, gender, dob) + if name or gender or dob: + p_query = db.query(Patient) + if name: + p_query = p_query.filter(Patient.name.ilike(f"%{name}%")) + if gender: + p_query = p_query.filter(Patient.gender.ilike(f"%{gender}%")) + if dob: + p_query = p_query.filter(Patient.dob.ilike(f"%{dob}%")) + + matching_patients = p_query.all() + matching_mrns = [p.mrn for p in matching_patients if p.mrn] + [p.patient_id for p in matching_patients if p.patient_id] + matching_devices = [p.device_id for p in matching_patients if p.device_id] + + query = query.filter( + or_( + VitalReading.patient_id.in_(matching_mrns), + VitalReading.device_id.in_(matching_devices) + ) + ) + + if device_id: + query = query.filter(VitalReading.device_id == device_id) + + return query.order_by(VitalReading.timestamp.desc()).offset(offset).limit(limit).all() + +@router.get("/transmission-logs") +def get_transmission_logs(limit: int = 50, db: Session = Depends(get_db)): + logs = db.query(TransmissionLog).order_by(TransmissionLog.timestamp.desc()).limit(limit).all() + return logs + +@router.post("/test-api") +def test_target_api_connection(): + """ + Endpoint to trigger a test POST to the target API. + """ + from config import settings + return { + "message": "Test triggered", + "target_url": settings.target_api_url + } + + +@router.get("/network-info") +def get_network_info(): + """ + Returns this PC's local IP addresses. + Useful for configuring the CMS7000PLUS monitor's CMS Server IP. + """ + from config import settings + ips = [] + try: + for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): + ip = info[4][0] + if ip not in ips and not ip.startswith('127.'): + ips.append(ip) + except Exception: + pass + # Find primary IP + primary_ip = None + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(('8.8.8.8', 80)) + primary_ip = s.getsockname()[0] + s.close() + if primary_ip not in ips: + ips.insert(0, primary_ip) + except Exception: + pass + + return { + "hostname": socket.gethostname(), + "ip_addresses": ips, + "primary_ip": primary_ip or (ips[0] if ips else None), + "monitor_model": settings.monitor_model, + "contec_ports": settings.contec_ports, + "instructions": [ + f"1. Connect your {settings.monitor_model} to this PC via Ethernet", + f"2. On the monitor: System Setup → Network → CMS Settings", + f"3. Set Server IP to: {primary_ip or 'your PC IP'}", + f"4. Set Server Port to: {settings.contec_ports[0] if settings.contec_ports else 511}", + f"5. Enable the CMS connection", + f"6. The dashboard will show data automatically" + ] + } diff --git a/fiveparaminte-main/static/dashboard.html b/fiveparaminte-main/static/dashboard.html new file mode 100644 index 0000000..18a6f21 --- /dev/null +++ b/fiveparaminte-main/static/dashboard.html @@ -0,0 +1,3120 @@ + + + + + + Patient Vital Signs Monitor — Contec CMS7000PLUS + + + + + + + +
+
+ +
+ + + + +
+
+
+ 📡 + CMS7000PLUS Setup Guide — Configure Your Monitor +
+
▾ Collapse
+
+
+
+
1
+
+

Connect via Ethernet

+

Connect your CMS7000PLUS to this PC using an Ethernet cable (direct or through a network switch).

+
+
+
+
2
+
+

Set PC IP Address

+

Set your PC's Ethernet adapter to a static IP, e.g. 192.168.1.50, subnet 255.255.255.0.

+
+
+
+
3
+
+

Configure Monitor CMS Settings

+

On your CMS7000PLUS: System Setup → Network → CMS Settings. Set the Server IP to your PC's IP.

+
+
+
+
4
+
+

Set Server Port

+

Set Server Port to 511 (default). If running without admin, use a port >1024 like 6060.

+
+
+
+
5
+
+

Enable CMS Connection

+

Enable the CMS/Central Monitor connection on your CMS7000PLUS. Set the sending interval (e.g., 5 seconds).

+
+
+
+
6
+
+

Data Will Appear Automatically

+

Once connected, vital signs will appear on this dashboard in real-time via WebSocket or polling.

+
+
+
+
+
🖥️ Your PC's Detected IP Address (set this on the monitor)
+
+
Detecting...
+
+
+
+ + +
+
+ 🔍 + +
+
+ + + +
+
+ + +
+
+ 🖥️ Monitors / Beds: +
+ +
+
+
+ + +
+
+ + +
+
+
⏳
+
Waiting for patient monitors to connect...
+
+ Local Wi-Fi monitors connect to 192.168.100.112:511 | Remote hotspot monitors connect to 159.223.110.159:61501 +
+
+
+ + + + + +
+
+ Model: + CMS7000PLUS +
+
+
+ Device: + Waiting... +
+
+
+ Patient: + Waiting... +
+
+
+ Readings: + 0 +
+
+
+ Last Update: + Never +
+
+
+ API: + Checking... +
+
+
+ + + + + + + + + +