Compare commits

...
6 Commits
34 changed files with 168 additions and 8326 deletions
+60 -54
View File
@@ -19,10 +19,14 @@ def get_http_client() -> httpx.AsyncClient:
)
return _http_client
def get_cloud_base_url() -> str:
url = getattr(settings, "cloud_api_url", None) or os.getenv("CLOUD_API_URL") or "https://services-api.a2zhealth.in/v1/contec-monitor"
return url.rstrip("/")
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 concurrently.
and to the Cloud server concurrently.
Returns True if at least one forward succeeds.
"""
headers = {"Content-Type": "application/json"}
@@ -37,67 +41,54 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
client = get_http_client()
# 1. Forward to primary target API (e.g. Injomo Dev Workflow)
async def _forward_primary() -> bool:
primary_url = settings.target_api_url
if not primary_url:
async def _post_target(url: str, label: str) -> bool:
if not url:
return False
curr_port = os.getenv("PORT", "8000")
if f":{curr_port}" in url and ("localhost" in url or "127.0.0.1" in url):
return False
try:
primary_headers = headers.copy()
req_headers = headers.copy()
if settings.api_token:
primary_headers["Authorization"] = f"Bearer {settings.api_token}"
response = await client.post(
primary_url,
json=payload,
headers=primary_headers
)
req_headers["Authorization"] = f"Bearer {settings.api_token}"
url_to_call = url
if settings.api_token and "?" not in url and "/api/push-vitals" in url:
url_to_call = f"{url}?api_key={settings.api_token}"
response = await client.post(url_to_call, json=payload, headers=req_headers)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}")
logger.info(f"Successfully forwarded vitals to {label} ({url}) for device {vitals.device_id}")
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to primary API: {type(e).__name__} ({err_msg})")
logger.warning(f"Error forwarding to {label} ({url}): {type(e).__name__} ({err_msg})")
return False
# 2. Forward to Render Cloud Server
async def _forward_render() -> bool:
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
try:
url_with_key = render_url
if settings.api_token:
url_with_key = f"{render_url}?api_key={settings.api_token}"
targets = []
if settings.target_api_url:
targets.append((settings.target_api_url, "Primary Target API"))
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}")
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to Render Cloud: {type(e).__name__} ({err_msg})")
return False
cloud_base = get_cloud_base_url()
is_on_cloud = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not (is_on_cloud and "onrender.com" in cloud_base):
cloud_push_url = f"{cloud_base}/api/push-vitals" if not cloud_base.endswith("/api/push-vitals") else cloud_base
if cloud_push_url not in [t[0] for t in targets]:
targets.append((cloud_push_url, "Cloud Server"))
results = await asyncio.gather(_forward_primary(), _forward_render(), return_exceptions=True)
for extra_url in getattr(settings, "forward_urls", []):
if extra_url and extra_url not in [t[0] for t in targets]:
targets.append((extra_url, "Configured Forward URL"))
tasks = [_post_target(url, lbl) for url, lbl in targets]
results = await asyncio.gather(*tasks, return_exceptions=True)
return any(r is True for r in results if not isinstance(r, Exception))
async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -> bool:
"""
Synchronizes patient demographics assignment to Render Cloud Server.
Synchronizes patient demographics assignment to Cloud and LAN servers.
Ensures local and remote dashboards always show the exact same patient.
"""
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
client = get_http_client()
render_assign_url = "https://contectfiveparamonitor.onrender.com/api/patient/assign"
payload = {
"device_id": device_id,
"mrn": patient_data.get("mrn") or patient_data.get("patient_id") or "",
@@ -107,14 +98,29 @@ async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -
"bed_id": patient_data.get("bed_id"),
"patient_type": patient_data.get("patient_type")
}
try:
url_with_key = render_assign_url
if settings.api_token:
url_with_key = f"{render_assign_url}?api_key={settings.api_token}"
resp = await client.post(url_with_key, json=payload, headers={"Content-Type": "application/json"})
resp.raise_for_status()
logger.info(f"Successfully synced patient {payload['mrn']} ({payload['name']}) to Render Cloud for {device_id}")
return True
except Exception as e:
logger.warning(f"Could not sync patient assignment to Render Cloud: {type(e).__name__} ({e})")
return False
targets = []
cloud_base = get_cloud_base_url()
targets.append(f"{cloud_base}/api/patient/assign")
for extra_url in getattr(settings, "forward_urls", []):
if "/api/push-vitals" in extra_url:
assign_url = extra_url.replace("/api/push-vitals", "/api/patient/assign")
if assign_url not in targets:
targets.append(assign_url)
async def _sync_one(url: str) -> bool:
try:
url_with_key = url
if settings.api_token:
url_with_key = f"{url}?api_key={settings.api_token}"
resp = await client.post(url_with_key, json=payload, headers={"Content-Type": "application/json"})
resp.raise_for_status()
logger.info(f"Successfully synced patient {payload['mrn']} to {url} for {device_id}")
return True
except Exception as e:
logger.warning(f"Could not sync patient assignment to {url}: {type(e).__name__} ({e})")
return False
results = await asyncio.gather(*[_sync_one(u) for u in targets], return_exceptions=True)
return any(r is True for r in results if not isinstance(r, Exception))
+7 -2
View File
@@ -1,13 +1,18 @@
{
"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],
"contec_ports": [511, 512, 513, 514, 515, 516, 517, 518, 519, 520, 8001, 8002, 8300, 9000, 9100, 9200, 10008, 12345, 61501],
"monitor_ip": "",
"monitor_model": "CMS8500",
"device_models": {
"192.168.100.120": "CMS7000PLUS"
},
"target_api_url": "https://innov-dev.beta.injomo.com/workflow.trigger/69b2712a7866bf86ca0060c3",
"target_api_url": "https://services-api.a2zhealth.in/v1/contec-monitor",
"cloud_api_url": "https://services-api.a2zhealth.in/v1/contec-monitor",
"forward_urls": [
"http://192.168.100.104:10000/api/push-vitals",
"https://services-api.a2zhealth.in/v1/contec-monitor"
],
"api_token": "",
"retry_interval_seconds": 60,
"log_file": "hl7_forwarder.log",
+8 -2
View File
@@ -5,17 +5,23 @@ 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]
contec_ports: list = [511, 515, 516, 517, 518, 519, 520, 61501]
monitor_ip: str = ""
monitor_model: str = "CMS7000PLUS"
device_models: dict = {}
target_api_url: str = "https://api.example.com/vitals"
target_api_url: str = "https://services-api.a2zhealth.in/v1/contec-monitor"
cloud_api_url: str = "https://services-api.a2zhealth.in/v1/contec-monitor"
forward_urls: list = [
"http://192.168.100.104:10000/api/push-vitals",
"https://services-api.a2zhealth.in/v1/contec-monitor"
]
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:
-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/
-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
-16
View File
@@ -1,16 +0,0 @@
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONPATH=/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}
-1
View File
@@ -1 +0,0 @@
web: python3 -m uvicorn main:app --host 0.0.0.0 --port $PORT
View File
-120
View File
@@ -1,120 +0,0 @@
import httpx
import logging
import asyncio
import os
from typing import Optional
from schemas import NormalizedVitals
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=5.0, read=5.0, write=5.0, pool=10.0),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=50)
)
return _http_client
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 concurrently.
Returns True if at least one forward succeeds.
"""
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)
if not payload.get("patient_id"):
payload["patient_id"] = "UNKNOWN"
client = get_http_client()
# 1. Forward to primary target API (e.g. Injomo Dev Workflow)
async def _forward_primary() -> bool:
primary_url = settings.target_api_url
if not primary_url:
return False
try:
primary_headers = headers.copy()
if settings.api_token:
primary_headers["Authorization"] = f"Bearer {settings.api_token}"
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}")
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to primary API: {type(e).__name__} ({err_msg})")
return False
# 2. Forward to Render Cloud Server
async def _forward_render() -> bool:
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
try:
url_with_key = render_url
if settings.api_token:
url_with_key = f"{render_url}?api_key={settings.api_token}"
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}")
return True
except Exception as e:
err_msg = str(e).strip() or "timeout / connection reset"
logger.warning(f"Error forwarding to Render Cloud: {type(e).__name__} ({err_msg})")
return False
results = await asyncio.gather(_forward_primary(), _forward_render(), return_exceptions=True)
return any(r is True for r in results if not isinstance(r, Exception))
async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -> bool:
"""
Synchronizes patient demographics assignment to Render Cloud Server.
Ensures local and remote dashboards always show the exact same patient.
"""
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if is_on_render:
return True
client = get_http_client()
render_assign_url = "https://contectfiveparamonitor.onrender.com/api/patient/assign"
payload = {
"device_id": device_id,
"mrn": patient_data.get("mrn") or patient_data.get("patient_id") or "",
"name": patient_data.get("name") or patient_data.get("patient_name") or "Unknown",
"dob": patient_data.get("dob") or patient_data.get("patient_dob"),
"gender": patient_data.get("gender") or patient_data.get("patient_gender"),
"bed_id": patient_data.get("bed_id"),
"patient_type": patient_data.get("patient_type")
}
try:
url_with_key = render_assign_url
if settings.api_token:
url_with_key = f"{render_assign_url}?api_key={settings.api_token}"
resp = await client.post(url_with_key, json=payload, headers={"Content-Type": "application/json"})
resp.raise_for_status()
logger.info(f"Successfully synced patient {payload['mrn']} ({payload['name']}) to Render Cloud for {device_id}")
return True
except Exception as e:
logger.warning(f"Could not sync patient assignment to Render Cloud: {type(e).__name__} ({e})")
return False
-169
View File
@@ -1,169 +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:
# 1. Fetch untransmitted readings in a quick DB session and close it immediately
db = SessionLocal()
readings_to_retry = []
try:
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
)
})
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}")
await asyncio.sleep(settings.retry_interval_seconds)
async def sync_cloud_state():
"""
Periodically syncs all assigned patients and their latest non-null vitals to Render Cloud.
Ensures that whenever Render restarts or deploys with an empty DB, the cloud dashboard
is immediately populated with the same data as localhost.
"""
import os
if os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", ""):
return # Do not run sync loop if we are running inside Render
from models import Patient
from api_client import sync_patient_assignment_to_cloud, forward_vitals_to_api
from datetime import datetime, timezone
from sqlalchemy import or_
logger.info("Starting background task for periodic cloud state sync to Render...")
await asyncio.sleep(5)
while True:
try:
db = SessionLocal()
try:
patients = db.query(Patient).filter(
Patient.patient_id != "UNKNOWN",
Patient.mrn != "UNKNOWN"
).all()
for p in patients:
if not p.device_id:
continue
# 1. Sync patient demographics assignment to Render
p_data = {
"mrn": p.mrn or p.patient_id,
"name": p.name,
"gender": p.gender,
"dob": p.dob,
"bed_id": p.bed_id,
"patient_type": getattr(p, "patient_type", None)
}
await sync_patient_assignment_to_cloud(p_data, p.device_id)
# 2. Forward latest non-null vitals to Render (Option B)
from sqlalchemy import or_
from datetime import datetime, timezone
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
VitalReading.device_id == p.device_id
)
hr_r = db.query(VitalReading.heart_rate).filter(patient_filters, VitalReading.heart_rate != None).order_by(VitalReading.timestamp.desc()).first()
spo2_r = db.query(VitalReading.spo2).filter(patient_filters, VitalReading.spo2 != None).order_by(VitalReading.timestamp.desc()).first()
bp_r = db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp).filter(patient_filters, VitalReading.systolic_bp != None).order_by(VitalReading.timestamp.desc()).first()
temp_r = db.query(VitalReading.temperature).filter(patient_filters, VitalReading.temperature != None).order_by(VitalReading.timestamp.desc()).first()
temp2_r = db.query(VitalReading.temperature2).filter(patient_filters, VitalReading.temperature2 != None).order_by(VitalReading.timestamp.desc()).first()
temp_diff_r = db.query(VitalReading.temperature_diff).filter(patient_filters, VitalReading.temperature_diff != None).order_by(VitalReading.timestamp.desc()).first()
rr_r = db.query(VitalReading.respiratory_rate).filter(patient_filters, VitalReading.respiratory_rate != None).order_by(VitalReading.timestamp.desc()).first()
if hr_r or spo2_r or temp_r or bp_r:
merged = NormalizedVitals(
device_id=p.device_id,
patient_id=p.mrn or p.patient_id,
patient_name=p.name,
patient_gender=p.gender,
patient_dob=p.dob,
bed_id=p.bed_id,
patient_type=getattr(p, "patient_type", None),
timestamp=datetime.now(timezone.utc),
heart_rate=hr_r[0] if hr_r else None,
spo2=spo2_r[0] if spo2_r else None,
systolic_bp=bp_r[0] if bp_r else None,
diastolic_bp=bp_r[1] if bp_r else None,
map_bp=bp_r[2] if bp_r else None,
temperature=temp_r[0] if temp_r else None,
temperature2=temp2_r[0] if temp2_r else None,
temperature_diff=temp_diff_r[0] if temp_diff_r else None,
respiratory_rate=rr_r[0] if rr_r else None
)
await forward_vitals_to_api(merged)
finally:
db.close()
except Exception as e:
logger.error(f"Error in sync_cloud_state background task: {e}")
await asyncio.sleep(15)
-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
-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."
-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"
}
-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()
-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.")
-530
View File
@@ -1,530 +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": "temperature2",
"T1": "temperature",
"T2": "temperature2",
"BODY_TEMP": "temperature",
"TEMP-1": "temperature",
"TEMP-2": "temperature2",
"BODYTEMP": "temperature",
"TD": "temperature_diff",
}
# 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, PV1 and OBX segments
for seg in segments[1:]:
fields = seg.split("|")
if not fields:
continue
seg_name = fields[0]
if seg_name == "PID":
# PID-3: MRN / Patient ID
if len(fields) > 3 and fields[3]:
patient_id = fields[3].split("^")[0].strip()
vitals_dict["patient_id"] = patient_id
# PID-5: Patient Name (format: FamilyName^GivenName^...)
if len(fields) > 5 and fields[5]:
name_parts = fields[5].split("^")
family = name_parts[0].strip() if len(name_parts) > 0 else ""
given = name_parts[1].strip() if len(name_parts) > 1 else ""
full_name = f"{given} {family}".strip() or family or given
if full_name:
vitals_dict["patient_name"] = full_name
# PID-7: Date of Birth (YYYYMMDD or YYYY-MM-DD)
if len(fields) > 7 and fields[7]:
vitals_dict["patient_dob"] = fields[7].strip()
# PID-8: Administrative Sex (M / F / U / O)
if len(fields) > 8 and fields[8]:
vitals_dict["patient_gender"] = fields[8].strip().upper()
# PID-35: Patient Category (ADULT / PEDIATRIC / NEONATAL)
if len(fields) > 35 and fields[35]:
vitals_dict["patient_type"] = fields[35].strip()
elif seg_name == "PV1":
# PV1-3: Assigned Patient Location (Room^Bed^...)
if len(fields) > 3 and fields[3]:
location_parts = fields[3].split("^")
# Use second component (Bed) if present, else first (Room)
bed = location_parts[1].strip() if len(location_parts) > 1 else location_parts[0].strip()
if bed:
vitals_dict["bed_id"] = bed
# PV1-18: Patient Type
if len(fields) > 18 and fields[18]:
vitals_dict["patient_type"] = fields[18].strip()
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: 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
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/47-byte packet (Subtype 22) — contains NIBP values and alarm limits.
Verified layout:
[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, 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) < 21:
return False
if "present_fields" not in vitals_dict:
vitals_dict["present_fields"] = []
logger.debug("DEBUG_NIBP_PKT len=%d nibp_hex=%s", len(data), data[8:24].hex())
nibp_sys = read_u16_le(data, 15)
nibp_dia = read_u16_le(data, 17)
nibp_map = read_u16_le(data, 19)
sentinels = (9999, 32767, 65535, 0)
if nibp_sys is not None and nibp_sys not in sentinels and 40 <= nibp_sys <= 260:
vitals_dict["systolic_bp"] = float(nibp_sys)
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 not in sentinels and 20 <= nibp_dia <= 200:
vitals_dict["diastolic_bp"] = float(nibp_dia)
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 not in sentinels and 20 <= nibp_map <= 250:
vitals_dict["map_bp"] = float(nibp_map)
if "map_bp" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("map_bp")
return True
def _parse_56_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 56-byte packet (Subtype 23) — contains float32 Temperature values for T1 and T2.
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"] = []
temp1 = read_f32_le(data, 8)
temp2 = read_f32_le(data, 12)
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
t1_val = round(temp1, 1) if is_real_temp(temp1) else None
t2_val = round(temp2, 1) if is_real_temp(temp2) else None
vitals_dict["temperature"] = t1_val
vitals_dict["temperature2"] = t2_val
if t1_val is not None and t2_val is not None:
vitals_dict["temperature_diff"] = round(abs(t1_val - t2_val), 1)
else:
vitals_dict["temperature_diff"] = None
if "temperature" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature")
if "temperature2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature2")
if "temperature_diff" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("temperature_diff")
return (t1_val is not None or t2_val is not None)
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")
# Check SpO2 at standard offset 264
spo2_val = read_u16_le(data, 264)
if spo2_val is not None and 50 <= spo2_val <= 100:
vitals_dict["spo2"] = float(spo2_val)
if "spo2" not in vitals_dict["present_fields"]:
vitals_dict["present_fields"].append("spo2")
# Pulse Rate / ECG HR is at offset 266 in the footer
ecg_hr = read_u16_le(data, 266)
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")
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'}"
)
return True
def _parse_286_byte_packet(data: bytes, vitals_dict: dict) -> bool:
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_288_byte_packet(data: bytes, vitals_dict: dict) -> bool:
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_341_byte_packet(data: bytes, vitals_dict: dict) -> bool:
return _parse_spo2_waveform_packet(data, vitals_dict)
def _parse_989_byte_packet(data: bytes, vitals_dict: dict) -> bool:
"""
Parse 989-byte waveform packet (Subtype 20 / 14h).
Verified layout:
[8..903] ECG waveform samples (8-bit unsigned, 896 samples @ ~250Hz)
[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"] = []
# 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")
logger.debug("DEBUG_ECG_PKT len=%d tail_hex=%s", len(data), data[890:989].hex())
hr_val = read_u16_le(data, 904)
if hr_val is not None and hr_val not in (65535, 32767, 9999, 0) and 20 <= hr_val <= 300:
if vitals_dict.get("heart_rate") is None:
vitals_dict["heart_rate"] = float(hr_val)
if "heart_rate" not in vitals_dict.setdefault("present_fields", []):
vitals_dict["present_fields"].append("heart_rate")
return True
def _find_vitals_block_in_stream(data: bytes, client_ip: str) -> Optional[NormalizedVitals]:
"""
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 = {
"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):
if _parse_45_byte_packet(pkt_data, vitals_dict):
found_any = True
elif pkt_len == 56:
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:
if _parse_989_byte_packet(pkt_data, vitals_dict):
found_any = True
# Handled specifically by known packet sizes above
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
-726
View File
@@ -1,726 +0,0 @@
import asyncio
import logging
import re
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}"
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 is_telemetry_word(s_lower: str) -> bool:
"""Check if string matches hardware telemetry or alarm words."""
if s_lower in TELEMETRY_WORDS:
return True
tokens = set(re.findall(r'[a-zA-Z0-9]+', s_lower))
alarm_tokens = {
'sensor', 'alarm', 'lead', 'leads', 'searching', 'pulse', 'cuff', 'probe',
'off', 'fall', 'leak', 'motion', 'weak', 'arr', 'asystole', 'vfib', 'vtac',
'pvc', 'brady', 'tachy', 'error', 'ecg', 'spo2', 'nibp', 'resp', 'temp',
'mv', 'bpm', 'mmhg'
}
if any(tok in alarm_tokens for tok in tokens):
return True
return False
def extract_demographics_from_packet(data: bytes, client_ip: str) -> dict:
"""
Extracts real patient demographics from Contec packets.
Only extracts clean, legitimate patient fields — strictly rejects sensor alarms,
waveform noise, and hardware telemetry.
"""
patient_type = None
data_upper = data.upper()
if b'PEDIATRIC' in data_upper or b'PED' in data_upper:
patient_type = 'Pediatric'
elif b'NEONATE' in data_upper or b'NEONATAL' in data_upper or b'NEO' in data_upper:
patient_type = 'Neonatal'
elif b'ADULT' in data_upper or b'ADU' in data_upper:
patient_type = 'Adult'
# Strictly reject binary waveform/measurement packets from generating false patient records.
# Patient MRN/Name extraction requires an authentic HL7 message with PID or MSH.
if b'PID|' not in data and b'MSH|' not in data:
if patient_type:
return {"patient_type": patient_type}
return None
clean = []
# Match strings of printable alphanumeric and reasonable characters, allowing single spaces
for m in re.finditer(rb'[a-zA-Z0-9_\-\.]{2,}(?:\s+[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 is_telemetry_word(s_lower):
continue
if len(set(s)) == 1:
continue
if set(s).issubset({'.', '-', '_', '/', '\\'}):
continue
if len(s) > 6 and re.match(r'^[A-Za-z0-9]+$', 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 and not patient_type:
return None
name = None
mrn = None
dob = None
gender = None
for s in clean:
s_up = s.upper()
s_low = s.lower()
# Check gender first
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
if is_telemetry_word(s_low):
continue
# Date of birth: 4-digit birth year between 1920 and 2026, or YYYY-MM-DD
if re.match(r'^(19\d\d|20[01]\d|202[0-6])$', 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: contains letters and a vowel, length between 3 and 40
if re.search(r'^[A-Za-z\s\.\-]{3,40}$', s) and re.search(r'[aeiouyAEIOUY]', s) and not name:
if not is_telemetry_word(s_low):
name = s
continue
# MRN candidate: numeric or alphanumeric ID, 2..15 chars, not a birth year
if not mrn and re.match(r'^[A-Za-z0-9\-]{2,15}$', s) and not is_telemetry_word(s_low):
if not (s.isdigit() and 1920 <= int(s) <= 2030):
if s.isdigit() and len(s) > 10:
continue
mrn = s
continue
# Accept if valid MRN or Name was detected
if mrn or name:
clean_mrn = mrn or name
clean_name = name or f"Patient {clean_mrn}"
return {
"name": clean_name,
"mrn": clean_mrn,
"patient_id": clean_mrn,
"dob": dob,
"gender": gender,
"patient_type": patient_type,
}
elif patient_type:
return {
"patient_type": patient_type,
}
return None
async def apply_auto_demographics(client_ip: str, demo: dict):
from config import settings
from api_client import sync_patient_assignment_to_cloud
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")
patient_type = demo.get("patient_type")
current = device_patient_map.get(device_id, {})
# If only patient_type updated
if not mrn and not name:
if patient_type and current.get("patient_type") != patient_type:
current["patient_type"] = patient_type
device_patient_map[device_id] = current
return
if (current.get("name") == name and current.get("mrn") == mrn and
current.get("dob") == dob and current.get("gender") == gender and
current.get("patient_type") == patient_type):
return
logger.info(f"*** REAL PATIENT DEMOGRAPHICS DETECTED FROM MACHINE ({client_ip}): Name={name}, MRN={mrn}, DOB={dob}, Gender={gender}, Type={patient_type} ***")
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,
"patient_type": patient_type or current.get("patient_type"),
}
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,
"patient_type": p_data.get("patient_type"),
})
except Exception as e:
logger.debug(f"Failed to broadcast auto patient: {e}")
# Forward to Render Cloud Server so remote is kept in sync
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
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}")
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)
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
# Dynamically extract and apply patient demographics sent from the machine
try:
demo = extract_demographics_from_packet(data, client_ip)
if demo:
await apply_auto_demographics(client_ip, demo)
except Exception as ex:
logger.debug(f"Demographics extraction error: {ex}")
buffer += data
# ---------------------------------------------------------------
# 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 = {}
# 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,
"patient_type": getattr(patient, "patient_type", None),
}
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")
cached.patient_type = patient_data.get("patient_type")
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] = {}
# 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")
vitals.patient_type = assigned.get("patient_type")
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,
"patient_type": getattr(vitals, "patient_type", None),
}
# Check if incoming packet has fresh waveforms
has_fresh_waveforms = (vitals.ecg_wave is not None) or (vitals.pleth_wave is not None)
# 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 present values
for field in active_fields:
val = getattr(vitals, field, None)
if val is not None:
setattr(cached, field, val)
device_field_timestamps[device_id][field] = now
else:
setattr(cached, field, None)
device_field_timestamps[device_id].pop(field, None)
# Clean up stale fields:
# Live vitals (HR, SpO2, Temp, RR) have a 15-second window
# NIBP (BP, MAP) has a 15-minute window because it is measured intermittently
for field in list(device_field_timestamps[device_id].keys()):
last_updated = device_field_timestamps[device_id][field]
timeout = 900.0 if field in ("systolic_bp", "diastolic_bp", "map_bp") else 15.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
display = get_terminal_display()
display.update_vitals(vitals)
logger.info(
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}"
)
# 3. Broadcast to live WebSockets dashboard in real-time
try:
from routers.dashboard import get_ws_manager
ws_manager = get_ws_manager()
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
vitals_dict["patient_type"] = getattr(vitals, "patient_type", None)
await ws_manager.broadcast_vitals(vitals_dict)
except Exception as e:
logger.debug(f"WebSocket broadcast failed: {e}")
# 4. Asynchronous background persistence and cloud forwarding (completely non-blocking, throttled to 1s)
now = datetime.now(timezone.utc)
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_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:
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)
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:
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,
patient_type=getattr(vitals, "patient_type", None),
device_id=vitals.device_id
)
db.add(patient)
else:
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
if vitals.patient_dob:
patient.dob = vitals.patient_dob
if vitals.bed_id:
patient.bed_id = vitals.bed_id
if getattr(vitals, "patient_type", None):
patient.patient_type = vitals.patient_type
if vitals.device_id:
patient.device_id = vitals.device_id
if patient:
if not vitals.patient_id or vitals.patient_id == "UNKNOWN":
vitals.patient_id = patient.mrn or patient.patient_id
if not vitals.patient_name or vitals.patient_name == "Unknown":
vitals.patient_name = patient.name
if not vitals.patient_gender and patient.gender:
vitals.patient_gender = patient.gender
if not vitals.patient_dob and patient.dob:
vitals.patient_dob = patient.dob
if not vitals.bed_id and patient.bed_id:
vitals.bed_id = patient.bed_id
if not getattr(vitals, "patient_type", None) and getattr(patient, "patient_type", None):
vitals.patient_type = patient.patient_type
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id or (patient.mrn if patient else "UNKNOWN"),
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,
temperature2=vitals.temperature2,
temperature_diff=vitals.temperature_diff,
transmitted=False
)
db.add(reading)
db.commit()
db.refresh(reading)
reading_id = reading.id
except Exception as e:
logger.error(f"Background database operation failed: {e}")
db.rollback()
finally:
db.close()
# 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()
-22
View File
@@ -1,22 +0,0 @@
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
from sqlalchemy.orm import sessionmaker, declarative_base
from config import settings
connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
engine = create_engine(
settings.database_url,
connect_args=connect_args,
poolclass=NullPool
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
-246
View File
@@ -1,246 +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, sync_cloud_state
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):
# Protect against Windows IOCP Proactor crashes (WinError 64 / 121 / 10054 on accept)
import sys
if sys.platform == "win32":
try:
loop = asyncio.get_running_loop()
def silence_windows_iocp_errors(loop, context):
msg = str(context.get("message", ""))
exc = context.get("exception")
winerr = getattr(exc, "winerror", None)
if winerr in (64, 121, 10054) or "Accept failed on a socket" in msg or "The specified network name is no longer available" in msg:
return
loop.default_exception_handler(context)
loop.set_exception_handler(silence_windows_iocp_errors)
except Exception:
pass
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}/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}")
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 background retry and cloud state sync tasks
retry_task = asyncio.create_task(retry_failed_transmissions())
cloud_sync_task = asyncio.create_task(sync_cloud_state())
# 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()
cloud_sync_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")
# 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")
from fastapi.responses import Response
@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
return Response(status_code=204, media_type="image/x-icon")
# === 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)
-49
View File
@@ -1,49 +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) # MRN from monitor PID-3
mrn = Column(String, index=True, nullable=True) # explicit MRN (same as patient_id)
name = Column(String, nullable=True) # full name from PID-5
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
patient_type = Column(String, nullable=True) # Adult / Pediatric / Neonatal
device_id = Column(String, nullable=True, index=True) # assigned machine / device_id
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)
temperature2 = Column(Float, nullable=True)
temperature_diff = 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))
-17
View File
@@ -1,17 +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: PYTHONPATH
value: /opt/render/project/src
- key: PORT
value: 10000
- key: PYTHON_VERSION
value: 3.12.0
- key: TARGET_API_URL
sync: false
- key: API_TOKEN
sync: false
-9
View File
@@ -1,9 +0,0 @@
fastapi
uvicorn
sqlalchemy
pydantic
pydantic-settings
httpx
gunicorn
rich
websockets
-1
View File
@@ -1 +0,0 @@
# Python package marker
-696
View File
@@ -1,696 +0,0 @@
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 datetime import datetime, timezone, timedelta
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
patient_type: 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(),
headers={
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
)
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(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_criteria = []
if mrn:
mrn_clean = mrn.strip()
patient_criteria.append(or_(Patient.mrn == mrn_clean, Patient.patient_id == mrn_clean))
if name:
name_clean = name.strip()
patient_criteria.append(Patient.name.ilike(f"%{name_clean}%"))
if dob:
dob_clean = dob.strip().replace("-", "").replace("/", "")
patient_criteria.append(or_(
Patient.dob.ilike(f"%{dob.strip()}%"),
Patient.dob.ilike(f"%{dob_clean}%"),
Patient.dob.ilike(f"%{dob.strip()[:4]}%") if len(dob.strip()) >= 4 else False
))
if patient_criteria:
from sqlalchemy import and_
matching_patients = db.query(Patient).filter(and_(*patient_criteria)).all()
if not matching_patients:
return []
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 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))
else:
return []
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.
Strictly enforce live data timeouts:
- Continuous parameters (HR, SpO2, RR, Temp): max 25 seconds age.
- Intermittent parameters (NIBP): max 15 minutes age.
- Never backfill from stale historical readings.
"""
patients = db.query(Patient).filter(Patient.patient_id != "UNKNOWN", Patient.mrn != "UNKNOWN").all()
results = []
now = datetime.now(timezone.utc)
live_cutoff = now - timedelta(seconds=25)
nibp_cutoff = now - timedelta(minutes=15)
for p in patients:
merged_vitals = None
# Primary source: in-memory live cache if running locally with active machine stream
try:
from contec_server import device_cache, device_field_timestamps
if p.device_id and p.device_id in device_cache:
cached = device_cache[p.device_id]
field_ts = device_field_timestamps.get(p.device_id, {})
cached_time = cached.timestamp
if cached_time:
if cached_time.tzinfo is None:
cached_time = cached_time.replace(tzinfo=timezone.utc)
if (now - cached_time).total_seconds() <= 25:
merged_vitals = {
"heart_rate": cached.heart_rate if (field_ts.get("heart_rate") and (now - field_ts["heart_rate"]).total_seconds() <= 25) else None,
"spo2": cached.spo2 if (field_ts.get("spo2") and (now - field_ts["spo2"]).total_seconds() <= 25) else None,
"respiratory_rate": cached.respiratory_rate if (field_ts.get("respiratory_rate") and (now - field_ts["respiratory_rate"]).total_seconds() <= 25) else None,
"temperature": cached.temperature if (field_ts.get("temperature") and (now - field_ts["temperature"]).total_seconds() <= 25) else None,
"temperature2": getattr(cached, 'temperature2', None) if (field_ts.get("temperature2") and (now - field_ts["temperature2"]).total_seconds() <= 25) else None,
"temperature_diff": getattr(cached, 'temperature_diff', None) if (field_ts.get("temperature_diff") and (now - field_ts["temperature_diff"]).total_seconds() <= 25) else None,
"systolic_bp": cached.systolic_bp if (field_ts.get("systolic_bp") and (now - field_ts["systolic_bp"]).total_seconds() <= 900) else None,
"diastolic_bp": cached.diastolic_bp if (field_ts.get("diastolic_bp") and (now - field_ts["diastolic_bp"]).total_seconds() <= 900) else None,
"map_bp": cached.map_bp if (field_ts.get("map_bp") and (now - field_ts["map_bp"]).total_seconds() <= 900) else None,
"timestamp": cached_time.isoformat()
}
except Exception as e:
logger.debug(f"Could not read from device_cache: {e}")
# Fallback or cloud/Render mode: query DB
if merged_vitals is None:
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
(VitalReading.device_id == p.device_id) if p.device_id else False
)
# 1. Try to find readings within the live cutoff window (last 25 seconds)
recent_readings = (
db.query(VitalReading)
.filter(patient_filters, VitalReading.timestamp >= live_cutoff)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
# 2. Option B: If no live reading in last 25s, retain latest recorded readings from DB
if not recent_readings:
recent_readings = (
db.query(VitalReading)
.filter(patient_filters)
.order_by(VitalReading.timestamp.desc())
.limit(20)
.all()
)
if recent_readings:
ts_iso = recent_readings[0].timestamp.isoformat() if recent_readings[0].timestamp else None
if ts_iso and not ts_iso.endswith('Z') and '+' not in ts_iso:
ts_iso += 'Z'
merged_vitals = {
"heart_rate": None,
"spo2": None,
"systolic_bp": None,
"diastolic_bp": None,
"map_bp": None,
"respiratory_rate": None,
"temperature": None,
"temperature2": None,
"temperature_diff": None,
"timestamp": ts_iso,
}
for r in reversed(recent_readings):
if r.heart_rate is not None: merged_vitals["heart_rate"] = r.heart_rate
if r.spo2 is not None: merged_vitals["spo2"] = r.spo2
if r.respiratory_rate is not None: merged_vitals["respiratory_rate"] = r.respiratory_rate
if r.temperature is not None: merged_vitals["temperature"] = r.temperature
if getattr(r, 'temperature2', None) is not None: merged_vitals["temperature2"] = r.temperature2
if getattr(r, 'temperature_diff', None) is not None: merged_vitals["temperature_diff"] = r.temperature_diff
if r.systolic_bp is not None: merged_vitals["systolic_bp"] = r.systolic_bp
if r.diastolic_bp is not None: merged_vitals["diastolic_bp"] = r.diastolic_bp
if r.map_bp is not None: merged_vitals["map_bp"] = r.map_bp
# Check NIBP separately if not already populated
if not merged_vitals or merged_vitals.get("systolic_bp") is None:
r_bp = (
db.query(VitalReading.systolic_bp, VitalReading.diastolic_bp, VitalReading.map_bp, VitalReading.timestamp)
.filter(patient_filters, VitalReading.systolic_bp != None)
.order_by(VitalReading.timestamp.desc())
.first()
)
if r_bp:
if merged_vitals is None:
ts_bp = r_bp[3].isoformat() if r_bp[3] else None
if ts_bp and not ts_bp.endswith('Z') and '+' not in ts_bp:
ts_bp += 'Z'
merged_vitals = {
"heart_rate": None, "spo2": None, "respiratory_rate": None,
"temperature": None, "temperature2": None, "temperature_diff": None,
"timestamp": ts_bp
}
merged_vitals["systolic_bp"] = r_bp[0]
merged_vitals["diastolic_bp"] = r_bp[1]
merged_vitals["map_bp"] = r_bp[2]
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": merged_vitals
})
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
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud({"mrn": "UNKNOWN", "name": "Unassigned Patient"}, device_id))
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,
patient_type=req.patient_type,
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
if req.patient_type is not None:
patient.patient_type = req.patient_type
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,
"patient_type": patient.patient_type,
}
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,
"patient_type": patient.patient_type,
})
except Exception as e:
logger.debug(f"Broadcast of patient_assigned failed: {e}")
from api_client import sync_patient_assignment_to_cloud
asyncio.create_task(sync_patient_assignment_to_cloud(p_data, device_id))
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.settimeout(0.5)
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 datetime import datetime, timezone
@router.post("/push-vitals")
async def push_vitals(vitals: NormalizedVitals, api_key: Optional[str] = None, db: Session = Depends(get_db)):
"""
Receive vitals pushed from a local instance and store + broadcast them to connected WebSocket clients.
"""
from config import settings
if settings.api_token and api_key != settings.api_token:
raise HTTPException(status_code=401, detail="Invalid or missing api_key")
try:
now = datetime.now(timezone.utc)
# 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",
last_seen=now
)
db.add(device)
else:
if vitals.ip_address:
device.ip_address = vitals.ip_address
device.status = "active"
device.last_seen = now
# 2. Patient tracking
mrn = vitals.patient_id
patient = None
if mrn and mrn != "UNKNOWN":
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=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:
if vitals.patient_name and vitals.patient_name != "Unknown":
patient.name = vitals.patient_name
if vitals.patient_gender:
patient.gender = vitals.patient_gender
if vitals.patient_dob:
patient.dob = vitals.patient_dob
if vitals.bed_id:
patient.bed_id = vitals.bed_id
patient.device_id = vitals.device_id
elif vitals.device_id:
# Check if there's already a patient assigned to this device
patient = db.query(Patient).filter(Patient.device_id == vitals.device_id, Patient.patient_id != "UNKNOWN").first()
if patient:
mrn = patient.mrn or patient.patient_id
# 3. Create VitalReading
reading = VitalReading(
device_id=vitals.device_id,
patient_id=mrn or "UNKNOWN",
timestamp=vitals.timestamp or now,
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
)
db.add(reading)
db.commit()
db.refresh(reading)
# 4. Broadcast to WebSocket clients
vitals_dict = vitals.model_dump(mode="json")
vitals_dict["mrn"] = mrn or (patient.mrn if patient else None)
vitals_dict["patient_id"] = vitals_dict["mrn"] or vitals.patient_id
vitals_dict["name"] = vitals.patient_name or (patient.name if patient else None)
vitals_dict["patient_name"] = vitals_dict["name"]
vitals_dict["gender"] = vitals.patient_gender or (patient.gender if patient else None)
vitals_dict["patient_gender"] = vitals_dict["gender"]
vitals_dict["dob"] = vitals.patient_dob or (patient.dob if patient else None)
vitals_dict["patient_dob"] = vitals_dict["dob"]
vitals_dict["bed_id"] = vitals.bed_id or (patient.bed_id if patient else None)
vitals_dict["patient_type"] = getattr(vitals, "patient_type", None)
await ws_manager.broadcast_vitals(vitals_dict)
return {
"status": "ok",
"reading_id": reading.id,
"device_id": vitals.device_id,
"patient_id": vitals_dict["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)}")
-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
-1
View File
@@ -1 +0,0 @@
python-3.12.0
-48
View File
@@ -1,48 +0,0 @@
from pydantic import BaseModel
from typing import Optional, List
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
temperature2: Optional[float] = None
temperature_diff: Optional[float] = None
present_fields: Optional[list] = None
# Raw waveform sample arrays (8-bit unsigned, 0-255)
ecg_wave: Optional[List[int]] = None # ECG waveform samples from 989-byte packet
pleth_wave: Optional[List[int]] = None # Pleth/SpO2 waveform samples from 286/288-byte packet
# Patient demographics parsed from HL7 PID/PV1 segments
patient_name: Optional[str] = None # PID-5: "Given Family"
patient_gender: Optional[str] = None # PID-8: M / F / U
patient_dob: Optional[str] = None # PID-7: YYYYMMDD
bed_id: Optional[str] = None # PV1-3: bed / room ID
patient_type: Optional[str] = None # Patient category: Adult / Pediatric / Neonatal
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
-82
View File
@@ -1,82 +0,0 @@
#!/bin/bash
# Exit on error
set -e
SERVICE_NAME="contec-forwarder"
SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service"
# Auto-detect script directory
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if [ -f "${SCRIPT_DIR}/main.py" ]; then
WORKING_DIR="${SCRIPT_DIR}"
else
WORKING_DIR="${PWD}"
fi
echo "=== Detected Working Directory: ${WORKING_DIR} ==="
# Ensure virtual environment exists
if [ ! -f "${WORKING_DIR}/venv/bin/python" ]; then
echo "=== Virtual environment not found. Creating venv... ==="
python3 -m venv "${WORKING_DIR}/venv" || sudo apt-get update && sudo apt-get install -y python3-venv python3-pip && python3 -m venv "${WORKING_DIR}/venv"
fi
PYTHON_EXEC="${WORKING_DIR}/venv/bin/python"
SCRIPT_PATH="${WORKING_DIR}/main.py"
# Ensure dependencies are installed
if [ -f "${WORKING_DIR}/requirements.txt" ]; then
echo "=== Installing/verifying dependencies ==="
"${WORKING_DIR}/venv/bin/pip" install --upgrade pip
"${WORKING_DIR}/venv/bin/pip" install -r "${WORKING_DIR}/requirements.txt"
fi
# Open firewall ports for Contec monitor (Port 511) and Web API (Port 8006)
if command -v ufw >/dev/null 2>&1; then
echo "=== Configuring firewall (UFW) for Port 511 and 8006 ==="
sudo ufw allow 511/tcp comment "Contec Patient Monitor CMS TCP" || true
sudo ufw allow 8006/tcp comment "Contec Forwarder HTTP API" || true
fi
echo "=== Creating systemd service file: ${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
Environment=PYTHONPATH=${WORKING_DIR}
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 "=== Restarting ${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 "Port 511 is open and listening for the Contec physical monitor."
echo "To check live logs, run:"
echo " sudo journalctl -u ${SERVICE_NAME} -f"
echo "=========================================================="
File diff suppressed because it is too large Load Diff
-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
+15 -2
View File
@@ -127,6 +127,7 @@ async def lifespan(app: FastAPI):
print(f" Contec Ports: {settings.contec_ports}")
print(f" Monitor Model: {model}")
print(f" API Target: {settings.target_api_url}")
print(f" Cloud Sync: {getattr(settings, 'cloud_api_url', settings.target_api_url)}")
print(f" Database: {settings.database_url}")
print(f" ---------------------------------------------------------")
print(f" YOUR PC's IP ADDRESSES (configure on your monitor):")
@@ -196,22 +197,31 @@ app.add_middleware(
allow_headers=["*"],
)
# Include dashboard routes
# Include dashboard routes under multiple prefixes so reverse proxies (Nginx)
# work whether they strip /v1/contec-monitor or forward it as-is
app.include_router(dashboard.router, prefix="/api")
app.include_router(dashboard.router)
app.include_router(dashboard.router, prefix="/v1/contec-monitor/api")
app.include_router(dashboard.router, prefix="/v1/contec-monitor")
# Also serve dashboard at root and /dashboard directly for convenience
# Also serve dashboard directly for convenience across various path styles
from fastapi.responses import HTMLResponse
@app.get("/", response_class=HTMLResponse)
@app.get("/dashboard", response_class=HTMLResponse)
@app.get("/v1/contec-monitor", response_class=HTMLResponse)
@app.get("/v1/contec-monitor/dashboard", response_class=HTMLResponse)
@app.get("/v1/contec-monitor/api/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")
app.mount("/v1/contec-monitor/static", StaticFiles(directory="static"), name="static_subpath")
from fastapi.responses import Response
@app.get("/favicon.ico", include_in_schema=False)
@app.get("/v1/contec-monitor/favicon.ico", include_in_schema=False)
async def favicon():
return Response(status_code=204, media_type="image/x-icon")
@@ -220,6 +230,9 @@ async def favicon():
# === WebSocket endpoint for live vitals ===
@app.websocket("/ws/vitals")
@app.websocket("/api/ws/vitals")
@app.websocket("/v1/contec-monitor/ws/vitals")
@app.websocket("/v1/contec-monitor/api/ws/vitals")
async def websocket_vitals(websocket: WebSocket):
manager = get_ws_manager()
await manager.connect(websocket)
+2
View File
@@ -660,6 +660,8 @@ async def push_vitals(vitals: NormalizedVitals, api_key: Optional[str] = None, d
map_bp=vitals.map_bp,
respiratory_rate=vitals.respiratory_rate,
temperature=vitals.temperature,
temperature2=getattr(vitals, 'temperature2', None),
temperature_diff=getattr(vitals, 'temperature_diff', None),
transmitted=True
)
db.add(reading)
+4 -3
View File
@@ -32,10 +32,11 @@ if [ -f "${WORKING_DIR}/requirements.txt" ]; then
"${WORKING_DIR}/venv/bin/pip" install -r "${WORKING_DIR}/requirements.txt"
fi
# Open firewall ports for Contec monitor (Port 511) and Web API (Port 8006)
# Open firewall ports for Contec monitor (Port 511, 61501) and Web API (Port 8006)
if command -v ufw >/dev/null 2>&1; then
echo "=== Configuring firewall (UFW) for Port 511 and 8006 ==="
sudo ufw allow 511/tcp comment "Contec Patient Monitor CMS TCP" || true
echo "=== Configuring firewall (UFW) for Port 511, 61501, and 8006 ==="
sudo ufw allow 511/tcp comment "Contec Patient Monitor CMS TCP 511" || true
sudo ufw allow 61501/tcp comment "Contec Patient Monitor CMS TCP 61501" || true
sudo ufw allow 8006/tcp comment "Contec Forwarder HTTP API" || true
fi
+71 -25
View File
@@ -3185,9 +3185,11 @@
p = patients.find(pt => pt.device_id === activeDeviceId);
}
if (!p && patients.length > 0) {
// Find patient corresponding to a live transmitting device
// Find patient corresponding to a live transmitting device, active device, patient with latest vitals, or first patient
p = patients.find(pt => pt.device_id && connectedDevices[pt.device_id]?._liveData) ||
patients.find(pt => pt.device_id && pt.device_id === activeDeviceId);
patients.find(pt => pt.device_id && pt.device_id === activeDeviceId) ||
patients.find(pt => pt.latest_vitals != null) ||
patients[0];
}
if (p) {
@@ -3195,6 +3197,9 @@
machVitals.patient_id = p.mrn || p.patient_id || null;
machVitals.bed_id = p.bed_id || null;
machVitals.device_id = p.device_id || activeDeviceId || '';
if (!activeDeviceId && p.device_id) {
activeDeviceId = p.device_id;
}
const elName = document.getElementById('machPatientName');
if (elName) elName.textContent = machVitals.patient_name ? machVitals.patient_name.toUpperCase() : '--';
@@ -3432,9 +3437,17 @@
}
// === Configuration ===
const API_BASE = window.location.origin + '/api';
function getAppPrefix() {
const p = window.location.pathname;
if (p.includes('/v1/contec-monitor')) return '/v1/contec-monitor';
if (p.includes('/api/')) return p.substring(0, p.indexOf('/api/'));
if (p.includes('/dashboard')) return p.substring(0, p.indexOf('/dashboard'));
return '';
}
const APP_PREFIX = getAppPrefix();
const API_BASE = window.location.origin + APP_PREFIX + '/api';
const POLL_INTERVAL = 2000; // 2 seconds
const WS_URL = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}/ws/vitals`;
const WS_URL = `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${window.location.host}${APP_PREFIX}/ws/vitals`;
// === Vital ranges for status calculation ===
const VITAL_RANGES = {
@@ -3455,6 +3468,7 @@
};
let waveformCanvases = {};
let useWebSocket = false;
let lastWebSocketMessageTime = 0;
let ws = null;
let audioAlertEnabled = false;
let audioContext = null;
@@ -3639,9 +3653,10 @@
: 'All sent ✓';
}
// If WebSocket is connected, it delivers sub-second live machine telemetry directly.
// Do NOT overwrite live WebSocket data with older throttled DB snapshots.
if (useWebSocket) {
// If WebSocket is connected and actively receiving messages, it delivers live telemetry.
// Only skip polling if WebSocket messages were received within the last 5 seconds.
const hasRecentWs = lastWebSocketMessageTime && (Date.now() - lastWebSocketMessageTime < 5000);
if (useWebSocket && hasRecentWs) {
return;
}
@@ -3758,7 +3773,28 @@
if (mrn && mrn !== 'UNKNOWN') {
patientRegistry[mrn] = p;
}
if (p.device_id && connectedDevices[p.device_id]) {
if (p.device_id) {
if (!connectedDevices[p.device_id]) {
connectedDevices[p.device_id] = {
device_id: p.device_id,
ip_address: p.device_ip || 'remote-push',
bedLabel: p.bed_id || getBedLabel(p.device_id, p.device_ip),
lastSeen: new Date(),
status: p.device_status || 'active',
_fromDb: true,
_liveData: false
};
}
if (p.latest_vitals) {
connectedDevices[p.device_id]._liveData = true;
if (p.latest_vitals.heart_rate != null) connectedDevices[p.device_id].heart_rate = p.latest_vitals.heart_rate;
if (p.latest_vitals.spo2 != null) connectedDevices[p.device_id].spo2 = p.latest_vitals.spo2;
if (p.latest_vitals.respiratory_rate != null) connectedDevices[p.device_id].respiratory_rate = p.latest_vitals.respiratory_rate;
if (p.latest_vitals.temperature != null) connectedDevices[p.device_id].temperature = p.latest_vitals.temperature;
if (p.latest_vitals.systolic_bp != null) connectedDevices[p.device_id].systolic_bp = p.latest_vitals.systolic_bp;
if (p.latest_vitals.diastolic_bp != null) connectedDevices[p.device_id].diastolic_bp = p.latest_vitals.diastolic_bp;
if (p.latest_vitals.map_bp != null) connectedDevices[p.device_id].map_bp = p.latest_vitals.map_bp;
}
Object.assign(connectedDevices[p.device_id], {
patient_id: mrn,
mrn: mrn,
@@ -3797,6 +3833,8 @@
renderSingleDevice(connectedDevices[activeDeviceId]);
}
} else {
const devIds = Object.keys(connectedDevices);
if (!activeDeviceId && devIds.length > 0) activeDeviceId = devIds[0];
if (viewMode === 'grid') renderMultiBedGrid();
else if (activeDeviceId && connectedDevices[activeDeviceId]) renderSingleDevice(connectedDevices[activeDeviceId]);
}
@@ -3861,8 +3899,11 @@
const tabsContainer = document.getElementById('deviceTabs');
if (!tabsContainer) return;
// Only show devices that have received a live WebSocket packet
let devIds = Object.keys(connectedDevices).filter(id => connectedDevices[id]._liveData === true);
// Show devices that are live via WebSocket, or have recent data/active status
let devIds = Object.keys(connectedDevices).filter(id => {
const dev = connectedDevices[id];
return dev._liveData === true || dev.status === 'active' || (dev.lastSeen && (Date.now() - dev.lastSeen.getTime()) < 900000);
});
if (hasTargetFilter) {
devIds = devIds.filter(id => matchesTargetPatient(connectedDevices[id], id));
}
@@ -3892,8 +3933,11 @@
function renderMultiBedGrid() {
const multiGrid = document.getElementById('multiBedGrid');
// Only show devices confirmed live via WebSocket — exclude DB-only ghost devices
let devIds = Object.keys(connectedDevices).filter(id => connectedDevices[id]._liveData === true);
// Show devices confirmed live via WebSocket or active with vitals
let devIds = Object.keys(connectedDevices).filter(id => {
const dev = connectedDevices[id];
return dev._liveData === true || dev.status === 'active' || (dev.lastSeen && (Date.now() - dev.lastSeen.getTime()) < 900000);
});
if (hasTargetFilter) {
devIds = devIds.filter(id => matchesTargetPatient(connectedDevices[id], id));
}
@@ -3978,28 +4022,29 @@
const pGender = (dev.patient_gender && dev.patient_gender !== 'Unknown') ? dev.patient_gender : null;
const pDob = (dev.patient_dob && dev.patient_dob !== 'Unknown') ? dev.patient_dob : null;
// A device is "live" only if it has received a live WebSocket packet within 8 seconds
// Devices loaded only from DB (never seen via WebSocket) are shown as DISCONNECTED
const isLive = dev._liveData === true && dev.lastSeen && (Date.now() - dev.lastSeen.getTime()) < 8000;
// Has active vitals (probe connected and sending data)
const hasVitals = isLive && (dev.heart_rate != null || dev.spo2 != null || dev.temperature != null);
const hr = hasVitals && dev.heart_rate != null ? Math.round(dev.heart_rate) : '---';
const spo2 = hasVitals && dev.spo2 != null ? Math.round(dev.spo2) : '---';
const sys = hasVitals && dev.systolic_bp != null ? Math.round(dev.systolic_bp) : '--';
const dia = hasVitals && dev.diastolic_bp != null ? Math.round(dev.diastolic_bp) : '--';
// A device is live if actively transmitting or active in DB within 30 seconds
const isLive = (dev._liveData === true || dev.status === 'active') && dev.lastSeen && (Date.now() - dev.lastSeen.getTime()) < 30000;
// Has recorded vitals
const hasVitals = (dev.heart_rate != null || dev.spo2 != null || dev.temperature != null);
const hr = dev.heart_rate != null ? Math.round(dev.heart_rate) : '---';
const spo2 = dev.spo2 != null ? Math.round(dev.spo2) : '---';
const sys = dev.systolic_bp != null ? Math.round(dev.systolic_bp) : '--';
const dia = dev.diastolic_bp != null ? Math.round(dev.diastolic_bp) : '--';
const bp = (sys !== '--' || dia !== '--') ? `${sys}/${dia}` : '---';
const temp = hasVitals && dev.temperature != null ? dev.temperature.toFixed(1) + '°C' : '---';
const rr = hasVitals && dev.respiratory_rate != null ? Math.round(dev.respiratory_rate) : '---';
const temp = dev.temperature != null ? dev.temperature.toFixed(1) + '°C' : '---';
const rr = dev.respiratory_rate != null ? Math.round(dev.respiratory_rate) : '---';
const timeStr = dev.lastSeen ? dev.lastSeen.toLocaleTimeString('en-US', { hour12: false }) : '--';
const genderIcon = pGender ? (pGender.toLowerCase().startsWith('m') ? '♂ Male' : pGender.toLowerCase().startsWith('f') ? '♀ Female' : pGender) : '';
// Badge: LIVE (data flowing) | CONNECTED (monitor on, probe off) | DISCONNECTED
// Badge: LIVE (data flowing) | CONNECTED (monitor on / recent data) | DISCONNECTED
const statusBadge = isLive
? (hasVitals
? `<span class="status-badge normal">Live ↗</span>`
: `<span class="status-badge warning" style="background:rgba(234,179,8,.15);color:#fbbf24;border:1px solid rgba(234,179,8,.3);">Connected</span>`)
: `<span class="status-badge offline" style="background:rgba(239,68,68,.15);color:#f87171;border:1px solid rgba(239,68,68,.3);">Disconnected</span>`;
: (hasVitals
? `<span class="status-badge warning" style="background:rgba(234,179,8,.15);color:#fbbf24;border:1px solid rgba(234,179,8,.3);">Recent</span>`
: `<span class="status-badge offline" style="background:rgba(239,68,68,.15);color:#f87171;border:1px solid rgba(239,68,68,.3);">Disconnected</span>`);
html += `
<div class="bed-card" onclick="selectDevice('${id}')" title="Click to view detailed waveforms for ${label}">
@@ -4114,6 +4159,7 @@
}
if (!data.device_id) return;
lastWebSocketMessageTime = Date.now();
// Hide setup guide once data starts flowing
if (!hasReceivedData) {