refactor: implement packet parsing heuristics and introduce device-level state caching to throttle database writes and vitals updates
This commit is contained in:
@@ -65,12 +65,15 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
|
||||
break
|
||||
|
||||
buffer += data
|
||||
# Temporary debug capture of raw binary stream
|
||||
with open("/home/prathiyuman/Prathiyuman/ContecMonitor/fiveparaminte-main/scratch/raw_contec_stream.bin", "ab") as f:
|
||||
f.write(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:
|
||||
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)
|
||||
@@ -102,7 +105,7 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
|
||||
b2 = buffer[consumed + 2]
|
||||
b3 = buffer[consumed + 3]
|
||||
|
||||
if b2 == 0x04 and b3 == 0x46:
|
||||
if (b2 == 0x04 or b2 == 0x01 or b2 == 0x00) and b3 == 0x46:
|
||||
pkt_len = b0 | (b1 << 8)
|
||||
pkt_end = consumed + pkt_len
|
||||
|
||||
@@ -129,7 +132,7 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
|
||||
|
||||
consumed = pkt_end
|
||||
|
||||
elif b2 == 0x04 and b3 == 0x47:
|
||||
elif (b2 == 0x04 or b2 == 0x01) and b3 == 0x47:
|
||||
# End-of-frame marker — skip 4 bytes
|
||||
consumed += 4
|
||||
|
||||
@@ -161,13 +164,63 @@ async def handle_contec_client(reader: asyncio.StreamReader, writer: asyncio.Str
|
||||
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.
|
||||
"""
|
||||
display = get_terminal_display()
|
||||
device_id = vitals.device_id
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# Update console display banner
|
||||
if device_id not in device_field_timestamps:
|
||||
device_field_timestamps[device_id] = {}
|
||||
|
||||
# 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(
|
||||
@@ -177,7 +230,7 @@ async def process_vitals(vitals):
|
||||
f"RR={vitals.respiratory_rate}, Temp={vitals.temperature}"
|
||||
)
|
||||
|
||||
# Broadcast to live WebSockets dashboard
|
||||
# 3. Broadcast to live WebSockets dashboard in real-time
|
||||
try:
|
||||
from routers.dashboard import get_ws_manager
|
||||
ws_manager = get_ws_manager()
|
||||
@@ -185,63 +238,78 @@ async def process_vitals(vitals):
|
||||
except Exception as e:
|
||||
logger.debug(f"WebSocket broadcast failed: {e}")
|
||||
|
||||
# Save to SQLite database
|
||||
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="unknown")
|
||||
db.add(device)
|
||||
else:
|
||||
device.last_seen = datetime.now(timezone.utc)
|
||||
|
||||
# 2. Patient tracking
|
||||
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
|
||||
if not patient:
|
||||
patient = Patient(patient_id=vitals.patient_id, name="Unknown")
|
||||
db.add(patient)
|
||||
|
||||
# 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
|
||||
# 4. Throttled Database & REST forwarding (at most once every 5 seconds)
|
||||
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
|
||||
patient = db.query(Patient).filter(Patient.patient_id == vitals.patient_id).first()
|
||||
if not patient:
|
||||
patient = Patient(patient_id=vitals.patient_id, name="Unknown")
|
||||
db.add(patient)
|
||||
|
||||
# 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
|
||||
)
|
||||
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()
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user