dwi_vendor/lis
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Dwi Swandhana
2026-04-14 09:54:48 +07:00
parent b34a379333
commit 676ec49cb2
1 changed files with 84 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
listener/app.py
+84 -9
View File
@@ -51,6 +51,7 @@ GENEXPERT_RESULT_QUERY_INITIAL_DELAY_SECONDS = 60
GENEXPERT_RESULT_QUERY_INTERVAL_SECONDS = 120
GENEXPERT_RESULT_QUERY_MAX_DURATION_SECONDS = 21600
GENEXPERT_RESULT_QUERY_INFLIGHT_TIMEOUT_SECONDS = 45
GENEXPERT_ENABLE_RESULT_QUERY_SCHEDULER = False
# Mapping Flag ke IP Address GeneXpert
# Pastikan IP ini SESUAI dengan settingan "Server IP" di masing-masing alat (Client Mode)
TARGET_MAPPING = {
@@ -81,7 +82,7 @@ GENEXPERT_TEST_MAPPING = {
GENEXPERT_IP_CAPABILITIES = {
"10.10.120.75": ["MTB-RIF", "MTB-RIF_ULTRA2", "MTB-XDR", "HIV-1_VL", "COV-2 2"],
"10.10.120.74": ["HCV", "HBV"],
"10.10.120.73": [] # Belum ada yang aktif
"10.10.120.73": ["MTB-RIF"] # Belum ada yang aktif
}
# Default code jika nama tes di database tidak dikenali
DEFAULT_GXP_CODE = "MTB-RIF"
@@ -248,6 +249,36 @@ def parse_genexpert_qpd(qpd_segment):
"param_2": param_2,
}
def resolve_genexpert_test_code(order, ip_addr=None):
assay_name = str(getattr(order, "tes", "") or "").strip()
test_code = GENEXPERT_TEST_MAPPING.get(assay_name)
supported_codes = None
if ip_addr is not None:
supported_codes = GENEXPERT_IP_CAPABILITIES.get(str(ip_addr or "").strip(), [])
if not test_code:
print(
f"[GENEXPERT] Skip rnoreg={getattr(order, 'rnoreg', '')} di {ip_addr}: "
f"mapping test code tidak ditemukan untuk tes='{assay_name}'"
)
return None
if supported_codes is not None and supported_codes and test_code not in supported_codes:
print(
f"[GENEXPERT] Skip rnoreg={getattr(order, 'rnoreg', '')} di {ip_addr}: "
f"test_code={test_code} tidak didukung alat"
)
return None
if supported_codes is not None and not supported_codes:
print(
f"[GENEXPERT] Skip rnoreg={getattr(order, 'rnoreg', '')} di {ip_addr}: "
"capability IP belum dikonfigurasi"
)
return None
return test_code
def get_genexpert_query_orders(ip_addr, hl7_msg):
flag = get_flag_by_device(ip_addr)
if not flag:
@@ -264,16 +295,48 @@ def get_genexpert_query_orders(ip_addr, hl7_msg):
if flag_attr is None:
return []
query = session.query(PaslabOrder).filter((flag_attr == False) | (flag_attr == None))
base_orders = session.query(PaslabOrder).filter(
(flag_attr == False) | (flag_attr == None)
).order_by(PaslabOrder.urut.asc()).all()
if param_1.upper() == "ALL":
return query.order_by(PaslabOrder.urut.asc()).all()
supported_codes = GENEXPERT_IP_CAPABILITIES.get(str(ip_addr or "").strip(), [])
if not supported_codes:
print(f"[GENEXPERT] Tidak ada capability test_code untuk IP {ip_addr}.")
return []
requested_sample_id = param_2 or param_1
if requested_sample_id:
return query.filter(PaslabOrder.rnoreg == requested_sample_id).order_by(PaslabOrder.urut.asc()).all()
requested_sample_id = (param_2 or param_1).strip() if (param_2 or param_1) else ""
if requested_sample_id and requested_sample_id.upper() != "ALL":
for order in base_orders:
if str(order.rnoreg or "").strip() != requested_sample_id:
continue
test_code = resolve_genexpert_test_code(order, ip_addr)
if test_code:
return [order]
print(
f"[GENEXPERT] Tidak ada order cocok/didukung untuk sample_id={requested_sample_id} di {ip_addr}"
)
return []
return query.order_by(PaslabOrder.urut.asc()).all()
selected_orders = []
selected_codes = set()
max_orders = max(len(supported_codes), 1)
for supported_code in supported_codes:
if len(selected_orders) >= max_orders:
break
for order in base_orders:
if any(existing.urut == order.urut for existing in selected_orders):
continue
test_code = resolve_genexpert_test_code(order, ip_addr)
if test_code != supported_code:
continue
if test_code in selected_codes:
continue
selected_orders.append(order)
selected_codes.add(test_code)
break
return selected_orders
finally:
session.close()
@@ -309,7 +372,7 @@ def create_genexpert_rsp_z02_response(orders, incoming_hl7):
patient_id = str(order.norm or order.rnoreg or "").strip()
sample_id = str(order.rnoreg or "").strip()
assay_name = str(order.tes or "").strip()
test_code = GENEXPERT_TEST_MAPPING.get(assay_name, DEFAULT_GXP_CODE)
test_code = resolve_genexpert_test_code(order) or GENEXPERT_TEST_MAPPING.get(assay_name) or DEFAULT_GXP_CODE
order_ts = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
segments.append(f"PID|{patient_idx}||{patient_id}")
@@ -569,6 +632,12 @@ def schedule_result_query_for_order(
interval_seconds=GENEXPERT_RESULT_QUERY_INTERVAL_SECONDS,
max_duration_seconds=GENEXPERT_RESULT_QUERY_MAX_DURATION_SECONDS,
):
if not GENEXPERT_ENABLE_RESULT_QUERY_SCHEDULER:
print(
f"[GENEXPERT-SCHEDULER] Nonaktif. Jadwal query hasil dilewati untuk accnumber={accnumber}"
)
return False
accnumber = str(accnumber or "").strip()
register_no = str(register_no or accnumber).strip()
target_ip = str(target_ip or "").strip() or None
@@ -619,6 +688,12 @@ def schedule_result_query_for_order(
return True
def trigger_result_query_to_genexpert(accnumber, register_no, target_ip=None, wait_seconds=20):
if not GENEXPERT_ENABLE_RESULT_QUERY_SCHEDULER:
return {
"ok": False,
"message": "Auto query hasil GeneXpert sedang dinonaktifkan sementara.",
}
active_ips = get_active_genexpert_ips()
if not active_ips:
return {"ok": False, "message": "Tidak ada koneksi GeneXpert aktif."}