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

Compare commits

base: aruncsh/ContecMonitor:0ecace84d29e150b5275303d91e4858ada2d177c
Branches Tags
aruncsh/ContecMonitor:main
..
compare: aruncsh/ContecMonitor:2c86c15f1d8b6763b15a37fb4ca240907e5076a5
Branches Tags
aruncsh/ContecMonitor:main
1 Commits
0ecace84d2 .. 2c86c15f1d
Author SHA1 Message Date
prathiyuman 2c86c15f1d Convert temperature display from Celsius to Fahrenheit 2026-07-13 16:05:17 +05:30
27 changed files with 0 additions and 4434 deletions
Expand all files Collapse all files
.dockerignore
-17
View File
@@ -1,17 +0,0 @@
__pycache__/
*.py[cod]
*.db
*.sqlite3
*.log
*.bin
.env
venv/
.venv/
build/
dist/
*.spec
scratch/
captures/
.git/
.idea/
.vscode/
.gitignore
-37
View File
@@ -1,37 +0,0 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# PyInstaller
build/
dist/
*.spec
# Databases
*.db
*.sqlite3
# Logs
*.log
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Scratch / debug
scratch/
captures/
*.bin
# IDE / OS
.idea/
.vscode/
*.swp
*.swo
.DS_Store
Dockerfile
-15
View File
@@ -1,15 +0,0 @@
FROM python:3.12-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Render sets PORT env var (default 10000)
EXPOSE 10000
CMD uvicorn main:app --host 0.0.0.0 --port ${PORT:-10000}
Procfile
-1
View File
@@ -1 +0,0 @@
web: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT
api_client.py
-63
View File
@@ -1,63 +0,0 @@
import httpx
import logging
from typing import Optional
from schemas import NormalizedVitals
from config import settings
logger = logging.getLogger(__name__)
async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
"""
Attempts to POST the normalized vitals to the configured target API
and to the Render cloud database.
Returns True if at least one forward succeeds.
"""
import os
headers = {"Content-Type": "application/json"}
# 1. Forward to primary target API (e.g. Injomo Dev Workflow)
primary_url = settings.target_api_url
primary_success = False
if primary_url:
try:
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
except Exception as e:
logger.error(f"Error forwarding to primary API: {e}")
# 2. Forward to Render Cloud Server
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
render_success = False
# Detect if we are already running on Render to prevent an infinite loop
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not is_on_render:
try:
url_with_key = render_url
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
except Exception as e:
logger.error(f"Error forwarding to Render Cloud: {e}")
return primary_success or render_success
background_tasks.py
-67
View File
@@ -1,67 +0,0 @@
import asyncio
import logging
from database import SessionLocal
from models import VitalReading, TransmissionLog
from schemas import NormalizedVitals
from api_client import forward_vitals_to_api
from config import settings
logger = logging.getLogger(__name__)
async def retry_failed_transmissions():
"""
Background task that periodically checks for untransmitted vital readings
and attempts to forward them to the API.
"""
logger.info("Starting background task for retrying failed transmissions...")
while True:
try:
db = SessionLocal()
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
)
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()
except Exception as e:
logger.error(f"Error in retry background task: {e}")
await asyncio.sleep(settings.retry_interval_seconds)
build.bat
-29
View File
@@ -1,29 +0,0 @@
@echo off
echo Building Mindray HL7 Forwarder Executable...
REM Clean previous builds
rmdir /s /q build
rmdir /s /q dist
REM Run PyInstaller with necessary hidden imports for FastAPI, Uvicorn, and SQLAlchemy
pyinstaller -F --clean ^
--name "MindrayHL7Forwarder" ^
--hidden-import uvicorn.logging ^
--hidden-import uvicorn.loops ^
--hidden-import uvicorn.loops.auto ^
--hidden-import uvicorn.protocols ^
--hidden-import uvicorn.protocols.http ^
--hidden-import uvicorn.protocols.http.auto ^
--hidden-import uvicorn.protocols.websockets ^
--hidden-import uvicorn.protocols.websockets.auto ^
--hidden-import uvicorn.lifespan ^
--hidden-import uvicorn.lifespan.on ^
--hidden-import uvicorn.lifespan.off ^
--hidden-import fastapi ^
--hidden-import pydantic ^
--hidden-import sqlalchemy ^
--hidden-import hl7 ^
main.py
echo Build complete! Executable is located in the dist/ folder.
pause
build.sh
-28
View File
@@ -1,28 +0,0 @@
#!/bin/bash
echo "Building Mindray HL7 Forwarder Executable for Linux..."
# Clean previous builds
rm -rf build
rm -rf dist
# Run PyInstaller with necessary hidden imports for FastAPI, Uvicorn, and SQLAlchemy
pyinstaller -F --clean \
--name "MindrayHL7Forwarder" \
--hidden-import uvicorn.logging \
--hidden-import uvicorn.loops \
--hidden-import uvicorn.loops.auto \
--hidden-import uvicorn.protocols \
--hidden-import uvicorn.protocols.http \
--hidden-import uvicorn.protocols.http.auto \
--hidden-import uvicorn.protocols.websockets \
--hidden-import uvicorn.protocols.websockets.auto \
--hidden-import uvicorn.lifespan \
--hidden-import uvicorn.lifespan.on \
--hidden-import uvicorn.lifespan.off \
--hidden-import fastapi \
--hidden-import pydantic \
--hidden-import sqlalchemy \
--hidden-import hl7 \
main.py
echo "Build complete! Executable is located in the dist/ folder."
config.json
-15
View File
@@ -1,15 +0,0 @@
{
"hl7_host": "0.0.0.0",
"hl7_port": 6060,
"contec_ports": [511, 512, 513, 514, 515, 516, 517, 518, 519, 520, 8001, 8002, 8300, 9000, 9100, 9200, 10008, 12345],
"monitor_ip": "",
"monitor_model": "CMS8500",
"device_models": {
"192.168.100.120": "CMS7000PLUS"
},
"target_api_url": "https://innov-dev.beta.injomo.com/workflow.trigger/69b2712a7866bf86ca0060c3",
"api_token": "",
"retry_interval_seconds": 60,
"log_file": "hl7_forwarder.log",
"database_url": "sqlite:///./hl7_forwarder.db"
}
config.py
-45
View File
@@ -1,45 +0,0 @@
import json
import os
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
hl7_host: str = "0.0.0.0"
hl7_port: int = 6060
contec_ports: list = [511, 515, 516, 517, 518, 519, 520]
monitor_ip: str = ""
monitor_model: str = "CMS7000PLUS"
device_models: dict = {}
target_api_url: str = "https://api.example.com/vitals"
api_token: str = ""
retry_interval_seconds: int = 60
log_file: str = "hl7_forwarder.log"
database_url: str = "sqlite:///./hl7_forwarder.db"
class Config:
env_file = ".env"
def get_settings() -> Settings:
base_dir = os.path.dirname(os.path.abspath(__file__))
config_file = os.path.join(base_dir, "config.json")
settings_obj = Settings()
if os.path.exists(config_file):
with open(config_file, "r") as f:
try:
data = json.load(f)
for k, v in data.items():
setattr(settings_obj, k, v)
except json.JSONDecodeError:
pass
if settings_obj.database_url.startswith("sqlite:///./"):
db_name = settings_obj.database_url.split("sqlite:///./")[1]
settings_obj.database_url = f"sqlite:///{os.path.join(base_dir, db_name)}"
if not os.path.isabs(settings_obj.log_file):
settings_obj.log_file = os.path.join(base_dir, settings_obj.log_file)
return settings_obj
settings = get_settings()
contec_discovery.py
-246
View File
@@ -1,246 +0,0 @@
"""
Contec CMS7000PLUS / CMS9200PLUS Discovery Tool
==================================
This script listens on multiple TCP ports simultaneously to discover
what data the CMS7000PLUS / CMS9200PLUS patient monitor sends over the network.
Usage:
python contec_discovery.py
It will listen on ports: 511, 515, 516, 517, 518, 519, 520, 6060
and display all received data in hex + ASCII format for protocol analysis.
IMPORTANT: Run as Administrator (required for port 511 on Windows)
"""
import asyncio
import logging
import sys
import os
from datetime import datetime
# --- Configuration ---
LISTEN_HOST = "0.0.0.0"
DISCOVERY_PORTS = [511, 515, 516, 517, 518, 519, 520, 6060]
LOG_DIR = "captures"
# --- Setup Logging ---
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("contec_discovery")
def hex_dump(data: bytes, width: int = 16) -> str:
"""Format bytes as a hex dump with ASCII representation."""
lines = []
for i in range(0, len(data), width):
chunk = data[i:i + width]
hex_part = ' '.join(f'{b:02x}' for b in chunk)
ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
lines.append(f" {i:04x} {hex_part:<{width * 3}} |{ascii_part}|")
return '\n'.join(lines)
def try_decode_hl7(data: bytes) -> str | None:
"""Attempt to decode the data as an HL7 message."""
try:
# Strip MLLP framing if present
cleaned = data
if cleaned.startswith(b'\x0b'):
cleaned = cleaned[1:]
if b'\x1c\x0d' in cleaned:
cleaned = cleaned[:cleaned.find(b'\x1c\x0d')]
text = cleaned.decode('utf-8', errors='ignore').strip()
if 'MSH|' in text or 'MSH' in text[:20]:
return text
except Exception:
pass
return None
async def handle_discovery_client(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
port: int,
capture_file
):
"""Handle a single incoming connection and dump all data."""
client_ip, client_port = writer.get_extra_info('peername')
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
print("\n" + "=" * 80)
print(f" NEW CONNECTION on port {port}")
print(f" From: {client_ip}:{client_port}")
print(f" Time: {timestamp}")
print("=" * 80)
capture_file.write(f"\n{'='*80}\n")
capture_file.write(f"Connection on port {port} from {client_ip}:{client_port} at {timestamp}\n")
capture_file.write(f"{'='*80}\n")
capture_file.flush()
packet_count = 0
try:
while True:
data = await asyncio.wait_for(reader.read(8192), timeout=30.0)
if not data:
break
packet_count += 1
recv_time = datetime.now().strftime("%H:%M:%S.%f")
print(f"\n--- Packet #{packet_count} on port {port} [{recv_time}] ({len(data)} bytes) ---")
print(hex_dump(data))
# Try to decode as HL7
hl7_text = try_decode_hl7(data)
if hl7_text:
print(f"\n >>> HL7 MESSAGE DETECTED <<<")
segments = hl7_text.replace('\n', '\r').split('\r')
for seg in segments:
seg = seg.strip()
if seg:
print(f" {seg}")
# Extract vital signs from OBX segments
if seg.startswith("OBX|"):
fields = seg.split('|')
if len(fields) > 5:
obs_id = fields[3] if len(fields) > 3 else "?"
obs_value = fields[5] if len(fields) > 5 else "?"
obs_unit = fields[6] if len(fields) > 6 else ""
print(f" -> Vital: {obs_id} = {obs_value} {obs_unit}")
# Also try plain text decode
try:
text = data.decode('utf-8', errors='ignore')
printable_ratio = sum(1 for c in text if c.isprintable() or c in '\r\n\t') / max(len(text), 1)
if printable_ratio > 0.7 and not hl7_text:
print(f"\n [Text representation]:")
for line in text.split('\r'):
line = line.strip()
if line:
print(f" {line}")
except Exception:
pass
# Write to capture file
capture_file.write(f"\nPacket #{packet_count} [{recv_time}] ({len(data)} bytes)\n")
capture_file.write(hex_dump(data) + "\n")
if hl7_text:
capture_file.write(f"HL7: {hl7_text}\n")
capture_file.write(f"Raw bytes: {data.hex()}\n")
capture_file.flush()
except asyncio.TimeoutError:
print(f" [Timeout - no data for 30s on port {port}]")
except asyncio.CancelledError:
pass
except ConnectionResetError:
print(f" [Connection reset by {client_ip} on port {port}]")
except Exception as e:
print(f" [Error on port {port}: {e}]")
finally:
print(f"\n Connection closed on port {port} from {client_ip} ({packet_count} packets received)")
writer.close()
try:
await writer.wait_closed()
except Exception:
pass
async def start_port_listener(port: int, capture_file):
"""Start a TCP server on a specific port."""
async def client_handler(reader, writer):
await handle_discovery_client(reader, writer, port, capture_file)
try:
server = await asyncio.start_server(client_handler, LISTEN_HOST, port)
logger.info(f" ✓ Listening on port {port}")
return server
except PermissionError:
logger.error(f" ✗ Port {port} - Permission denied (try running as Administrator)")
return None
except OSError as e:
if "already in use" in str(e).lower() or e.errno == 10048:
logger.error(f" ✗ Port {port} - Already in use by another application")
else:
logger.error(f" ✗ Port {port} - {e}")
return None
async def main():
print(r"""
╔══════════════════════════════════════════════════════════════╗
║ Contec CMS7000PLUS / CMS9200PLUS Discovery Tool ║
║ ║
║ This tool listens for incoming data from your patient ║
║ monitor to discover its communication protocol. ║
║ ║
║ SETUP REQUIRED: ║
║ 1. Connect monitor to your PC via Ethernet ║
║ 2. Set your PC's Ethernet IP to: 192.168.1.50 ║
║ (or whatever subnet your monitor uses) ║
║ 3. On the monitor, set CMS/Server IP to YOUR PC's IP ║
║ 4. Run this script as Administrator ║
║ ║
║ Press Ctrl+C to stop ║
╚══════════════════════════════════════════════════════════════╝
""")
# Create captures directory
os.makedirs(LOG_DIR, exist_ok=True)
capture_filename = os.path.join(
LOG_DIR,
f"capture_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
)
capture_file = open(capture_filename, 'w', encoding='utf-8')
logger.info(f"Capture log: {capture_filename}")
# Show current network config hint
print("\nStarting listeners on all discovery ports...")
print("(If port 511 fails, run PowerShell as Administrator)\n")
servers = []
for port in DISCOVERY_PORTS:
server = await start_port_listener(port, capture_file)
if server:
servers.append(server)
if not servers:
logger.error("No ports could be opened! Check permissions and existing services.")
capture_file.close()
return
active_ports = [s.sockets[0].getsockname()[1] for s in servers if s.sockets]
print(f"\n{'─' * 60}")
print(f" Active on ports: {active_ports}")
print(f" Waiting for data from CMS7000PLUS / CMS9200PLUS...")
print(f" Make sure the monitor's CMS/Server IP points to this PC")
print(f"{'─' * 60}\n")
try:
# Keep running until Ctrl+C
await asyncio.Event().wait()
except asyncio.CancelledError:
pass
finally:
print("\nShutting down...")
for server in servers:
server.close()
for server in servers:
await server.wait_closed()
capture_file.close()
print(f"Capture saved to: {capture_filename}")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n\nDiscovery stopped by user.")
contec_parser.py
-553
View File
@@ -1,553 +0,0 @@
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
contec_server.py
-312
View File
@@ -1,312 +0,0 @@
import asyncio
import logging
from datetime import datetime, timezone
from contec_parser import parse_contec_data
from database import SessionLocal
from models import Device, Patient, VitalReading, TransmissionLog
from api_client import forward_vitals_to_api
from terminal_display import get_terminal_display
logger = logging.getLogger(__name__)
# Constants for MLLP framing (used on port 511 for Contec HL7 text)
VT = b'\x0b'
FS_CR = b'\x1c\x0d'
def generate_ack(received_msg: str) -> str:
"""Generates a simple HL7 ACK message to satisfy the monitor's TCP client."""
try:
segments = received_msg.split('\r')
msh = segments[0] if segments else ""
if not msh.startswith("MSH|"):
return ""
fields = msh.split('|')
if len(fields) < 10:
return ""
sending_app = fields[2]
sending_fac = fields[3]
rec_app = fields[4]
rec_fac = fields[5]
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_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 ""
async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int):
"""
TCP server handler for Contec CMS7000PLUS client connections.
Directs incoming traffic to the unified parse_contec_data handler.
"""
client_ip, client_port = writer.get_extra_info('peername')
logger.info(f"Accepted Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
display = get_terminal_display()
display.connection_opened(client_ip, port)
buffer = b""
try:
while True:
data = await reader.read(8192)
if not data:
logger.info(f"Contec client {client_ip}:{client_port} on port {port} sent EOF (0 bytes)")
break
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)
# ---------------------------------------------------------------
if VT in buffer and FS_CR in buffer and b"MSH|" in buffer:
while True:
start_idx = buffer.find(VT)
end_idx = buffer.find(FS_CR)
if start_idx != -1 and end_idx != -1 and start_idx < end_idx:
raw_hl7 = buffer[start_idx + 1:end_idx].decode('utf-8', errors='ignore')
buffer = buffer[end_idx + 2:]
logger.info(f"Contec MLLP message detected on port {port}")
vitals = parse_contec_data(raw_hl7.encode('utf-8'), client_ip)
if vitals:
await process_vitals(vitals)
ack_msg = generate_ack(raw_hl7)
if ack_msg:
writer.write(VT + ack_msg.encode('utf-8') + FS_CR)
await writer.drain()
else:
break
continue # Go back to reading more data
# ---------------------------------------------------------------
# Binary Contec proprietary protocol (port 514)
# Packets are framed as: [len_lo][len_hi][04][46] ... payload ...
# The monitor sends each packet as a complete TCP write,
# but we buffer anyway in case of TCP segmentation.
# ---------------------------------------------------------------
consumed = 0
while consumed < len(buffer) - 4:
b0 = buffer[consumed]
b1 = buffer[consumed + 1]
b2 = buffer[consumed + 2]
b3 = buffer[consumed + 3]
if (b2 == 0x04 or b2 == 0x01 or b2 == 0x00) and b3 == 0x46:
pkt_len = b0 | (b1 << 8)
pkt_end = consumed + pkt_len
if pkt_len < 5 or pkt_len > 8192:
# Bad length — skip one byte and keep scanning
consumed += 1
continue
if pkt_end > len(buffer):
# Don't have the full packet yet — wait for more data
break
# Parse the complete packet
pkt = buffer[consumed:pkt_end]
vitals = parse_contec_data(pkt, client_ip)
if vitals:
logger.info(
f"[PORT {port}] Vitals from {client_ip}: "
f"SpO2={vitals.spo2}%, HR={vitals.heart_rate}bpm "
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp} "
f"Temp={vitals.temperature}C"
)
await process_vitals(vitals)
consumed = pkt_end
elif (b2 == 0x04 or b2 == 0x01) and b3 == 0x47:
# End-of-frame marker — skip 4 bytes
consumed += 4
else:
# Not a recognised frame start — advance one byte
consumed += 1
# Discard consumed bytes from front of buffer
buffer = buffer[consumed:]
# Safety valve
if len(buffer) > 32768:
logger.warning(f"Buffer too large ({len(buffer)} bytes) on port {port}, resetting")
buffer = b""
except asyncio.CancelledError:
pass
except ConnectionResetError:
logger.info(f"Contec CMS7000PLUS connection reset from {client_ip} on port {port}")
except Exception as e:
logger.error(f"Error in Contec client handler (port {port}): {e}")
finally:
logger.info(f"Closing Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
display.connection_closed(client_ip, port)
writer.close()
try:
await writer.wait_closed()
except Exception:
pass
# Cache to hold merged vitals per device to prevent fragmentation
device_cache = {}
device_field_timestamps = {} # {device_id: {field_name: datetime}}
last_db_write = {}
async def process_vitals(vitals):
"""
Log, database, WebSocket broadcast, and REST forward parsed vitals.
"""
device_id = vitals.device_id
now = datetime.now(timezone.utc)
if device_id not in device_field_timestamps:
device_field_timestamps[device_id] = {}
# 1. Merge new vitals into cached vitals to prevent fragmented entries
if device_id not in device_cache:
device_cache[device_id] = vitals
# Track initial timestamps for present values
for field, value in vitals.model_dump().items():
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
continue
if value is not None:
device_field_timestamps[device_id][field] = now
else:
cached = device_cache[device_id]
# Determine active fields in incoming packet
active_fields = []
if vitals.present_fields is not None:
active_fields = vitals.present_fields
else:
for field, value in vitals.model_dump().items():
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
continue
if value is not None:
active_fields.append(field)
# Update cache & record timestamps
for field in active_fields:
val = getattr(vitals, field)
setattr(cached, field, val)
device_field_timestamps[device_id][field] = now
# Clean up stale fields (older than 15 seconds)
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:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
cached.timestamp = vitals.timestamp
vitals = cached
# 2. Update console display banner
display = get_terminal_display()
display.update_vitals(vitals)
logger.info(
f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id}: "
f"HR={vitals.heart_rate}, SpO2={vitals.spo2}, "
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp}, "
f"RR={vitals.respiratory_rate}, Temp={vitals.temperature}"
)
# 3. Broadcast to live WebSockets dashboard in real-time
try:
from routers.dashboard import get_ws_manager
ws_manager = get_ws_manager()
await ws_manager.broadcast_vitals(vitals.model_dump(mode="json"))
except Exception as e:
logger.debug(f"WebSocket broadcast failed: {e}")
# 4. Throttled Database & REST forwarding (at most once every 5 seconds)
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
last_db_write[device_id] = now
if should_write_db:
# Save to SQLite database and REST forward
db = SessionLocal()
try:
# 1. Device tracking
device = db.query(Device).filter(Device.device_id == vitals.device_id).first()
if not device:
device = Device(
device_id=vitals.device_id,
ip_address=vitals.ip_address or "unknown",
status="active"
)
db.add(device)
else:
if vitals.ip_address:
device.ip_address = vitals.ip_address
device.status = "active"
device.last_seen = datetime.now(timezone.utc)
# 2. Patient tracking
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
if not patient:
patient = Patient(patient_id=vitals.patient_id, name="Unknown")
db.add(patient)
# 3. Create Reading Entry
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
systolic_bp=vitals.systolic_bp,
diastolic_bp=vitals.diastolic_bp,
map_bp=vitals.map_bp,
respiratory_rate=vitals.respiratory_rate,
temperature=vitals.temperature,
transmitted=False
)
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()
except Exception as e:
logger.error(f"Database operation failed: {e}")
db.rollback()
finally:
db.close()
database.py
-17
View File
@@ -1,17 +0,0 @@
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from config import settings
engine = create_engine(
settings.database_url, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
main.py
-189
View File
@@ -1,189 +0,0 @@
import asyncio
import logging
import multiprocessing
import socket
from contextlib import asynccontextmanager
from functools import partial
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import uvicorn
from config import settings
from database import engine, Base
from contec_server import handle_contec_client
from background_tasks import retry_failed_transmissions
from routers import dashboard
from routers.dashboard import get_ws_manager
# Setup Logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(settings.log_file),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Create database tables if they don't exist
Base.metadata.create_all(bind=engine)
def _make_contec_handler(port: int):
"""Create a connection handler closure for a specific Contec port."""
async def handler(reader, writer):
await handle_contec_client(reader, writer, port)
return handler
def _get_local_ips():
"""Get all local IP addresses for this machine."""
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
# Also try connecting to an external address to find the primary IP
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 ips if ips else ['Could not detect — check ipconfig']
@asynccontextmanager
async def lifespan(app: FastAPI):
servers = []
# === Start Contec CMS7000PLUS TCP Servers ===
logger.info("Starting Contec CMS7000PLUS listeners...")
for port in settings.contec_ports:
try:
contec_server = await asyncio.start_server(
_make_contec_handler(port),
settings.hl7_host,
port
)
servers.append(contec_server)
logger.info(f" [OK] Contec listener on port {port}")
except PermissionError:
logger.warning(
f" [FAIL] Port {port} - Permission denied. "
f"Run as Administrator for ports below 1024."
)
except OSError as e:
if "already in use" in str(e).lower() or getattr(e, 'errno', 0) == 10048:
logger.warning(f" [FAIL] Port {port} - Already in use")
else:
logger.warning(f" [FAIL] Port {port} - {e}")
if servers:
active_ports = []
for s in servers:
for sock in s.sockets:
active_ports.append(sock.getsockname()[1])
app.state.active_ports = active_ports
logger.info(f" Active listener ports: {active_ports}")
else:
app.state.active_ports = []
logger.error("No servers could be started! Check port availability and permissions.")
# === Print startup banner ===
import os
port = int(os.environ.get("PORT", 8000))
local_ips = _get_local_ips()
model = settings.monitor_model
print("\n" + "=" * 65)
print(f" Contec {model} Patient Monitor - Vital Signs Forwarder")
print("=" * 65)
print(f" DASHBOARD: 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}")
print(f" API Target: {settings.target_api_url}")
print(f" Database: {settings.database_url}")
print(f" ---------------------------------------------------------")
print(f" YOUR PC's IP ADDRESSES (configure on your monitor):")
for ip in local_ips:
print(f" -> {ip}")
print("=" * 65)
print(f" Waiting for {model} connections...")
print(f" On your monitor: System Setup > Network > CMS Settings")
print(f" Set Server IP = one of the IPs above, Port = {settings.contec_ports[0] if settings.contec_ports else 511}")
print("=" * 65 + "\n")
# Startup: Start the background retry task
retry_task = asyncio.create_task(retry_failed_transmissions())
yield
# Shutdown
logger.info("Shutting down servers...")
retry_task.cancel()
for server in servers:
server.close()
for server in servers:
await server.wait_closed()
# Initialize FastAPI app
app = FastAPI(
title="Contec CMS7000PLUS Vital Signs Forwarder",
description="Service to receive vital signs from Contec CMS7000PLUS patient monitors and forward them to a REST API",
version="2.0.0",
lifespan=lifespan
)
# Enable CORS for cross-origin app integrations (e.g. Electron apps, local dev servers)
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include dashboard routes
app.include_router(dashboard.router, prefix="/api")
# Serve static folder
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")
# === WebSocket endpoint for live vitals ===
@app.websocket("/ws/vitals")
async def websocket_vitals(websocket: WebSocket):
manager = get_ws_manager()
await manager.connect(websocket)
try:
while True:
# Keep connection alive, listen for any client messages
await websocket.receive_text()
except WebSocketDisconnect:
manager.disconnect(websocket)
except Exception:
manager.disconnect(websocket)
if __name__ == "__main__":
# Required for Windows when building with PyInstaller using multiprocessing (uvicorn workers)
multiprocessing.freeze_support()
logger.info("Initializing Application...")
# Run the FastAPI server.
# Notice we run on port 8000 (or PORT env var) for the REST Dashboard, while HL7 is on settings.hl7_port (e.g. 6060)
import os
port = int(os.environ.get("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port)
models.py
-41
View File
@@ -1,41 +0,0 @@
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from datetime import datetime, timezone
from database import Base
class Device(Base):
__tablename__ = "devices"
id = Column(Integer, primary_key=True, index=True)
device_id = Column(String, unique=True, index=True)
ip_address = Column(String)
status = Column(String, default="active")
last_seen = Column(DateTime, default=lambda: datetime.now(timezone.utc))
class Patient(Base):
__tablename__ = "patients"
id = Column(Integer, primary_key=True, index=True)
patient_id = Column(String, unique=True, index=True)
name = Column(String)
class VitalReading(Base):
__tablename__ = "vital_readings"
id = Column(Integer, primary_key=True, index=True)
device_id = Column(String, index=True)
patient_id = Column(String, index=True)
timestamp = Column(DateTime, index=True)
heart_rate = Column(Float, nullable=True)
spo2 = Column(Float, nullable=True)
systolic_bp = Column(Float, nullable=True)
diastolic_bp = Column(Float, nullable=True)
map_bp = Column(Float, nullable=True)
respiratory_rate = Column(Float, nullable=True)
temperature = Column(Float, nullable=True)
transmitted = Column(Boolean, default=False, index=True)
class TransmissionLog(Base):
__tablename__ = "transmission_logs"
id = Column(Integer, primary_key=True, index=True)
reading_id = Column(Integer, index=True)
status = Column(String)
response_code = Column(Integer, nullable=True)
error_message = Column(String, nullable=True)
timestamp = Column(DateTime, default=lambda: datetime.now(timezone.utc))
render.yaml
-15
View File
@@ -1,15 +0,0 @@
services:
- type: web
name: contec-monitor-api
runtime: python
buildCommand: pip install -r requirements.txt
startCommand: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT
envVars:
- key: PORT
value: 10000
- key: PYTHON_VERSION
value: 3.12.0
- key: TARGET_API_URL
sync: false
- key: API_TOKEN
sync: false
requirements.txt
-9
View File
@@ -1,9 +0,0 @@
fastapi
uvicorn
sqlalchemy
pydantic
pydantic-settings
httpx
gunicorn
rich
websockets
routers/__init__.py
-1
View File
@@ -1 +0,0 @@
# Python package marker
routers/dashboard.py
-271
View File
@@ -1,271 +0,0 @@
import asyncio
import json
import logging
import socket
from typing import List
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from database import get_db
from models import Device, VitalReading, TransmissionLog
logger = logging.getLogger(__name__)
router = APIRouter()
# === 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="<h1>Dashboard not found</h1><p>static/dashboard.html is missing</p>", 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(request: Request, 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()
active_ports = getattr(request.app.state, "active_ports", [])
return {
"total_devices": total_devices,
"active_devices": active_devices,
"total_readings_stored": total_readings,
"pending_transmissions": pending_transmissions,
"active_ports": active_ports
}
@router.get("/devices")
def list_devices(db: Session = Depends(get_db)):
devices = db.query(Device).all()
now = datetime.now(timezone.utc)
updated = False
for device in devices:
last_seen = device.last_seen
if last_seen.tzinfo is None:
last_seen = last_seen.replace(tzinfo=timezone.utc)
# If not seen for more than 15 seconds, mark offline
if (now - last_seen).total_seconds() > 15.0:
if device.status != "offline":
device.status = "offline"
updated = True
else:
if device.status != "active":
device.status = "active"
updated = True
if updated:
db.commit()
return devices
@router.get("/latest-readings")
def get_latest_readings(device_id: str = None, limit: int = 10, db: Session = Depends(get_db)):
query = db.query(VitalReading)
if device_id:
query = query.filter(VitalReading.device_id == device_id)
readings = query.order_by(VitalReading.timestamp.desc()).limit(limit).all()
return readings
@router.get("/readings")
def get_readings(device_id: str = None, patient_id: str = None, limit: int = 50, db: Session = Depends(get_db)):
query = db.query(VitalReading)
if device_id:
query = query.filter(VitalReading.device_id == device_id)
if patient_id:
query = query.filter(VitalReading.patient_id == patient_id)
return query.order_by(VitalReading.timestamp.desc()).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"
]
}
# === Push Vitals Endpoint (for local-to-cloud sync) ===
from schemas import NormalizedVitals
from models import Patient
@router.post("/push-vitals")
async def push_vitals(vitals: NormalizedVitals, api_key: str = None, db: Session = Depends(get_db)):
"""
Receive vitals pushed from a local instance and store + broadcast them.
This enables the 'split architecture' where:
- Local instance receives data from physical monitors via TCP
- Local instance POSTs vitals here for cloud storage & dashboard access
Optional: pass ?api_key=YOUR_KEY for basic authentication.
"""
from config import settings
# Optional API key check
if settings.api_token and api_key != settings.api_token:
if settings.api_token: # Only enforce if a token is configured
raise HTTPException(status_code=401, detail="Invalid or missing api_key")
try:
# 1. Device tracking
device = db.query(Device).filter(Device.device_id == vitals.device_id).first()
if not device:
device = Device(
device_id=vitals.device_id,
ip_address=vitals.ip_address or "remote-push",
status="active"
)
db.add(device)
else:
if vitals.ip_address:
device.ip_address = vitals.ip_address
device.status = "active"
device.last_seen = datetime.now(timezone.utc)
# 2. Patient tracking
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
if not patient:
patient = Patient(patient_id=vitals.patient_id, name="Unknown")
db.add(patient)
# 3. Create vital reading
from models import VitalReading as VR
reading = VR(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
systolic_bp=vitals.systolic_bp,
diastolic_bp=vitals.diastolic_bp,
map_bp=vitals.map_bp,
respiratory_rate=vitals.respiratory_rate,
temperature=vitals.temperature,
transmitted=True # Already received = transmitted
)
db.add(reading)
db.commit()
db.refresh(reading)
# 4. Broadcast to WebSocket clients
await ws_manager.broadcast_vitals(vitals.model_dump(mode="json"))
logger.info(f"Push-vitals received for device {vitals.device_id}")
return {
"status": "ok",
"reading_id": reading.id,
"device_id": vitals.device_id,
"patient_id": vitals.patient_id
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Push-vitals error: {e}")
db.rollback()
raise HTTPException(status_code=500, detail=f"Failed to store vitals: {str(e)}")
run.sh
-21
View File
@@ -1,21 +0,0 @@
#!/bin/bash
set -e
echo "=== Mindray HL7 Forwarder ==="
# 1. Create a virtual environment if it doesn't exist
if [ ! -d "venv" ]; then
echo "Creating Python virtual environment..."
python3 -m venv venv
fi
# 2. Activate the virtual environment
source venv/bin/activate
# 3. Install dependencies quietly
echo "Checking and installing dependencies..."
pip install -r requirements.txt -q
# 4. Start the server
echo "Starting the application..."
python main.py
runtime.txt
-1
View File
@@ -1 +0,0 @@
python-3.12.0
schemas.py
-37
View File
@@ -1,37 +0,0 @@
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
class NormalizedVitals(BaseModel):
device_id: str
patient_id: str
timestamp: datetime
ip_address: Optional[str] = None
heart_rate: Optional[float] = None
spo2: Optional[float] = None
systolic_bp: Optional[float] = None
diastolic_bp: Optional[float] = None
map_bp: Optional[float] = None
respiratory_rate: Optional[float] = None
temperature: Optional[float] = None
present_fields: Optional[list] = None
class DeviceStatus(BaseModel):
device_id: str
ip_address: str
status: str
last_seen: datetime
class Config:
from_attributes = True
class TransmissionLogResponse(BaseModel):
id: int
reading_id: int
status: str
response_code: Optional[int]
error_message: Optional[str]
timestamp: datetime
class Config:
from_attributes = True
setup_service.sh
-50
View File
@@ -1,50 +0,0 @@
#!/bin/bash
# Exit on error
set -e
SERVICE_NAME="contec-forwarder"
SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service"
WORKING_DIR="/home/prathiyuman/Prathiyuman/ContecMonitor/fiveparaminte-main"
PYTHON_EXEC="${WORKING_DIR}/venv/bin/python"
SCRIPT_PATH="${WORKING_DIR}/main.py"
echo "=== Creating systemd service file ==="
sudo bash -c "cat > ${SERVICE_FILE}" <<EOF
[Unit]
Description=Contec Patient Monitor Vital Signs Forwarder
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=${WORKING_DIR}
Environment=PORT=8006
ExecStart=${PYTHON_EXEC} ${SCRIPT_PATH}
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
EOF
echo "=== Reloading systemd daemon ==="
sudo systemctl daemon-reload
echo "=== Enabling ${SERVICE_NAME} to start on boot ==="
sudo systemctl enable ${SERVICE_NAME}
echo "=== Starting ${SERVICE_NAME} service ==="
sudo systemctl restart ${SERVICE_NAME}
echo "=== Service Status ==="
sudo systemctl status ${SERVICE_NAME} --no-pager
echo ""
echo "=========================================================="
echo "SUCCESS: Contec Forwarder is now running in the background!"
echo "To check live service logs, run:"
echo " sudo journalctl -u contec-forwarder -f"
echo "=========================================================="
static/contec-vitals-sdk.js
-249
View File
@@ -1,249 +0,0 @@
/**
* ContecVitalsClient - A reusable JavaScript client for the Contec CMS7000PLUS Patient Monitor.
* Works in Web, React, React Native, and Node.js environments.
*
* Features:
* - Real-time streaming via WebSockets
* - Automatic HTTP polling fallback if WebSockets are blocked or disconnect
* - Exponential backoff auto-reconnection
* - Lifecycle callbacks for UI updates
*/
class ContecVitalsClient {
/**
* @param {Object} config - Client configuration options
* @param {string} config.baseUrl - Server URL (e.g. 'https://contectfiveparamonitor.onrender.com')
* @param {string} [config.deviceId] - Optional device filter (e.g. 'CMS7000PLUS_192.168.100.120')
* @param {number} [config.pollingInterval=3000] - Polling interval in ms if falling back to HTTP
* @param {function} [config.onVitals] - Callback when new vitals data arrives
* @param {function} [config.onStatusChange] - Callback when connection status changes ('idle', 'connecting', 'connected', 'polling', 'disconnected')
* @param {function} [config.onError] - Callback on error
*/
constructor(config) {
this.baseUrl = config.baseUrl.replace(/\/$/, '');
this.deviceId = config.deviceId || null;
this.pollingInterval = config.pollingInterval || 3000;
// Callbacks
this.onVitals = config.onVitals || (() => {});
this.onStatusChange = config.onStatusChange || (() => {});
this.onError = config.onError || (() => {});
// State management
this.ws = null;
this.pollTimer = null;
this.status = 'idle';
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
this.reconnectDelay = 1000;
this.isStarted = false;
this.lastVitalsTime = 0;
}
/**
* Start listening for vital signs.
*/
start() {
if (this.isStarted) return;
this.isStarted = true;
this.reconnectAttempts = 0;
this._setStatus('connecting');
this._connectWebSocket();
}
/**
* Stop listening and clean up all connections/timers.
*/
stop() {
this.isStarted = false;
this._clearPollTimer();
this._closeWebSocket();
this._setStatus('idle');
}
/**
* Change current status and trigger callback.
* @private
*/
_setStatus(newStatus) {
if (this.status !== newStatus) {
this.status = newStatus;
this.onStatusChange(newStatus);
}
}
/**
* Parse the base URL to construct the WS URL.
* @private
*/
_getWebSocketUrl() {
const wsProto = this.baseUrl.startsWith('https') ? 'wss' : 'ws';
const cleanUrl = this.baseUrl.replace(/^(https?:\/\/)/, '');
return `${wsProto}://${cleanUrl}/ws/vitals`;
}
/**
* Open WebSocket connection.
* @private
*/
_connectWebSocket() {
this._clearPollTimer();
const wsUrl = this._getWebSocketUrl();
try {
this.ws = new WebSocket(wsUrl);
this.ws.onopen = () => {
console.log('ContecVitalsClient: WebSocket connected');
this._setStatus('connected');
this.reconnectAttempts = 0;
this.reconnectDelay = 1000;
};
this.ws.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
this._processIncomingVitals(data);
} catch (err) {
this.onError(new Error('Failed to parse incoming WS message: ' + err.message));
}
};
this.ws.onerror = (err) => {
this.onError(err);
};
this.ws.onclose = (event) => {
this.ws = null;
if (!this.isStarted) return;
console.log(`ContecVitalsClient: WebSocket closed (code: ${event.code})`);
// Attempt reconnection
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++;
const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
console.log(`ContecVitalsClient: Reconnecting in ${delay}ms (Attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})...`);
this._setStatus('connecting');
setTimeout(() => {
if (this.isStarted) this._connectWebSocket();
}, delay);
} else {
// Exceeded Max WebSocket retries -> Fallback to HTTP Polling
console.log('ContecVitalsClient: WebSocket connection limit exceeded. Falling back to HTTP polling.');
this._startPolling();
}
};
} catch (err) {
this.onError(err);
this._startPolling();
}
}
/**
* Close WebSocket connection cleanly.
* @private
*/
_closeWebSocket() {
if (this.ws) {
// Remove event listeners to prevent onclose reconnect logic
this.ws.onopen = null;
this.ws.onmessage = null;
this.ws.onerror = null;
this.ws.onclose = null;
try {
this.ws.close();
} catch (e) {}
this.ws = null;
}
}
/**
* Start HTTP Polling.
* @private
*/
_startPolling() {
this._closeWebSocket();
this._setStatus('polling');
this._poll();
}
/**
* Perform single HTTP poll.
* @private
*/
async _poll() {
if (!this.isStarted || this.status !== 'polling') return;
try {
const url = `${this.baseUrl}/api/latest-readings?limit=1` + (this.deviceId ? `&device_id=${encodeURIComponent(this.deviceId)}` : '');
const response = await fetch(url);
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
const readings = await response.json();
if (Array.isArray(readings) && readings.length > 0) {
this._processIncomingVitals(readings[0]);
}
} catch (err) {
console.error('ContecVitalsClient polling error:', err.message);
this.onError(err);
}
// Schedule next poll
this.pollTimer = setTimeout(() => this._poll(), this.pollingInterval);
}
/**
* Clear the HTTP polling timer.
* @private
*/
_clearPollTimer() {
if (this.pollTimer) {
clearTimeout(this.pollTimer);
this.pollTimer = null;
}
}
/**
* Normalizes and emits the vitals data payload.
* @private
*/
_processIncomingVitals(data) {
if (!data) return;
// Filter by device if configured
if (this.deviceId && data.device_id !== this.deviceId) {
return;
}
// Normalize property names and types
const normalized = {
deviceId: data.device_id,
patientId: data.patient_id,
timestamp: data.timestamp,
heartRate: typeof data.heart_rate === 'number' ? Math.round(data.heart_rate) : null,
spo2: typeof data.spo2 === 'number' ? Math.round(data.spo2) : null,
temperature: typeof data.temperature === 'number' ? parseFloat(data.temperature.toFixed(1)) : null,
systolicBp: typeof data.systolic_bp === 'number' ? Math.round(data.systolic_bp) : null,
diastolicBp: typeof data.diastolic_bp === 'number' ? Math.round(data.diastolic_bp) : null,
mapBp: typeof data.map_bp === 'number' ? Math.round(data.map_bp) : null,
respiratoryRate: typeof data.respiratory_rate === 'number' ? Math.round(data.respiratory_rate) : null,
};
// Calculate BP display string helper
normalized.bloodPressureDisplay = (normalized.systolicBp && normalized.diastolicBp)
? `${normalized.systolicBp}/${normalized.diastolicBp}`
: null;
this.onVitals(normalized);
}
}
// Export for ES6/Node compatibility
if (typeof module !== 'undefined' && module.exports) {
module.exports = ContecVitalsClient;
} else if (typeof window !== 'undefined') {
window.ContecVitalsClient = ContecVitalsClient;
}
static/dashboard.html
-1859
View File
File diff suppressed because it is too large Load Diff
terminal_display.py
-246
View File
@@ -1,246 +0,0 @@
"""
Terminal Display for Live Vital Signs
======================================
Provides a rich, color-coded terminal dashboard showing real-time
vital signs received from patient monitors.
Works with both Mindray uMEC 100 and Contec CMS9200PLUS monitors.
"""
import logging
from datetime import datetime, timezone
from typing import Optional
from schemas import NormalizedVitals
logger = logging.getLogger(__name__)
# Try to use 'rich' for beautiful output, fall back to basic print
try:
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from rich.layout import Layout
from rich.live import Live
from rich import box
HAS_RICH = True
except ImportError:
HAS_RICH = False
logger.info("'rich' library not installed. Using basic terminal output.")
# === Normal Range Definitions ===
VITAL_RANGES = {
"heart_rate": {"low": 60, "high": 100, "critical_low": 40, "critical_high": 150, "unit": "bpm", "label": "Heart Rate"},
"spo2": {"low": 95, "high": 100, "critical_low": 90, "critical_high": 101, "unit": "%", "label": "SpO2"},
"systolic_bp": {"low": 90, "high": 140, "critical_low": 70, "critical_high": 180, "unit": "mmHg", "label": "BP Systolic"},
"diastolic_bp": {"low": 60, "high": 90, "critical_low": 40, "critical_high": 120, "unit": "mmHg", "label": "BP Diastolic"},
"map_bp": {"low": 70, "high": 105, "critical_low": 60, "critical_high": 130, "unit": "mmHg", "label": "MAP"},
"respiratory_rate": {"low": 12, "high": 20, "critical_low": 8, "critical_high": 30, "unit": "/min", "label": "Resp Rate"},
"temperature": {"low": 36.1,"high": 37.2,"critical_low": 35.0,"critical_high": 39.0,"unit": "C", "label": "Temperature"},
}
def get_vital_color(param_name: str, value: Optional[float]) -> str:
"""Return a color name based on whether the value is normal, warning, or critical."""
if value is None:
return "dim"
ranges = VITAL_RANGES.get(param_name)
if not ranges:
return "white"
if value <= ranges["critical_low"] or value >= ranges["critical_high"]:
return "bold red"
elif value < ranges["low"] or value > ranges["high"]:
return "yellow"
else:
return "bold green"
def format_vital_value(param_name: str, value: Optional[float]) -> str:
"""Format a vital sign value with its unit."""
if value is None:
return "---"
ranges = VITAL_RANGES.get(param_name, {})
unit = ranges.get("unit", "")
# Format based on typical precision
if param_name == "temperature":
return f"{value:.1f} {unit}"
elif param_name == "spo2":
return f"{value:.0f} {unit}"
else:
return f"{value:.0f} {unit}"
class TerminalDisplay:
"""Manages the live terminal display of vital signs."""
def __init__(self):
self._latest_vitals: dict[str, NormalizedVitals] = {} # keyed by device_id
self._last_update: Optional[datetime] = None
self._connection_count: int = 0
self._message_count: int = 0
if HAS_RICH:
self.console = Console()
else:
self.console = None
def update_vitals(self, vitals: NormalizedVitals):
"""Update the display with new vital signs data."""
self._latest_vitals[vitals.device_id] = vitals
self._last_update = datetime.now(timezone.utc)
self._message_count += 1
self._print_vitals(vitals)
def connection_opened(self, client_ip: str, port: int):
"""Track a new connection."""
self._connection_count += 1
if HAS_RICH:
self.console.print(
f" [bold cyan]Connected:[/] {client_ip} on port {port} "
f"(total connections: {self._connection_count})"
)
else:
print(f" Connected: {client_ip} on port {port}")
def connection_closed(self, client_ip: str, port: int):
"""Track a closed connection."""
if HAS_RICH:
self.console.print(f" [dim]Disconnected:[/dim] {client_ip} on port {port}")
else:
print(f" Disconnected: {client_ip} on port {port}")
def _print_vitals(self, vitals: NormalizedVitals):
"""Print vital signs to terminal."""
if HAS_RICH:
self._print_vitals_rich(vitals)
else:
self._print_vitals_basic(vitals)
def _print_vitals_rich(self, vitals: NormalizedVitals):
"""Print vital signs using rich library for beautiful output."""
table = Table(
title=f"Live Vitals -- Device: {vitals.device_id} | Patient: {vitals.patient_id}",
box=box.HEAVY_EDGE,
show_header=True,
header_style="bold white on dark_blue",
title_style="bold white",
border_style="blue",
padding=(0, 2),
)
table.add_column("Parameter", style="bold white", width=16)
table.add_column("Value", justify="right", width=14)
table.add_column("Status", justify="center", width=10)
table.add_column("Normal Range", justify="center", style="dim", width=14)
vital_params = [
("heart_rate", vitals.heart_rate),
("spo2", vitals.spo2),
("systolic_bp", vitals.systolic_bp),
("diastolic_bp", vitals.diastolic_bp),
("map_bp", vitals.map_bp),
("respiratory_rate", vitals.respiratory_rate),
("temperature", vitals.temperature),
]
for param_name, value in vital_params:
ranges = VITAL_RANGES.get(param_name, {})
color = get_vital_color(param_name, value)
formatted = format_vital_value(param_name, value)
label = ranges.get("label", param_name)
# Status indicator
if value is None:
status = "[dim]-[/dim]"
elif color == "bold green":
status = "[green]Normal[/green]"
elif color == "yellow":
status = "[yellow]Warning[/yellow]"
elif color == "bold red":
status = "[red]ALERT[/red]"
else:
status = "[dim]-[/dim]"
# Normal range
low = ranges.get("low", "?")
high = ranges.get("high", "?")
unit = ranges.get("unit", "")
range_str = f"{low}-{high} {unit}"
table.add_row(
label,
f"[{color}]{formatted}[/{color}]",
status,
range_str,
)
self.console.print()
self.console.print(table)
# Timestamp footer
time_str = vitals.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC") if vitals.timestamp else "N/A"
self.console.print(
f" [dim]Last update: {time_str} | "
f"Messages received: {self._message_count} | "
f"Active connections: {self._connection_count}[/dim]"
)
def _print_vitals_basic(self, vitals: NormalizedVitals):
"""Fallback plain-text vital signs display."""
print("\n" + "=" * 65)
print(f" LIVE VITALS — Device: {vitals.device_id} | Patient: {vitals.patient_id}")
print("=" * 65)
vital_params = [
("Heart Rate", vitals.heart_rate, "bpm"),
("SpO2", vitals.spo2, "%"),
("BP Systolic", vitals.systolic_bp, "mmHg"),
("BP Diastolic",vitals.diastolic_bp, "mmHg"),
("MAP", vitals.map_bp, "mmHg"),
("Resp Rate", vitals.respiratory_rate, "/min"),
("Temperature", vitals.temperature, "C"),
]
for label, value, unit in vital_params:
if value is not None:
if "temp" in label.lower():
val_str = f"{value:.1f} {unit}"
else:
val_str = f"{value:.0f} {unit}"
# Simple text status
ranges = VITAL_RANGES.get(label.lower().replace(" ", "_"), {})
if ranges:
if value < ranges.get("critical_low", 0) or value > ranges.get("critical_high", 999):
status = " *** CRITICAL ***"
elif value < ranges.get("low", 0) or value > ranges.get("high", 999):
status = " ! WARNING"
else:
status = " OK"
else:
status = ""
print(f" {label:<16} {val_str:>12}{status}")
else:
print(f" {label:<16} {'---':>12}")
time_str = vitals.timestamp.strftime("%H:%M:%S") if vitals.timestamp else "N/A"
print(f"\n Time: {time_str} | Messages: {self._message_count}")
print("-" * 65)
# Global singleton
_display: Optional[TerminalDisplay] = None
def get_terminal_display() -> TerminalDisplay:
"""Get or create the global terminal display instance."""
global _display
if _display is None:
_display = TerminalDisplay()
return _display