""" 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