357 lines
14 KiB
Python
357 lines
14 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from contec_parser import parse_contec_data
|
|
from database import SessionLocal
|
|
from models import Device, Patient, VitalReading, TransmissionLog
|
|
from api_client import forward_vitals_to_api
|
|
from terminal_display import get_terminal_display
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Constants for MLLP framing (used on port 511 for Contec HL7 text)
|
|
VT = b'\x0b'
|
|
FS_CR = b'\x1c\x0d'
|
|
|
|
|
|
def generate_ack(received_msg: str) -> str:
|
|
"""Generates a simple HL7 ACK message to satisfy the monitor's TCP client."""
|
|
try:
|
|
segments = received_msg.split('\r')
|
|
msh = segments[0] if segments else ""
|
|
if not msh.startswith("MSH|"):
|
|
return ""
|
|
|
|
fields = msh.split('|')
|
|
if len(fields) < 10:
|
|
return ""
|
|
|
|
sending_app = fields[2]
|
|
sending_fac = fields[3]
|
|
rec_app = fields[4]
|
|
rec_fac = fields[5]
|
|
msg_control_id = fields[9]
|
|
|
|
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
|
|
ack_msh = f"MSH|^~\\&|{rec_app}|{rec_fac}|{sending_app}|{sending_fac}|{now_str}||ACK^R01|{msg_control_id}|P|2.3.1\r"
|
|
ack_msa = f"MSA|AA|{msg_control_id}\r"
|
|
|
|
return ack_msh + ack_msa
|
|
except Exception as e:
|
|
logger.debug(f"Could not generate ACK: {e}")
|
|
return ""
|
|
|
|
|
|
async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, port: int):
|
|
"""
|
|
TCP server handler for Contec CMS7000PLUS client connections.
|
|
Directs incoming traffic to the unified parse_contec_data handler.
|
|
"""
|
|
client_ip, client_port = writer.get_extra_info('peername')
|
|
logger.info(f"Accepted Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
|
|
|
|
display = get_terminal_display()
|
|
display.connection_opened(client_ip, port)
|
|
|
|
buffer = b""
|
|
|
|
try:
|
|
while True:
|
|
data = await reader.read(8192)
|
|
if not data:
|
|
logger.info(f"Contec client {client_ip}:{client_port} on port {port} sent EOF (0 bytes)")
|
|
break
|
|
|
|
buffer += data
|
|
logger.debug(f"Received {len(data)} bytes from {client_ip}:{client_port} on port {port} (buf={len(buffer)})")
|
|
|
|
# ---------------------------------------------------------------
|
|
# MLLP / HL7 text path (typically port 511)
|
|
# ---------------------------------------------------------------
|
|
if VT in buffer and FS_CR in buffer and b"MSH|" in buffer:
|
|
while True:
|
|
start_idx = buffer.find(VT)
|
|
end_idx = buffer.find(FS_CR)
|
|
if start_idx != -1 and end_idx != -1 and start_idx < end_idx:
|
|
raw_hl7 = buffer[start_idx + 1:end_idx].decode('utf-8', errors='ignore')
|
|
buffer = buffer[end_idx + 2:]
|
|
logger.info(f"Contec MLLP message detected on port {port}")
|
|
vitals = parse_contec_data(raw_hl7.encode('utf-8'), client_ip)
|
|
if vitals:
|
|
await process_vitals(vitals)
|
|
ack_msg = generate_ack(raw_hl7)
|
|
if ack_msg:
|
|
writer.write(VT + ack_msg.encode('utf-8') + FS_CR)
|
|
await writer.drain()
|
|
else:
|
|
break
|
|
continue # Go back to reading more data
|
|
|
|
# ---------------------------------------------------------------
|
|
# Binary Contec proprietary protocol (port 514)
|
|
# Packets are framed as: [len_lo][len_hi][04][46] ... payload ...
|
|
# The monitor sends each packet as a complete TCP write,
|
|
# but we buffer anyway in case of TCP segmentation.
|
|
# ---------------------------------------------------------------
|
|
consumed = 0
|
|
while consumed < len(buffer) - 4:
|
|
b0 = buffer[consumed]
|
|
b1 = buffer[consumed + 1]
|
|
b2 = buffer[consumed + 2]
|
|
b3 = buffer[consumed + 3]
|
|
|
|
if (b2 == 0x04 or b2 == 0x01 or b2 == 0x00) and b3 == 0x46:
|
|
pkt_len = b0 | (b1 << 8)
|
|
pkt_end = consumed + pkt_len
|
|
|
|
if pkt_len < 5 or pkt_len > 8192:
|
|
# Bad length — skip one byte and keep scanning
|
|
consumed += 1
|
|
continue
|
|
|
|
if pkt_end > len(buffer):
|
|
# Don't have the full packet yet — wait for more data
|
|
break
|
|
|
|
# Parse the complete packet
|
|
pkt = buffer[consumed:pkt_end]
|
|
vitals = parse_contec_data(pkt, client_ip)
|
|
if vitals:
|
|
logger.info(
|
|
f"[PORT {port}] Vitals from {client_ip}: "
|
|
f"SpO2={vitals.spo2}%, HR={vitals.heart_rate}bpm "
|
|
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp} "
|
|
f"Temp={vitals.temperature}C"
|
|
)
|
|
await process_vitals(vitals)
|
|
|
|
consumed = pkt_end
|
|
|
|
elif (b2 == 0x04 or b2 == 0x01) and b3 == 0x47:
|
|
# End-of-frame marker — skip 4 bytes
|
|
consumed += 4
|
|
|
|
else:
|
|
# Not a recognised frame start — advance one byte
|
|
consumed += 1
|
|
|
|
# Discard consumed bytes from front of buffer
|
|
buffer = buffer[consumed:]
|
|
|
|
# Safety valve
|
|
if len(buffer) > 32768:
|
|
logger.warning(f"Buffer too large ({len(buffer)} bytes) on port {port}, resetting")
|
|
buffer = b""
|
|
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except ConnectionResetError:
|
|
logger.info(f"Contec CMS7000PLUS connection reset from {client_ip} on port {port}")
|
|
except Exception as e:
|
|
logger.error(f"Error in Contec client handler (port {port}): {e}")
|
|
finally:
|
|
logger.info(f"Closing Contec CMS7000PLUS connection from {client_ip}:{client_port} on port {port}")
|
|
display.connection_closed(client_ip, port)
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# Cache to hold merged vitals per device to prevent fragmentation
|
|
device_cache = {}
|
|
device_field_timestamps = {} # {device_id: {field_name: datetime}}
|
|
last_db_write = {}
|
|
|
|
|
|
async def process_vitals(vitals):
|
|
"""
|
|
Log, database, WebSocket broadcast, and REST forward parsed vitals.
|
|
"""
|
|
device_id = vitals.device_id
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if device_id not in device_field_timestamps:
|
|
device_field_timestamps[device_id] = {}
|
|
|
|
# Check if incoming packet has fresh waveforms
|
|
has_fresh_waveforms = (vitals.ecg_wave is not None) or (vitals.pleth_wave is not None)
|
|
|
|
# 1. Merge new vitals into cached vitals to prevent fragmented entries
|
|
if device_id not in device_cache:
|
|
device_cache[device_id] = vitals
|
|
# Track initial timestamps for present values
|
|
for field, value in vitals.model_dump().items():
|
|
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
|
|
continue
|
|
if value is not None:
|
|
device_field_timestamps[device_id][field] = now
|
|
else:
|
|
cached = device_cache[device_id]
|
|
|
|
# Determine active fields in incoming packet
|
|
active_fields = []
|
|
if vitals.present_fields is not None:
|
|
active_fields = vitals.present_fields
|
|
else:
|
|
for field, value in vitals.model_dump().items():
|
|
if field == "present_fields" or field in ["device_id", "patient_id", "timestamp"]:
|
|
continue
|
|
if value is not None:
|
|
active_fields.append(field)
|
|
|
|
# Update cache & record timestamps
|
|
for field in active_fields:
|
|
val = getattr(vitals, field)
|
|
setattr(cached, field, val)
|
|
device_field_timestamps[device_id][field] = now
|
|
|
|
# Clean up stale fields (older than 15 seconds)
|
|
for field in list(device_field_timestamps[device_id].keys()):
|
|
last_updated = device_field_timestamps[device_id][field]
|
|
if (now - last_updated).total_seconds() > 15.0:
|
|
setattr(cached, field, None)
|
|
device_field_timestamps[device_id].pop(field, None)
|
|
|
|
cached.timestamp = vitals.timestamp
|
|
vitals = cached
|
|
|
|
# 2. Update console display banner
|
|
display = get_terminal_display()
|
|
display.update_vitals(vitals)
|
|
|
|
logger.info(
|
|
f"Vitals for Device {vitals.device_id}, Patient {vitals.patient_id}: "
|
|
f"HR={vitals.heart_rate}, SpO2={vitals.spo2}, "
|
|
f"BP={vitals.systolic_bp}/{vitals.diastolic_bp}, "
|
|
f"RR={vitals.respiratory_rate}, Temp={vitals.temperature}"
|
|
)
|
|
|
|
# 3. Broadcast to live WebSockets dashboard in real-time
|
|
try:
|
|
from routers.dashboard import get_ws_manager
|
|
ws_manager = get_ws_manager()
|
|
await ws_manager.broadcast_vitals(vitals.model_dump(mode="json"))
|
|
except Exception as e:
|
|
logger.debug(f"WebSocket broadcast failed: {e}")
|
|
|
|
# 4. Throttled Database & REST forwarding (at most once every 5 seconds)
|
|
# Note: If we have fresh waveforms, we bypass throttle for REST forwarding to keep cloud graphs real-time
|
|
now = datetime.now(timezone.utc)
|
|
should_write_db = False
|
|
if device_id not in last_db_write or (now - last_db_write[device_id]).total_seconds() >= 5.0:
|
|
should_write_db = True
|
|
last_db_write[device_id] = now
|
|
|
|
if should_write_db:
|
|
# Save to SQLite database and REST forward
|
|
db = SessionLocal()
|
|
try:
|
|
# 1. Device tracking
|
|
device = db.query(Device).filter(Device.device_id == vitals.device_id).first()
|
|
if not device:
|
|
device = Device(
|
|
device_id=vitals.device_id,
|
|
ip_address=vitals.ip_address or "unknown",
|
|
status="active"
|
|
)
|
|
db.add(device)
|
|
else:
|
|
if vitals.ip_address:
|
|
device.ip_address = vitals.ip_address
|
|
device.status = "active"
|
|
device.last_seen = datetime.now(timezone.utc)
|
|
|
|
# 2. Patient tracking — store/update demographics from HL7
|
|
if vitals.patient_id == "UNKNOWN":
|
|
last_known_reading = (
|
|
db.query(VitalReading)
|
|
.filter(VitalReading.device_id == vitals.device_id)
|
|
.filter(VitalReading.patient_id != "UNKNOWN")
|
|
.order_by(VitalReading.timestamp.desc())
|
|
.first()
|
|
)
|
|
if last_known_reading:
|
|
vitals.patient_id = last_known_reading.patient_id
|
|
|
|
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
|
|
if not patient:
|
|
patient = Patient(
|
|
patient_id=vitals.patient_id,
|
|
mrn=vitals.patient_id,
|
|
name=vitals.patient_name or "Unknown",
|
|
gender=vitals.patient_gender,
|
|
dob=vitals.patient_dob,
|
|
bed_id=vitals.bed_id,
|
|
)
|
|
db.add(patient)
|
|
else:
|
|
# Refresh demographics whenever monitor sends updated PID fields
|
|
if vitals.patient_name:
|
|
patient.name = vitals.patient_name
|
|
if vitals.patient_gender:
|
|
patient.gender = vitals.patient_gender
|
|
if vitals.patient_dob:
|
|
patient.dob = vitals.patient_dob
|
|
if vitals.bed_id:
|
|
patient.bed_id = vitals.bed_id
|
|
|
|
|
|
# 3. Create Reading Entry
|
|
reading = VitalReading(
|
|
device_id=vitals.device_id,
|
|
patient_id=vitals.patient_id,
|
|
timestamp=vitals.timestamp,
|
|
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,
|
|
transmitted=False
|
|
)
|
|
db.add(reading)
|
|
db.commit()
|
|
db.refresh(reading)
|
|
|
|
# 4. REST forwarding
|
|
success = await forward_vitals_to_api(vitals)
|
|
|
|
if success:
|
|
reading.transmitted = True
|
|
log_entry = TransmissionLog(
|
|
reading_id=reading.id,
|
|
status="success",
|
|
response_code=200
|
|
)
|
|
else:
|
|
log_entry = TransmissionLog(
|
|
reading_id=reading.id,
|
|
status="failed",
|
|
error_message="Immediate REST forward failed"
|
|
)
|
|
|
|
db.add(log_entry)
|
|
db.commit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Database operation failed: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
elif has_fresh_waveforms:
|
|
# Bypass throttle for REST forwarding to keep cloud graphs real-time
|
|
try:
|
|
await forward_vitals_to_api(vitals)
|
|
except Exception as e:
|
|
logger.error(f"Real-time REST forwarding failed: {e}")
|
|
|
|
# Clear waveforms from cache to prevent repetition in subsequent non-waveform sub-packets
|
|
if device_id in device_cache:
|
|
device_cache[device_id].ecg_wave = None
|
|
device_cache[device_id].pleth_wave = None
|