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

fix: reliable cloud sync and connection pooling for Render live dashboard

Browse Source
This commit is contained in:
rkcsh
2026-09-17 16:41:52 +05:30
parent 68bda2519f
commit 37445d2ae9
8 changed files with 286 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_client.py
+24 -18
View File
@@ -1,5 +1,7 @@
import httpx
import logging
import asyncio
import os
from typing import Optional
from schemas import NormalizedVitals
from config import settings
@@ -12,18 +14,17 @@ def get_http_client() -> httpx.AsyncClient:
global _http_client
if _http_client is None or _http_client.is_closed:
_http_client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=3.0, read=5.0, write=3.0, pool=3.0),
limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
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.
and to the Render cloud database concurrently.
Returns True if at least one forward succeeds.
"""
import os
headers = {"Content-Type": "application/json"}
# Strip heavy wave sample arrays before REST transmission (cloud APIs only ingest vital signs)
@@ -31,12 +32,16 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
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)
primary_url = settings.target_api_url
primary_success = False
if primary_url:
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:
@@ -48,19 +53,18 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}")
primary_success = True
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
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
render_success = False
# Detect if we are already running on Render to prevent an infinite loop
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not is_on_render:
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:
@@ -73,12 +77,14 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}")
render_success = True
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
return primary_success or render_success
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:
@@ -86,7 +92,6 @@ async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -
Synchronizes patient demographics assignment to Render Cloud Server.
Ensures local and remote dashboards always show the exact same patient.
"""
import os
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
@@ -100,6 +105,7 @@ async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -
"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
background_tasks.py
+83
View File
@@ -79,3 +79,86 @@ async def retry_failed_transmissions():
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. Backfill and forward latest complete non-null vitals to Render
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
VitalReading.device_id == p.device_id
)
recent_r = db.query(VitalReading).filter(patient_filters).order_by(VitalReading.timestamp.desc()).first()
if recent_r:
spo2_r = db.query(VitalReading.spo2).filter(patient_filters, VitalReading.spo2 != None).order_by(VitalReading.timestamp.desc()).first()
hr_r = db.query(VitalReading.heart_rate).filter(patient_filters, VitalReading.heart_rate != 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()
rr_r = db.query(VitalReading.respiratory_rate).filter(patient_filters, VitalReading.respiratory_rate != None).order_by(VitalReading.timestamp.desc()).first()
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=recent_r.timestamp or 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,
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)
contec_server.py
+30 -1
View File
@@ -657,10 +657,39 @@ async def _persist_and_forward_bg(vitals, should_persist):
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
# Backfill non-null vitals from device cache so REST forwarders always get complete vitals
cached = device_cache.get(vitals.device_id)
if cached:
if vitals.heart_rate is None and cached.heart_rate is not None:
vitals.heart_rate = cached.heart_rate
if vitals.spo2 is None and cached.spo2 is not None:
vitals.spo2 = cached.spo2
if vitals.systolic_bp is None and cached.systolic_bp is not None:
vitals.systolic_bp = cached.systolic_bp
vitals.diastolic_bp = cached.diastolic_bp
vitals.map_bp = cached.map_bp
if vitals.respiratory_rate is None and cached.respiratory_rate is not None:
vitals.respiratory_rate = cached.respiratory_rate
if vitals.temperature is None and cached.temperature is not None:
vitals.temperature = cached.temperature
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
patient_id=vitals.patient_id or (patient.mrn if patient else "UNKNOWN"),
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
fiveparaminte-main/api_client.py
+24 -18
View File
@@ -1,5 +1,7 @@
import httpx
import logging
import asyncio
import os
from typing import Optional
from schemas import NormalizedVitals
from config import settings
@@ -12,18 +14,17 @@ def get_http_client() -> httpx.AsyncClient:
global _http_client
if _http_client is None or _http_client.is_closed:
_http_client = httpx.AsyncClient(
timeout=httpx.Timeout(connect=3.0, read=5.0, write=3.0, pool=3.0),
limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
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.
and to the Render cloud database concurrently.
Returns True if at least one forward succeeds.
"""
import os
headers = {"Content-Type": "application/json"}
# Strip heavy wave sample arrays before REST transmission (cloud APIs only ingest vital signs)
@@ -31,12 +32,16 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
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)
primary_url = settings.target_api_url
primary_success = False
if primary_url:
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:
@@ -48,19 +53,18 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to primary API for device {vitals.device_id}")
primary_success = True
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
render_url = "https://contectfiveparamonitor.onrender.com/api/push-vitals"
render_success = False
# Detect if we are already running on Render to prevent an infinite loop
is_on_render = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "")
if not is_on_render:
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:
@@ -73,12 +77,14 @@ async def forward_vitals_to_api(vitals: NormalizedVitals) -> bool:
)
response.raise_for_status()
logger.info(f"Successfully forwarded vitals to Render Cloud for device {vitals.device_id}")
render_success = True
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
return primary_success or render_success
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:
@@ -86,7 +92,6 @@ async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -
Synchronizes patient demographics assignment to Render Cloud Server.
Ensures local and remote dashboards always show the exact same patient.
"""
import os
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
@@ -100,6 +105,7 @@ async def sync_patient_assignment_to_cloud(patient_data: dict, device_id: str) -
"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
fiveparaminte-main/background_tasks.py
+83
View File
@@ -79,3 +79,86 @@ async def retry_failed_transmissions():
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. Backfill and forward latest complete non-null vitals to Render
patient_filters = or_(
VitalReading.patient_id == p.mrn,
VitalReading.patient_id == p.patient_id,
VitalReading.device_id == p.device_id
)
recent_r = db.query(VitalReading).filter(patient_filters).order_by(VitalReading.timestamp.desc()).first()
if recent_r:
spo2_r = db.query(VitalReading.spo2).filter(patient_filters, VitalReading.spo2 != None).order_by(VitalReading.timestamp.desc()).first()
hr_r = db.query(VitalReading.heart_rate).filter(patient_filters, VitalReading.heart_rate != 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()
rr_r = db.query(VitalReading.respiratory_rate).filter(patient_filters, VitalReading.respiratory_rate != None).order_by(VitalReading.timestamp.desc()).first()
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=recent_r.timestamp or 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,
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)
fiveparaminte-main/contec_server.py
+32 -5
View File
@@ -522,14 +522,12 @@ async def process_vitals(vitals):
if value is not None:
active_fields.append(field)
# Update cache & record timestamps
# Update cache & record timestamps for present values
for field in active_fields:
val = getattr(vitals, field, None)
setattr(cached, field, val)
if val is not None:
setattr(cached, field, val)
device_field_timestamps[device_id][field] = now
else:
device_field_timestamps[device_id].pop(field, None)
# Clean up stale fields:
# Live vitals (HR, SpO2, Temp, RR) have a 60-second window because monitor cycles take ~24s
@@ -659,10 +657,39 @@ async def _persist_and_forward_bg(vitals, should_persist):
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
# Backfill non-null vitals from device cache so REST forwarders always get complete vitals
cached = device_cache.get(vitals.device_id)
if cached:
if vitals.heart_rate is None and cached.heart_rate is not None:
vitals.heart_rate = cached.heart_rate
if vitals.spo2 is None and cached.spo2 is not None:
vitals.spo2 = cached.spo2
if vitals.systolic_bp is None and cached.systolic_bp is not None:
vitals.systolic_bp = cached.systolic_bp
vitals.diastolic_bp = cached.diastolic_bp
vitals.map_bp = cached.map_bp
if vitals.respiratory_rate is None and cached.respiratory_rate is not None:
vitals.respiratory_rate = cached.respiratory_rate
if vitals.temperature is None and cached.temperature is not None:
vitals.temperature = cached.temperature
reading = VitalReading(
device_id=vitals.device_id,
patient_id=vitals.patient_id,
patient_id=vitals.patient_id or (patient.mrn if patient else "UNKNOWN"),
timestamp=vitals.timestamp,
heart_rate=vitals.heart_rate,
spo2=vitals.spo2,
fiveparaminte-main/main.py
+4 -2
View File
@@ -11,7 +11,7 @@ import uvicorn
from config import settings
from database import engine, Base
from contec_server import handle_contec_client
from background_tasks import retry_failed_transmissions
from background_tasks import retry_failed_transmissions, sync_cloud_state
from routers import dashboard
from routers.dashboard import get_ws_manager
@@ -138,8 +138,9 @@ async def lifespan(app: FastAPI):
print(f" Set Server IP = one of the IPs above, Port = {settings.contec_ports[0] if settings.contec_ports else 511}")
print("=" * 65 + "\n")
# Startup: Start the background retry task
# 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
@@ -170,6 +171,7 @@ async def lifespan(app: FastAPI):
except Exception:
pass
retry_task.cancel()
cloud_sync_task.cancel()
for server in servers:
server.close()
for server in servers:
main.py
+4 -2
View File
@@ -11,7 +11,7 @@ import uvicorn
from config import settings
from database import engine, Base
from contec_server import handle_contec_client
from background_tasks import retry_failed_transmissions
from background_tasks import retry_failed_transmissions, sync_cloud_state
from routers import dashboard
from routers.dashboard import get_ws_manager
@@ -138,8 +138,9 @@ async def lifespan(app: FastAPI):
print(f" Set Server IP = one of the IPs above, Port = {settings.contec_ports[0] if settings.contec_ports else 511}")
print("=" * 65 + "\n")
# Startup: Start the background retry task
# 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
@@ -170,6 +171,7 @@ async def lifespan(app: FastAPI):
except Exception:
pass
retry_task.cancel()
cloud_sync_task.cancel()
for server in servers:
server.close()
for server in servers: