247 lines
9.2 KiB
Python
247 lines
9.2 KiB
Python
"""
|
|
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
|