Spaces:
Sleeping
Sleeping
File size: 5,644 Bytes
d18fef3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
"""
CDSS Rule Validator Component
"""
import gradio as gr
import json
import ast
from models import Vitals, PatientState
def test_condition(
patient_type, sbp, dbp, hr, rr, temp_c, spo2, labs_text, condition, alert_text
):
"""
Tests a single condition against a manually defined patient state.
"""
try:
labs = json.loads(labs_text)
except json.JSONDecodeError:
return "Error: Invalid JSON in Labs field."
vitals = Vitals(
sbp=int(sbp),
dbp=int(dbp),
hr=int(hr),
rr=int(rr),
temp_c=float(temp_c),
spo2=int(spo2),
)
state = PatientState(
scenario="Validation",
patient_type=patient_type,
notes="",
labs=labs,
vitals=vitals,
)
# Dynamically create a rule function for testing
rule_fnc_str = f"""
def dynamic_rule(state):
v = state.vitals
labs = state.labs
alerts = []
if {condition}:
alerts.append("{alert_text}")
if not alerts:
return "No alert triggered."
return "- ".join(["ALERT:"] + alerts)
"""
try:
exec(rule_fnc_str, globals())
result = dynamic_rule(state)
return result
except Exception as e:
return f"Error in condition syntax: {e}"
def add_rule_to_set(patient_type, condition, alert_text):
"""
Adds the new rule to the rules.py file.
"""
if not condition or not alert_text:
return "Error: Condition and Alert text cannot be empty."
try:
with open("rules.py", "r") as f:
tree = ast.parse(f.read())
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name == "rule_based_cdss":
for body_item in node.body:
if (
isinstance(body_item, ast.If)
and hasattr(body_item.test, "comparators")
and body_item.test.comparators
and isinstance(body_item.test.comparators[0], ast.Constant)
and body_item.test.comparators[0].value == patient_type
):
new_rule_str = (
f'if {condition}:\n alerts.append("{alert_text}")'
)
new_rule_node = ast.parse(new_rule_str).body[0]
body_item.body.append(new_rule_node)
break
new_code = ast.unparse(tree)
with open("rules.py", "w") as f:
f.write(new_code)
return f"Rule added to {patient_type} ruleset and saved to rules.py."
except Exception as e:
return f"Failed to add rule: {e}"
def validator_ui():
with gr.TabItem("Rule Validator"):
gr.Markdown("## Validate and Add New Rules")
with gr.Row():
with gr.Column():
gr.Markdown("### 1. Define Patient State")
patient_type_validate = gr.Radio(
["Mother", "Neonate", "Gyn"],
label="Patient Type",
value="Mother",
)
sbp_validate = gr.Number(label="SBP", value=120)
dbp_validate = gr.Number(label="DBP", value=80)
hr_validate = gr.Number(label="HR", value=80)
rr_validate = gr.Number(label="RR", value=18)
temp_c_validate = gr.Number(label="Temp (°C)", value=37.0)
spo2_validate = gr.Number(label="SpO₂ (%)", value=98)
labs_validate = gr.Textbox(
label="Labs (JSON format)",
value='{"Hb": 12.0}',
lines=3,
)
with gr.Column():
gr.Markdown("### 2. Define and Test Rule")
condition_validate = gr.Textbox(
label="Condition (Python expression)",
value="v.sbp > 140",
lines=3,
)
alert_validate = gr.Textbox(
label="Alert Message",
value="Preeclampsia suspected",
lines=3,
)
test_button = gr.Button("Test Rule", variant="secondary")
validation_result = gr.Textbox(
label="Validation Result", interactive=False
)
gr.Markdown("### 3. Add Rule to Ruleset")
add_rule_button = gr.Button(
"Add Rule to Ruleset", variant="primary"
)
add_rule_status = gr.Textbox(
label="Status", interactive=False
)
test_button.click(
test_condition,
inputs=[
patient_type_validate,
sbp_validate,
dbp_validate,
hr_validate,
rr_validate,
temp_c_validate,
spo2_validate,
labs_validate,
condition_validate,
alert_validate,
],
outputs=validation_result,
)
add_rule_button.click(
add_rule_to_set,
inputs=[
patient_type_validate,
condition_validate,
alert_validate,
],
outputs=add_rule_status,
)
return patient_type_validate, sbp_validate, dbp_validate, hr_validate, rr_validate, temp_c_validate, spo2_validate, labs_validate, condition_validate, alert_validate, test_button, validation_result, add_rule_button, add_rule_status
|