"""Application orchestrator: wires config → servers → uploader → heartbeats. The web admin UI is always started first, so the operator can configure the cloud connection and devices from a browser even when the JSON file is empty. Saving from the UI calls `reload()`, which swaps the cloud client and restarts the device listeners without dropping the admin UI or the outbox. """ from __future__ import annotations import logging import threading import time from typing import Any, Dict, List from .buffer import Outbox from .cloud import CloudClient, CloudError from .config import DEFAULT_BARCODE_FIELDS, DEFAULT_VALUE_TYPES, Config, save_config from .server import DeviceServer log = logging.getLogger("app") class Application: """Owns the lifecycle of all background workers + the admin UI.""" def __init__(self, config: Config, config_path: str) -> None: self.cfg = config self.config_path = config_path self.outbox = Outbox(config.buffer_db) self.cloud = CloudClient(config) self._servers_lock = threading.Lock() self.servers: List[DeviceServer] = [] self._stop = threading.Event() self.admin = None # ----- lifecycle ------------------------------------------------------ def run(self) -> None: # Admin UI first — always reachable so config can be fixed live. from .admin import AdminServer self.admin = AdminServer(self) try: self.admin.start() log.info("admin UI on http://%s:%s", self.cfg.admin_host, self.cfg.admin_port) except OSError as exc: log.error("admin UI could not start on %s:%s — %s", self.cfg.admin_host, self.cfg.admin_port, exc) self._start_devices() try: self.cloud.ensure_token() # Pull the cloud-managed device list on every start so the local # listeners always mirror LIS → Machines. if self.cfg.can_upload: summary = self.sync_from_cloud() log.info("startup sync from cloud: %s", summary) except CloudError as exc: log.warning("cloud auth not ready (configure it in the admin UI): %s", exc) threading.Thread(target=self._uploader_loop, name="uploader", daemon=True).start() threading.Thread(target=self._heartbeat_loop, name="heartbeat", daemon=True).start() log.info("middleware running — %d active device(s), %d pending in buffer", len(self.servers), self.outbox.count_pending()) try: while not self._stop.is_set(): time.sleep(1) except KeyboardInterrupt: log.info("shutdown requested") finally: self.stop() def stop(self) -> None: self._stop.set() self._stop_devices() if self.admin: self.admin.stop() # ----- device lifecycle ---------------------------------------------- def _start_devices(self) -> None: with self._servers_lock: self.servers = [] for d in self.cfg.devices: if not d.runnable: continue if d.is_serial: from .serial_server import SerialDevice srv: Any = SerialDevice(d, self.outbox, order_provider=self._orders_for) else: srv = DeviceServer(d, self.outbox) self.servers.append(srv) srv.start() def _orders_for(self, machine_id: int, barcode: str) -> list: """Bidirectional ASTM: which analyzer test codes are ordered for a tube.""" try: self.cloud.ensure_token() return self.cloud.orders_for(machine_id, barcode) except CloudError as exc: log.warning("order lookup for %s failed: %s", barcode, exc) return [] def _stop_devices(self) -> None: with self._servers_lock: for srv in self.servers: srv.stop() self.servers = [] def reload(self, new_raw: Dict[str, Any]) -> Config: """Persist a new raw config and hot-restart cloud + device listeners.""" self.cfg = save_config(self.config_path, new_raw) self._stop_devices() self.cloud = CloudClient(self.cfg) self._start_devices() log.info("config reloaded — %d active device(s)", len(self.servers)) return self.cfg # ----- cloud device sync --------------------------------------------- def sync_from_cloud(self) -> Dict[str, Any]: """Pull LIS → Machines and rebuild the device list from it. The cloud is the source of truth: each active network machine (interface tcp/hl7) whose `connection_settings.port` is set becomes a listener. Local per-device tweaks are preserved by machine_id. A machine is only auto-enabled once it has an explicit driver in the cloud. """ machines = self.cloud.list_machines() existing = {d.machine_id: d for d in self.cfg.devices} new_devices = [] added = updated = skipped = 0 for m in machines: it = m.get("interface_type") it = it.get("value") if isinstance(it, dict) else it cs = m.get("connection_settings") or {} port = cs.get("port") com = cs.get("com_port") # Network (tcp/hl7 with a port) OR serial (com_port) machines sync. is_net = it in ("tcp", "hl7") and port is_ser = bool(com) if not (is_net or is_ser): skipped += 1 continue try: mid = int(m.get("id")) port = int(port) if port else 0 except (TypeError, ValueError): skipped += 1 continue prev = existing.get(mid) driver = cs.get("driver") or (prev.driver if prev else ("maglumi_astm" if is_ser else "dymind_hl7")) new_devices.append({ "name": m.get("name") or (prev.name if prev else "device"), "machine_id": mid, "driver": driver, "listen_host": cs.get("host") or (prev.listen_host if prev else "0.0.0.0"), # Cloud is the source of truth for the listen port — a changed # port in LIS → Machines must apply on next sync (don't pin the # old local value). "listen_port": port, "barcode_fields": cs.get("barcode_fields") or (prev.barcode_fields if prev else list(DEFAULT_BARCODE_FIELDS)), "value_types": cs.get("value_types") or (prev.value_types if prev else list(DEFAULT_VALUE_TYPES)), "send_ack": cs.get("send_ack", prev.send_ack if prev else True), "forward_host": cs.get("forward_host") or (prev.forward_host if prev else ""), "forward_port": cs.get("forward_port") or (prev.forward_port if prev else 0), # Serial (RS-232) fields for ASTM analyzers. "com_port": com or (prev.com_port if prev else ""), "baud_rate": cs.get("baud_rate") or (prev.baud_rate if prev else 9600), "data_bits": cs.get("data_bits") or (prev.data_bits if prev else 8), "stop_bits": cs.get("stop_bits") or (prev.stop_bits if prev else 1), "parity": cs.get("parity") or (prev.parity if prev else "none"), # ENABLEMENT IS A LOCAL DECISION per workstation: the cloud holds # every machine of every branch, but each station only runs the # one or two physically wired to it. Preserve the local flag; # newly-discovered machines arrive DISABLED until the operator # turns them on here. "enabled": (prev.enabled if prev else False), }) updated += 1 if prev else 0 added += 0 if prev else 1 raw = self.cfg.to_dict() raw["devices"] = new_devices self.reload(raw) return {"added": added, "updated": updated, "skipped": skipped, "active": len(self.servers)} # ----- status (for the admin UI) ------------------------------------- def status(self) -> Dict[str, Any]: with self._servers_lock: running = {(s.cfg.name, s.cfg.listen_port): s.stats() for s in self.servers} devices: List[Dict[str, Any]] = [] for d in self.cfg.devices: stat = running.get((d.name, d.listen_port)) if stat is None: stat = { "name": d.name, "machine_id": d.machine_id, "driver": d.driver, "listen_host": d.listen_host, "listen_port": d.listen_port, "enabled": d.enabled, "listening": False, "bind_error": None, "connections_open": 0, "messages": 0, "results": 0, "errors": 0, "last_message_at": None, "last_barcode": None, "last_peer": None, "last_error": None, "note": "disabled" if not d.enabled else "incomplete config", } devices.append(stat) return { "workstation": self.cfg.workstation_name, "cloud": { "base_url": self.cfg.base_url, "has_token": self.cloud.has_token, "login_enabled": self.cfg.login_enabled, "can_upload": self.cfg.can_upload, }, "buffer": {"pending": self.outbox.count_pending()}, "devices": devices, "messages": self.cfg.validation_messages(), } # ----- workers -------------------------------------------------------- def _uploader_loop(self) -> None: """Drain the outbox to the cloud, FIFO, retrying on failure.""" while not self._stop.is_set(): if not self.cfg.can_upload: self._stop.wait(self.cfg.retry_interval) continue batch = self.outbox.pending(self.cfg.drain_batch) if not batch: self._stop.wait(self.cfg.retry_interval) continue stalled = False for row_id, payload in batch: if self._stop.is_set(): break try: self.cloud.ensure_token() if payload.get("_kind") == "comm_log": # Full raw message archive (audit) → its own endpoint. self.cloud.log_communication( payload.get("machine_id"), payload.get("raw_data", ""), payload.get("parsed_summary", ""), payload.get("protocol", "hl7")) self.outbox.mark_sent(row_id) log.info("archived full message for machine %s", payload.get("machine_id")) continue resp = self.cloud.post_result(payload) meta = (resp or {}).get("meta", {}) self.outbox.mark_sent(row_id) log.info("uploaded %s/%s matched=%s applied=%s", payload.get("barcode"), payload.get("test_code"), meta.get("matched"), meta.get("auto_applied")) except CloudError as exc: self.outbox.mark_failed(row_id, str(exc)) log.warning("upload failed (will retry): %s", exc) if exc.status is None or exc.status >= 500: stalled = True break if stalled: self._stop.wait(self.cfg.retry_interval) def _heartbeat_loop(self) -> None: """Periodically mark each active device online in the cloud.""" while not self._stop.is_set(): if self.cfg.can_upload: with self._servers_lock: servers = list(self.servers) for srv in servers: try: self.cloud.ensure_token() self.cloud.heartbeat(srv.cfg.machine_id, self.cfg.workstation_name) except CloudError as exc: log.debug("heartbeat failed for %s: %s", srv.cfg.name, exc) self._stop.wait(self.cfg.heartbeat_interval)