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 def get_cloud_base_url() -> str: url = getattr(settings, "cloud_api_url", None) or os.getenv("CLOUD_API_URL") or "https://contectfiveparamonitor.onrender.com" 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 Cloud server 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 Cloud Server async def _forward_cloud() -> bool: cloud_base = get_cloud_base_url() cloud_url = f"{cloud_base}/api/push-vitals" is_on_cloud = os.getenv("RENDER") is not None or os.getenv("PORT") == "10000" or "render.local" in os.getenv("HOSTNAME", "") if is_on_cloud and ("onrender.com" in cloud_base): return True try: url_with_key = cloud_url if settings.api_token: url_with_key = f"{cloud_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 Cloud 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 Cloud API: {type(e).__name__} ({err_msg})") return False results = await asyncio.gather(_forward_primary(), _forward_cloud(), 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 Cloud Server. Ensures local and remote dashboards always show the exact same patient. """ 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 is_on_cloud and ("onrender.com" in cloud_base): return True client = get_http_client() cloud_assign_url = f"{cloud_base}/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 = cloud_assign_url if settings.api_token: url_with_key = f"{cloud_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 Cloud for {device_id}") return True except Exception as e: logger.warning(f"Could not sync patient assignment to Cloud: {type(e).__name__} ({e})") return False