74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import socket
|
|
import time
|
|
import random
|
|
from datetime import datetime
|
|
import threading
|
|
|
|
HOST = "127.0.0.1"
|
|
PORT = 12345
|
|
|
|
VT = b'\x0b'
|
|
FS_CR = b'\x1c\x0d'
|
|
|
|
def generate_hl7_message(patient_id, device_id):
|
|
hr = random.randint(70, 95)
|
|
spo2 = random.randint(96, 99)
|
|
sys = random.randint(115, 125)
|
|
dia = random.randint(75, 83)
|
|
map_bp = int(dia + (sys - dia) / 3)
|
|
resp = random.randint(14, 18)
|
|
temp = round(random.uniform(36.5, 37.1), 1)
|
|
|
|
now_str = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
|
|
msh = f"MSH|^~\\&|{device_id}|MOCK_FACILITY|RECEIVING_APP|RECEIVING_FACILITY|{now_str}||ORU^R01|MSG{now_str}|P|2.3.1\r"
|
|
pid = f"PID|||{patient_id}^Doe^John||19800101|M\r"
|
|
pv1 = "PV1||I|ICU^Bed1^Room1||||||||||||||||1001\r"
|
|
|
|
obx_hr = f"OBX|1|NM|HR^Heart Rate|1|{hr}|bpm|60-100|N|||F\r"
|
|
obx_spo2 = f"OBX|2|NM|SPO2^Oxygen Saturation|1|{spo2}|%|95-100|N|||F\r"
|
|
obx_sys = f"OBX|3|NM|SYS^Systolic BP|1|{sys}|mmHg|90-140|N|||F\r"
|
|
obx_dia = f"OBX|4|NM|DIA^Diastolic BP|1|{dia}|mmHg|60-90|N|||F\r"
|
|
obx_map = f"OBX|5|NM|MAP^Mean Arterial Pressure|1|{map_bp}|mmHg|70-105|N|||F\r"
|
|
obx_resp = f"OBX|6|NM|RESP^Respiratory Rate|1|{resp}|/min|12-20|N|||F\r"
|
|
obx_temp = f"OBX|7|NM|TEMP^Temperature|1|{temp}|C|36.1-37.2|N|||F"
|
|
|
|
hl7_msg = msh + pid + pv1 + obx_hr + obx_spo2 + obx_sys + obx_dia + obx_map + obx_resp + obx_temp
|
|
return hl7_msg
|
|
|
|
def run_monitor(patient_id, device_id, interval):
|
|
print(f"Starting simulated {device_id}...")
|
|
while True:
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.connect((HOST, PORT))
|
|
print(f"[{device_id}] Connected to server")
|
|
while True:
|
|
hl7_msg = generate_hl7_message(patient_id, device_id)
|
|
framed_msg = VT + hl7_msg.encode('utf-8') + FS_CR
|
|
s.sendall(framed_msg)
|
|
try:
|
|
s.settimeout(1.0)
|
|
response = s.recv(4096)
|
|
except socket.timeout:
|
|
pass
|
|
time.sleep(interval)
|
|
except Exception as e:
|
|
print(f"[{device_id}] Error: {e}. Reconnecting in 3 seconds...")
|
|
time.sleep(3)
|
|
|
|
def main():
|
|
t1 = threading.Thread(target=run_monitor, args=("PAT-1001", "CMS7000PLUS_127.0.0.1", 3), daemon=True)
|
|
t2 = threading.Thread(target=run_monitor, args=("PAT-8500", "CMS8500_127.0.0.1", 4), daemon=True)
|
|
t1.start()
|
|
t2.start()
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print("Stopping simulators.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|