Membuat Middleware HL7 Listener untuk LIS Laboratorium

Artikel ini menjelaskan cara membuat middleware dengan Python untuk menerima data hasil laboratorium dari alat atau sistem lain dalam format HL7 dan mengirimkannya ke sistem LIS (Laboratory Information System) melalui API.

🧩 Tujuan Middleware

• Menerima data HL7 (ORM^O01 dan ORU^R01) via TCP/IP (port 2575).
• Melakukan parsing dan konversi ke format JSON.
• Mengirim data JSON ke REST API milik LIS.

🧱 Struktur Folder Proyek

hl7_lab_middleware/
├── config.py
├── hl7_parser.py
├── listener.py
└── sender.py

📄 config.py

HOST = '0.0.0.0'
PORT = 2575
API_ENDPOINT = 'http://localhost:8000/api/lab-results/'
API_KEY = 'your_api_key_here'

📄 hl7_parser.py

Mengandung 2 fungsi: parse_orm_message() dan parse_oru_message().

import hl7

def parse_orm_message(raw_hl7):
    h = hl7.parse(raw_hl7)
    result = {"tests": []}
    for segment in h:
        tag = segment[0]
        if tag == 'PID':
            result["patient_id"] = segment[2][0]
            result["patient_name"] = f"{segment[5][0][0]} {segment[5][0][1]}"
            result["dob"] = segment[7][0]
            result["gender"] = segment[8][0]
            addr = segment[11][0]
            result["address"] = f"{addr[0]}, {addr[2]}, {addr[3]} {addr[4]}"
        elif tag == 'PV1':
            doc = segment[7][0]
            result["doctor"] = f"{doc[1]} {doc[2]}"
        elif tag == 'ORC':
            result["order"] = {
                "order_id": segment[1][0],
                "order_status": segment[5][0],
                "order_date": segment[9][0],
                "ordered_by": segment[10][1] + " " + segment[10][2]
            }
        elif tag == 'OBR':
            test_code = segment[4][0]
            test_name = segment[4][1]
            result["tests"].append({"test_code": test_code, "test_name": test_name})
        elif tag == 'DG1':
            result["diagnosis_code"] = segment[3][0]
            result["diagnosis"] = segment[3][1]
    return result

def parse_oru_message(raw_hl7):
    h = hl7.parse(raw_hl7)
    result = {"tests": []}
    for segment in h:
        tag = segment[0]
        if tag == 'PID':
            result["patient_id"] = segment[3][0]
            result["patient_name"] = f"{segment[5][0][1]} {segment[5][0][0]}"
            result["dob"] = segment[7][0]
            result["gender"] = segment[8][0]
        elif tag == 'OBR':
            result["order_id"] = segment[2][0]
        elif tag == 'OBX':
            result["tests"].append({
                "test_code": segment[3][0],
                "test_name": segment[3][1],
                "value": segment[5][0],
                "unit": segment[6][0],
                "reference_range": segment[7][0],
                "abnormal_flag": segment[8][0]
            })
    return result

📄 sender.py

import requests
from config import API_ENDPOINT, API_KEY

def send_to_lis(data):
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }
    try:
        response = requests.post(API_ENDPOINT, json=data, headers=headers)
        print(f"✅ Sent to LIS | Status: {response.status_code} | Response: {response.text}")
    except Exception as e:
        print(f"❌ Failed to send to LIS: {e}")

📄 listener.py

import socket
from hl7_parser import parse_orm_message, parse_oru_message
from sender import send_to_lis
from config import HOST, PORT

def start_listener():
    print(f"🔌 Starting HL7 Listener on {HOST}:{PORT}")
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        s.listen(5)

        while True:
            conn, addr = s.accept()
            with conn:
                print(f"📥 Connection from {addr}")
                data = b""
                while True:
                    chunk = conn.recv(1024)
                    if not chunk:
                        break
                    data += chunk
                message = data.decode(errors='ignore').strip()
                print(f"[RAW HL7 MESSAGE]\\n{message}\\n")

                if "ORM^O01" in message:
                    parsed = parse_orm_message(message)
                elif "ORU^R01" in message:
                    parsed = parse_oru_message(message)
                else:
                    print("❓ Unknown HL7 message type.")
                    parsed = None

                print("[PARSED]", parsed)
                if parsed:
                    send_to_lis(parsed)

if __name__ == "__main__":
    start_listener()

⚙️ Cara Menjalankan Middleware

1. Install dependencies:

   pip install hl7 requests

2. Jalankan listener:

   python listener.py

3. Kirim HL7 melalui TCP ke port 2575 (gunakan netcat, Mirth Connect, atau HL7 Simulator).

✅ Catatan Tambahan

• Format HL7 dari alat atau HIS bisa berbeda. Selalu validasi dan log data mentah.
• Tambahkan retry dan logging untuk lingkungan produksi.
• Bisa diintegrasikan lebih lanjut dengan Docker, Mirth Connect, dan server LIS sebenarnya.

Artikel ini dibuat untuk mendukung implementasi sistem informasi laboratorium (LIS) yang interoperabel dan fleksibel.