validator.py

"""
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