diff --git a/api_client.py b/api_client.py index a07825f..16c29fd 100644 --- a/api_client.py +++ b/api_client.py @@ -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 primary_success or render_success + 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: @@ -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 diff --git a/background_tasks.py b/background_tasks.py index f27b5c4..837c648 100644 --- a/background_tasks.py +++ b/background_tasks.py @@ -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) + diff --git a/contec_server.py b/contec_server.py index 0a43f8a..31e170d 100644 --- a/contec_server.py +++ b/contec_server.py @@ -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, diff --git a/fiveparaminte-main/api_client.py b/fiveparaminte-main/api_client.py index a07825f..16c29fd 100644 --- a/fiveparaminte-main/api_client.py +++ b/fiveparaminte-main/api_client.py @@ -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 primary_success or render_success + 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: @@ -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 diff --git a/fiveparaminte-main/background_tasks.py b/fiveparaminte-main/background_tasks.py index f27b5c4..837c648 100644 --- a/fiveparaminte-main/background_tasks.py +++ b/fiveparaminte-main/background_tasks.py @@ -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) + diff --git a/fiveparaminte-main/contec_server.py b/fiveparaminte-main/contec_server.py index 6ced765..31e170d 100644 --- a/fiveparaminte-main/contec_server.py +++ b/fiveparaminte-main/contec_server.py @@ -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, diff --git a/fiveparaminte-main/main.py b/fiveparaminte-main/main.py index 8e3d9c9..6d8c61c 100644 --- a/fiveparaminte-main/main.py +++ b/fiveparaminte-main/main.py @@ -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: diff --git a/main.py b/main.py index 8e3d9c9..6d8c61c 100644 --- a/main.py +++ b/main.py @@ -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: