refactored

cdss.py

@@ -18,596 +18,23 @@ python app.py
 """
 
 from __future__ import annotations
-import os
-import random
 import time
-from dataclasses import asdict
-from typing import Dict, Any, Tuple, List
-from datetime import datetime
 
 import gradio as gr
 import pandas as pd
-import plotly.express as px
 
 from models import Vitals, PatientState
-from rules import rule_based_cdss
-
-GEMINI_MODEL_NAME = "gemini-2.5-pro"
-# --- Gemini setup (simplified) ---
-try:
-    import google.generativeai as genai
-
-    genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
-    GEMINI_MODEL = genai.GenerativeModel(GEMINI_MODEL_NAME)
-    GEMINI_ERR = None
-except Exception as e:
-    GEMINI_MODEL, GEMINI_ERR = None, f"Gemini import/config error: {e}"
-
-
-# --- Data structures & Scenarios (Full list included) ---
-
-
-def scenario_A0_Normal() -> PatientState:
-    return PatientState(
-        "A0 Normal Case",
-        "Mother",
-        "Pemeriksaan rutin.",
-        {"Hb": 12.5},
-        Vitals(110, 70, 80, 16, 36.7, 99),
-    )
-
-
-def scenario_A1_PPH() -> PatientState:
-    return PatientState(
-        "A1 PPH",
-        "Mother",
-        "30 menit postpartum; kehilangan darah ~900 ml.",
-        {"Hb": 9},
-        Vitals(90, 60, 120, 24, 36.8, 96),
-    )
-
-
-def scenario_A2_Preeclampsia() -> PatientState:
-    return PatientState(
-        "A2 Preeklampsia",
-        "Mother",
-        "36 minggu; sakit kepala, pandangan kabur.",
-        {"Proteinuria": "3+"},
-        Vitals(165, 105, 98, 20, 36.9, 98),
-    )
-
-
-def scenario_A3_MaternalSepsis() -> PatientState:
-    return PatientState(
-        "A3 Sepsis Maternal",
-        "Mother",
-        "POD2 pasca SC; luka purulen.",
-        {"Leukosit": 17000},
-        Vitals(95, 60, 110, 24, 39.0, 96),
-    )
-
-
-def scenario_B1_Prematurity() -> PatientState:
-    return PatientState(
-        "B1 Prematuritas/BBLR",
-        "Neonate",
-        "34 minggu; berat 1900 g; hipotermia ringan; SpO2 borderline",
-        {"BB": 1900, "UsiaGestasi_mgg": 34},
-        Vitals(60, 35, 150, 50, 35.0, 90),
-    )
-
-
-def scenario_B2_Asphyxia() -> PatientState:
-    return PatientState(
-        "B2 Asfiksia Perinatal",
-        "Neonate",
-        "APGAR 3 menit 1; tidak menangis >1 menit",
-        {"APGAR_1m": 3},
-        Vitals(55, 30, 80, 10, 36.5, 82),
-    )
-
-
-def scenario_B3_NeonatalSepsis() -> PatientState:
-    return PatientState(
-        "B3 Sepsis Neonatal",
-        "Neonate",
-        "Hari ke-4; lemas, malas minum",
-        {"CRP": 25, "Leukosit": 19000},
-        Vitals(60, 35, 170, 60, 38.5, 93),
-    )
-
-
-def scenario_C1_GynSurgComp() -> PatientState:
-    return PatientState(
-        "C1 Komplikasi Bedah Ginekologis",
-        "Gyn",
-        "Pasca histerektomi; nyeri perut bawah; urine output turun",
-        {"UrineOutput_ml_hr": 10},
-        Vitals(100, 65, 105, 20, 37.8, 98),
-    )
-
-
-def scenario_C2_PostOpInfection() -> PatientState:
-    return PatientState(
-        "C2 Infeksi Pasca-Bedah",
-        "Gyn",
-        "Pasca kistektomi; luka bengkak & kemerahan; demam",
-        {"Luka": "bengkak+kemerahan"},
-        Vitals(105, 70, 108, 22, 38.0, 98),
-    )
-
-
-def scenario_C3_DelayedGynCancer() -> PatientState:
-    return PatientState(
-        "C3 Keterlambatan Diagnostik Kanker Ginekologi",
-        "Gyn",
-        "45 th; perdarahan pascamenopause; Pap abnormal 6 bulan lalu tanpa tindak lanjut",
-        {"PapSmear": "abnormal 6 bln lalu"},
-        Vitals(120, 78, 86, 18, 36.8, 99),
-    )
-
-
-SCENARIOS = {
-    "A0": scenario_A0_Normal,
-    "A1": scenario_A1_PPH,
-    "A2": scenario_A2_Preeclampsia,
-    "A3": scenario_A3_MaternalSepsis,
-    "B1": scenario_B1_Prematurity,
-    "B2": scenario_B2_Asphyxia,
-    "B3": scenario_B3_NeonatalSepsis,
-    "C1": scenario_C1_GynSurgComp,
-    "C2": scenario_C2_PostOpInfection,
-    "C3": scenario_C3_DelayedGynCancer,
-}
-
-
-# --- Simulation & CDSS Logic (simplified) ---
-def drift_vitals(state: PatientState) -> PatientState:
-    v = state.vitals
-    clamp = lambda val, lo, hi: max(lo, min(hi, val))
-    drift_factor = 0 if state.scenario.startswith("A0") else 1
-    v.hr = clamp(v.hr + random.randint(-2, 2) * drift_factor, 40, 200)
-    v.sbp = clamp(v.sbp + random.randint(-2, 2) * drift_factor, 50, 220)
-    v.rr = clamp(v.rr + random.randint(-1, 1) * drift_factor, 8, 80)
-    state.vitals = v
-    return state
-
-
-# --- Rule-based fallback (no AI or AI disabled) ---
-
-
-def gemini_cdss(state: PatientState) -> str:
-    if not GEMINI_MODEL:
-        return f"[CDSS AI ERROR] {GEMINI_ERR}"
-    try:
-        v = state.vitals
-        prompt = f"CDSS for {state.scenario}. Vitals: SBP {v.sbp}/{v.dbp}, HR {v.hr}. Analyze risks, give concise steps in Indonesian."
-        return GEMINI_MODEL.generate_content(prompt).text or "[CDSS AI] No response."
-    except Exception as e:
-        return f"[CDSS AI error] {e}"
-
-
-# --- Plotting & Data Helpers ---
-def create_vital_plot(
-    df: pd.DataFrame, y_cols: List[str] | str, title: str, y_lim: List[int]
-):
-    """Creates a customized Plotly figure for a specific vital sign."""
-    # Create an empty plot if there is no data to prevent errors
-    if df.empty:
-        fig = px.line(title=title)
-    else:
-        fig = px.line(df, x="timestamp", y=y_cols, title=title, markers=True)
-        # Customize x-axis to show only first and last tick
-        if len(df) > 1:
-            fig.update_xaxes(
-                tickvals=[df["timestamp"].iloc[0], df["timestamp"].iloc[-1]]
-            )
-
-    # Apply standard layout settings
-    fig.update_layout(
-        height=250,
-        yaxis_range=y_lim,
-        margin=dict(t=40, b=10, l=10, r=10),  # Tighten margins
-    )
-    return fig
-
-
-def _row_from_state(ps: PatientState) -> Dict[str, Any]:
-    return {"timestamp": datetime.now(), "scenario": ps.scenario, **asdict(ps.vitals)}
-
-
-def prepare_df_for_display(df: pd.DataFrame) -> pd.DataFrame:
-    if df is None or df.empty:
-        return pd.DataFrame(
-            columns=[
-                "timestamp",
-                "scenario",
-                "sbp",
-                "dbp",
-                "hr",
-                "rr",
-                "temp_c",
-                "spo2",
-            ]
-        )
-    df_display = df.copy()
-    df_display["timestamp"] = pd.to_datetime(df_display["timestamp"])
-    df_display = df_display.sort_values("timestamp")
-    df_display["timestamp"] = df_display["timestamp"].dt.strftime("%Y-%m-%d %H:%M:%S")
-    return df_display
-
-
-def generate_all_plots(df: pd.DataFrame):
-    """Helper to generate all 5 plot figures from a dataframe."""
-    df_display = prepare_df_for_display(df)
-    bp_fig = create_vital_plot(
-        df_display,
-        y_cols=["sbp", "dbp"],
-        title="Blood Pressure (mmHg)",
-        y_lim=[40, 200],
-    )
-    hr_fig = create_vital_plot(
-        df_display, y_cols="hr", title="Heart Rate (bpm)", y_lim=[40, 200]
-    )
-    rr_fig = create_vital_plot(
-        df_display, y_cols="rr", title="Respiratory Rate (/min)", y_lim=[0, 70]
-    )
-    temp_fig = create_vital_plot(
-        df_display, y_cols="temp_c", title="Temperature (°C)", y_lim=[34, 42]
-    )
-    spo2_fig = create_vital_plot(
-        df_display, y_cols="spo2", title="Oxygen Saturation (%)", y_lim=[70, 101]
-    )
-    return df_display, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig
-
-
-# --- Gradio App Logic ---
-def process_and_update(
-    ps: PatientState, history_df: pd.DataFrame, historic_text: str, cdss_on: bool
-):
-    """Centralized function to process state, update history, and generate all UI component outputs."""
-    interpretation = gemini_cdss(ps) if cdss_on else rule_based_cdss(ps)
-    new_row = _row_from_state(ps)
-    history_df = pd.concat([history_df, pd.DataFrame([new_row])], ignore_index=True)
-
-    df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig = generate_all_plots(
-        history_df
-    )
-
-    return (
-        asdict(ps),
-        *state_to_panels(ps),
-        str(ps.labs),  # For labs_text
-        str(ps.labs),  # For labs_show
-        interpretation,
-        history_df,
-        df_for_table,
-        historic_text.strip(),
-        time.time(),
-        bp_fig,
-        hr_fig,
-        rr_fig,
-        temp_fig,
-        spo2_fig,
-    )
-
-
-def state_to_panels(state: PatientState) -> Tuple:
-    v = state.vitals
-    return (
-        state.scenario,
-        state.patient_type,
-        state.notes,
-        v.sbp,
-        v.dbp,
-        v.hr,
-        v.rr,
-        v.temp_c,
-        v.spo2,
-    )
-
-
-def inject_scenario(
-    tag: str, cdss_on: bool, history_df: pd.DataFrame, historic_text: str
-):
-    ps = SCENARIOS[tag]()
-    if historic_text:  # Add a newline if text already exists
-        historic_text += f"\n[{datetime.now().strftime('%H:%M:%S')}] Scenario Injected: {ps.scenario}"
-    else:
-        historic_text = (
-            f"[{datetime.now().strftime('%H:%M:%S')}] Scenario Injected: {ps.scenario}"
-        )
-    return process_and_update(ps, history_df, historic_text, cdss_on)
-
-
-def manual_edit(
-    sbp,
-    dbp,
-    hr,
-    rr,
-    temp_c,
-    spo2,
-    notes,
-    labs_text,
-    cdss_on,
-    patient_type,
-    current_state,
-    history_df,
-    historic_text,
-):
-    try:
-        labs = eval(labs_text)
-    except:
-        labs = {"raw": labs_text}
-    ps = PatientState(
-        current_state.get("scenario", "Manual"),
-        patient_type,
-        notes,
-        labs,
-        Vitals(int(sbp), int(dbp), int(hr), int(rr), float(temp_c), int(spo2)),
-    )
-    if ps.notes and ps.notes.strip():
-        historic_text += f"\n[{datetime.now().strftime('%H:%M:%S')}] {ps.notes}"
-    return process_and_update(ps, history_df, historic_text, cdss_on)
-
-
-def tick_timer(cdss_on, current_state, history_df, historic_text):
-    if not current_state:
-        return [gr.update()] * 22
-    ps = PatientState(**current_state)
-    ps.vitals = Vitals(**ps.vitals)
-    ps = drift_vitals(ps)
-    return process_and_update(ps, history_df, historic_text, cdss_on)
-
-
-def load_csv(file, history_df: pd.DataFrame):
-    try:
-        if file is not None:
-            df_new = pd.read_csv(file.name)
-            df_new["timestamp"] = pd.to_datetime(df_new["timestamp"])
-            history_df = (
-                pd.concat([history_df, df_new], ignore_index=True)
-                if not history_df.empty
-                else df_new
-            )
-    except Exception as e:
-        print(f"Error loading CSV: {e}")
-    df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig = generate_all_plots(
-        history_df
-    )
-    return history_df, df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig
-
-
-def countdown_tick(last_tick_ts: float):
-    if not last_tick_ts:
-        return "Next update in —"
-    return f"Next update in {max(0, 30 - int(time.time() - last_tick_ts))}s"
-
-
-import json
-import ast
-
-
-def parse_rules():
-    with open("rules.py", "r") as f:
-        tree = ast.parse(f.read())
-
-    rules = {"Mother": [], "Neonate": [], "Gyn": []}
-
-    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 isinstance(body_item.test, ast.Compare)
-                    and isinstance(body_item.test.left, ast.Attribute)
-                    and isinstance(body_item.test.left.value, ast.Name)
-                    and body_item.test.left.value.id == "state"
-                    and body_item.test.left.attr == "patient_type"
-                    and isinstance(body_item.test.ops[0], ast.Eq)
-                    and isinstance(body_item.test.comparators[0], ast.Constant)
-                ):
-
-                    patient_type = body_item.test.comparators[0].value
-                    if patient_type in rules:
-                        for rule_node in body_item.body:
-                            if isinstance(rule_node, ast.If):
-                                conditions = ast.unparse(rule_node.test)
-                                alert = ""
-                                for item in rule_node.body:
-                                    if (
-                                        isinstance(item, ast.Expr)
-                                        and isinstance(item.value, ast.Call)
-                                        and hasattr(item.value.func, "value")
-                                        and hasattr(item.value.func.value, "id")
-                                        and item.value.func.value.id == "alerts"
-                                        and item.value.func.attr == "append"
-                                        and isinstance(item.value.args[0], ast.Constant)
-                                    ):
-                                        alert = item.value.args[0].value
-                                rules[patient_type].append(
-                                    {"conditions": conditions, "alert": alert}
-                                )
-    return rules
-
-
-def rules_to_dataframes(rules):
-    dataframes = {}
-    for patient_type, rules_list in rules.items():
-        data = {"Conditions": [], "Alert": []}
-        for rule in rules_list:
-            data["Conditions"].append(rule["conditions"])
-            data["Alert"].append(rule["alert"])
-        df = pd.DataFrame(data)
-        dataframes[patient_type] = df
-    return dataframes
-
-
-def dataframes_to_rules(dfs):
-    rules = {"Mother": [], "Neonate": [], "Gyn": []}
-    for patient_type, df in dfs.items():
-        if df is not None:
-            for index, row in df.iterrows():
-                if row["Conditions"] and row["Alert"]:
-                    rules[patient_type].append(
-                        {"conditions": row["Conditions"], "alert": row["Alert"]}
-                    )
-    return rules
-
-
-def add_row(df):
-    if df is None:
-        df = pd.DataFrame(columns=["Conditions", "Alert"])
-    df.loc[len(df)] = ["", ""]
-    return df
-
-
-def save_rules(df_mother, df_neonate, df_gyn):
-    dfs = {"Mother": df_mother, "Neonate": df_neonate, "Gyn": df_gyn}
-    for patient_type, df in dfs.items():
-        if not isinstance(df, pd.DataFrame):
-            dfs[patient_type] = pd.DataFrame(df, columns=["Conditions", "Alert"])
-
-    rules = dataframes_to_rules(dfs)
-
-    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":
-            node.body = []
-            node.body.append(ast.parse("v = state.vitals").body[0])
-            node.body.append(ast.parse("labs = state.labs").body[0])
-            node.body.append(ast.parse("alerts = []").body[0])
-
-            for patient_type, rule_list in rules.items():
-                if_patient_type_body = []
-                for rule in rule_list:
-                    conditions = (
-                        rule["conditions"].replace("\r", " ").replace("\n", " ")
-                    )
-                    if_rule_str = f"if {conditions}:\n    alerts.append({json.dumps(rule['alert'])})"
-                    if_rule = ast.parse(if_rule_str).body[0]
-                    if_patient_type_body.append(if_rule)
-
-                if if_patient_type_body:
-                    if_patient_type = ast.If(
-                        test=ast.Compare(
-                            left=ast.Attribute(
-                                value=ast.Name(id="state", ctx=ast.Load()),
-                                attr="patient_type",
-                                ctx=ast.Load(),
-                            ),
-                            ops=[ast.Eq()],
-                            comparators=[ast.Constant(value=patient_type)],
-                        ),
-                        body=if_patient_type_body,
-                        orelse=[],
-                    )
-                    node.body.append(if_patient_type)
-
-            node.body.append(
-                ast.parse(
-                    'if not alerts:\n    return "Tidak ada alert prioritas tinggi. Lanjutkan pemantauan dan dokumentasi."'
-                ).body[0]
-            )
-            node.body.append(
-                ast.parse(
-                    'return "\\n- ".join(["ALERT:"] + alerts)', mode="single"
-                ).body[0]
-            )
-
-    new_code = ast.unparse(tree)
-    with open("rules.py", "w") as f:
-        f.write(new_code)
-
-    return "Rules saved successfully."
-
-
-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}"
+from simulator import (
+    simulator_ui,
+    inject_scenario,
+    manual_edit,
+    tick_timer,
+    load_csv,
+    countdown_tick,
+    SCENARIOS,
+)
+from editor import editor_ui, save_rules
+from validator import validator_ui, test_condition, add_rule_to_set
 
 
 # --- Build UI ---
@@ -622,148 +49,24 @@ with gr.Blocks(
     interpretation = gr.Textbox(label="CDSS Interpretation", lines=2, interactive=False)
 
     with gr.Tabs():
-        with gr.TabItem("CDSS Simulator"):
-            with gr.Accordion("History, Trends, and Data Loading", open=True):
-                with gr.Row():
-                    with gr.Tabs():
-                        with gr.Tab("Blood Pressure"):
-                            bp_plot = gr.Plot()
-                        with gr.Tab("Heart Rate"):
-                            hr_plot = gr.Plot()
-                        with gr.Tab("Respiration"):
-                            rr_plot = gr.Plot()
-                        with gr.Tab("Temperature"):
-                            temp_plot = gr.Plot()
-                        with gr.Tab("SpO₂"):
-                            spo2_plot = gr.Plot()
-
-        with gr.TabItem("Rule Editor"):
-            with gr.Tabs():
-                with gr.TabItem("Edit Rules"):
-                    gr.Markdown("## CDSS Rule Editor")
-
-                    initial_rules = parse_rules()
-                    initial_dfs = rules_to_dataframes(initial_rules)
-
-                    with gr.Tabs():
-                        with gr.Tab("Mother"):
-                            df_mother = gr.DataFrame(
-                                value=initial_dfs["Mother"],
-                                headers=["Conditions", "Alert"],
-                                interactive=True,
-                                row_count=(len(initial_dfs["Mother"]) + 1, "dynamic"),
-                                type="pandas",
-                            )
-                            add_mother_btn = gr.Button("➕ Add Mother Rule")
-                            add_mother_btn.click(
-                                add_row, inputs=df_mother, outputs=df_mother
-                            )
-
-                        with gr.Tab("Neonate"):
-                            df_neonate = gr.DataFrame(
-                                value=initial_dfs["Neonate"],
-                                headers=["Conditions", "Alert"],
-                                interactive=True,
-                                row_count=(len(initial_dfs["Neonate"]) + 1, "dynamic"),
-                                type="pandas",
-                            )
-                            add_neonate_btn = gr.Button("➕ Add Neonate Rule")
-                            add_neonate_btn.click(
-                                add_row, inputs=df_neonate, outputs=df_neonate
-                            )
-
-                        with gr.Tab("Gyn"):
-                            df_gyn = gr.DataFrame(
-                                value=initial_dfs["Gyn"],
-                                headers=["Conditions", "Alert"],
-                                interactive=True,
-                                row_count=(len(initial_dfs["Gyn"]) + 1, "dynamic"),
-                                type="pandas",
-                            )
-                            add_gyn_btn = gr.Button("➕ Add Gyn Rule")
-                            add_gyn_btn.click(add_row, inputs=df_gyn, outputs=df_gyn)
-
-                    save_button = gr.Button("💾 Save Rules")
-                    status_textbox = gr.Textbox(label="Status", interactive=False)
-
-                    save_button.click(
-                        save_rules,
-                        inputs=[df_mother, df_neonate, df_gyn],
-                        outputs=status_textbox,
-                    )
-                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,
-                    )
+        bp_plot, hr_plot, rr_plot, temp_plot, spo2_plot = simulator_ui()
+        (df_mother, df_neonate, df_gyn, save_button, status_textbox) = editor_ui()
+        (
+            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,
+        ) = validator_ui()
 
     with gr.Row():
         with gr.Column(scale=2):
@@ -878,7 +181,7 @@ with gr.Blocks(
     gr.Timer(30.0).tick(tick_timer, timer_inputs, ui_outputs)
     gr.Timer(1.0).tick(countdown_tick, [last_tick_ts], [countdown_lbl])
 
-    demo.load(inject_scenario, [gr.State("A0"), cdss_toggle], ui_outputs)
+    demo.load(inject_scenario, [gr.State("A0"), cdss_toggle, history_df, historic_text], ui_outputs)
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()

editor.py

@@ -0,0 +1,202 @@
+"""
+CDSS Rule Editor Component
+"""
+
+import gradio as gr
+import pandas as pd
+import json
+import ast
+
+
+def parse_rules():
+    with open("rules.py", "r") as f:
+        tree = ast.parse(f.read())
+
+    rules = {"Mother": [], "Neonate": [], "Gyn": []}
+
+    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 isinstance(body_item.test, ast.Compare)
+                    and isinstance(body_item.test.left, ast.Attribute)
+                    and isinstance(body_item.test.left.value, ast.Name)
+                    and body_item.test.left.value.id == "state"
+                    and body_item.test.left.attr == "patient_type"
+                    and isinstance(body_item.test.ops[0], ast.Eq)
+                    and isinstance(body_item.test.comparators[0], ast.Constant)
+                ):
+
+                    patient_type = body_item.test.comparators[0].value
+                    if patient_type in rules:
+                        for rule_node in body_item.body:
+                            if isinstance(rule_node, ast.If):
+                                conditions = ast.unparse(rule_node.test)
+                                alert = ""
+                                for item in rule_node.body:
+                                    if (
+                                        isinstance(item, ast.Expr)
+                                        and isinstance(item.value, ast.Call)
+                                        and hasattr(item.value.func, "value")
+                                        and hasattr(item.value.func.value, "id")
+                                        and item.value.func.value.id == "alerts"
+                                        and item.value.func.attr == "append"
+                                        and isinstance(item.value.args[0], ast.Constant)
+                                    ):
+                                        alert = item.value.args[0].value
+                                rules[patient_type].append(
+                                    {"conditions": conditions, "alert": alert}
+                                )
+    return rules
+
+
+def rules_to_dataframes(rules):
+    dataframes = {}
+    for patient_type, rules_list in rules.items():
+        data = {"Conditions": [], "Alert": []}
+        for rule in rules_list:
+            data["Conditions"].append(rule["conditions"])
+            data["Alert"].append(rule["alert"])
+        df = pd.DataFrame(data)
+        dataframes[patient_type] = df
+    return dataframes
+
+
+def dataframes_to_rules(dfs):
+    rules = {"Mother": [], "Neonate": [], "Gyn": []}
+    for patient_type, df in dfs.items():
+        if df is not None:
+            for index, row in df.iterrows():
+                if row["Conditions"] and row["Alert"]:
+                    rules[patient_type].append(
+                        {"conditions": row["Conditions"], "alert": row["Alert"]}
+                    )
+    return rules
+
+
+def add_row(df):
+    if df is None:
+        df = pd.DataFrame(columns=["Conditions", "Alert"])
+    df.loc[len(df)] = ["", ""]
+    return df
+
+
+def save_rules(df_mother, df_neonate, df_gyn):
+    dfs = {"Mother": df_mother, "Neonate": df_neonate, "Gyn": df_gyn}
+    for patient_type, df in dfs.items():
+        if not isinstance(df, pd.DataFrame):
+            dfs[patient_type] = pd.DataFrame(df, columns=["Conditions", "Alert"])
+
+    rules = dataframes_to_rules(dfs)
+
+    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":
+            node.body = []
+            node.body.append(ast.parse("v = state.vitals").body[0])
+            node.body.append(ast.parse("labs = state.labs").body[0])
+            node.body.append(ast.parse("alerts = []").body[0])
+
+            for patient_type, rule_list in rules.items():
+                if_patient_type_body = []
+                for rule in rule_list:
+                    conditions = (
+                        rule["conditions"].replace("\r", " ").replace("\n", " ")
+                    )
+                    if_rule_str = f"if {conditions}:\n    alerts.append({json.dumps(rule['alert'])})"
+                    if_rule = ast.parse(if_rule_str).body[0]
+                    if_patient_type_body.append(if_rule)
+
+                if if_patient_type_body:
+                    if_patient_type = ast.If(
+                        test=ast.Compare(
+                            left=ast.Attribute(
+                                value=ast.Name(id="state", ctx=ast.Load()),
+                                attr="patient_type",
+                                ctx=ast.Load(),
+                            ),
+                            ops=[ast.Eq()],
+                            comparators=[ast.Constant(value=patient_type)],
+                        ),
+                        body=if_patient_type_body,
+                        orelse=[],
+                    )
+                    node.body.append(if_patient_type)
+
+            node.body.append(
+                ast.parse(
+                    'if not alerts:\\n    return "Tidak ada alert prioritas tinggi. Lanjutkan pemantauan dan dokumentasi."'
+                ).body[0]
+            )
+            node.body.append(
+                ast.parse(
+                    'return "\\n- ".join(["ALERT:"] + alerts)', mode="single"
+                ).body[0]
+            )
+
+    new_code = ast.unparse(tree)
+    with open("rules.py", "w") as f:
+        f.write(new_code)
+
+    return "Rules saved successfully."
+
+
+def editor_ui():
+    with gr.TabItem("Rule Editor"):
+        with gr.Tabs():
+            with gr.TabItem("Edit Rules"):
+                gr.Markdown("## CDSS Rule Editor")
+
+                initial_rules = parse_rules()
+                initial_dfs = rules_to_dataframes(initial_rules)
+
+                with gr.Tabs():
+                    with gr.Tab("Mother"):
+                        df_mother = gr.DataFrame(
+                            value=initial_dfs["Mother"],
+                            headers=["Conditions", "Alert"],
+                            interactive=True,
+                            row_count=(len(initial_dfs["Mother"]) + 1, "dynamic"),
+                            type="pandas",
+                        )
+                        add_mother_btn = gr.Button("➕ Add Mother Rule")
+                        add_mother_btn.click(
+                            add_row, inputs=df_mother, outputs=df_mother
+                        )
+
+                    with gr.Tab("Neonate"):
+                        df_neonate = gr.DataFrame(
+                            value=initial_dfs["Neonate"],
+                            headers=["Conditions", "Alert"],
+                            interactive=True,
+                            row_count=(len(initial_dfs["Neonate"]) + 1, "dynamic"),
+                            type="pandas",
+                        )
+                        add_neonate_btn = gr.Button("➕ Add Neonate Rule")
+                        add_neonate_btn.click(
+                            add_row, inputs=df_neonate, outputs=df_neonate
+                        )
+
+                    with gr.Tab("Gyn"):
+                        df_gyn = gr.DataFrame(
+                            value=initial_dfs["Gyn"],
+                            headers=["Conditions", "Alert"],
+                            interactive=True,
+                            row_count=(len(initial_dfs["Gyn"]) + 1, "dynamic"),
+                            type="pandas",
+                        )
+                        add_gyn_btn = gr.Button("➕ Add Gyn Rule")
+                        add_gyn_btn.click(add_row, inputs=df_gyn, outputs=df_gyn)
+
+                save_button = gr.Button("💾 Save Rules")
+                status_textbox = gr.Textbox(label="Status", interactive=False)
+
+                save_button.click(
+                    save_rules,
+                    inputs=[df_mother, df_neonate, df_gyn],
+                    outputs=status_textbox,
+                )
+    return df_mother, df_neonate, df_gyn, save_button, status_textbox

requirements.txt

@@ -1,7 +1,6 @@
 fastapi
 uvicorn
 gradio
-graphviz
 google-generativeai
 pandas
 plotly

simulator.py

@@ -0,0 +1,361 @@
+"""
+CDSS Simulator Component
+"""
+import random
+import time
+from dataclasses import asdict
+from typing import Dict, Any, Tuple, List
+from datetime import datetime
+
+import gradio as gr
+import pandas as pd
+import plotly.express as px
+
+from models import Vitals, PatientState
+from rules import rule_based_cdss
+
+# --- Gemini setup (simplified) ---
+try:
+    import google.generativeai as genai
+    import os
+    genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
+    GEMINI_MODEL = genai.GenerativeModel("gemini-1.5-pro")
+    GEMINI_ERR = None
+except Exception as e:
+    GEMINI_MODEL, GEMINI_ERR = None, f"Gemini import/config error: {e}"
+
+
+# --- Data structures & Scenarios (Full list included) ---
+
+def scenario_A0_Normal() -> PatientState:
+    return PatientState(
+        "A0 Normal Case",
+        "Mother",
+        "Pemeriksaan rutin.",
+        {"Hb": 12.5},
+        Vitals(110, 70, 80, 16, 36.7, 99),
+    )
+
+def scenario_A1_PPH() -> PatientState:
+    return PatientState(
+        "A1 PPH",
+        "Mother",
+        "30 menit postpartum; kehilangan darah ~900 ml.",
+        {"Hb": 9},
+        Vitals(90, 60, 120, 24, 36.8, 96),
+    )
+
+def scenario_A2_Preeclampsia() -> PatientState:
+    return PatientState(
+        "A2 Preeklampsia",
+        "Mother",
+        "36 minggu; sakit kepala, pandangan kabur.",
+        {"Proteinuria": "3+"},
+        Vitals(165, 105, 98, 20, 36.9, 98),
+    )
+
+def scenario_A3_MaternalSepsis() -> PatientState:
+    return PatientState(
+        "A3 Sepsis Maternal",
+        "Mother",
+        "POD2 pasca SC; luka purulen.",
+        {"Leukosit": 17000},
+        Vitals(95, 60, 110, 24, 39.0, 96),
+    )
+
+def scenario_B1_Prematurity() -> PatientState:
+    return PatientState(
+        "B1 Prematuritas/BBLR",
+        "Neonate",
+        "34 minggu; berat 1900 g; hipotermia ringan; SpO2 borderline",
+        {"BB": 1900, "UsiaGestasi_mgg": 34},
+        Vitals(60, 35, 150, 50, 35.0, 90),
+    )
+
+def scenario_B2_Asphyxia() -> PatientState:
+    return PatientState(
+        "B2 Asfiksia Perinatal",
+        "Neonate",
+        "APGAR 3 menit 1; tidak menangis >1 menit",
+        {"APGAR_1m": 3},
+        Vitals(55, 30, 80, 10, 36.5, 82),
+    )
+
+def scenario_B3_NeonatalSepsis() -> PatientState:
+    return PatientState(
+        "B3 Sepsis Neonatal",
+        "Neonate",
+        "Hari ke-4; lemas, malas minum",
+        {"CRP": 25, "Leukosit": 19000},
+        Vitals(60, 35, 170, 60, 38.5, 93),
+    )
+
+def scenario_C1_GynSurgComp() -> PatientState:
+    return PatientState(
+        "C1 Komplikasi Bedah Ginekologis",
+        "Gyn",
+        "Pasca histerektomi; nyeri perut bawah; urine output turun",
+        {"UrineOutput_ml_hr": 10},
+        Vitals(100, 65, 105, 20, 37.8, 98),
+    )
+
+def scenario_C2_PostOpInfection() -> PatientState:
+    return PatientState(
+        "C2 Infeksi Pasca-Bedah",
+        "Gyn",
+        "Pasca kistektomi; luka bengkak & kemerahan; demam",
+        {"Luka": "bengkak+kemerahan"},
+        Vitals(105, 70, 108, 22, 38.0, 98),
+    )
+
+def scenario_C3_DelayedGynCancer() -> PatientState:
+    return PatientState(
+        "C3 Keterlambatan Diagnostik Kanker Ginekologi",
+        "Gyn",
+        "45 th; perdarahan pascamenopause; Pap abnormal 6 bulan lalu tanpa tindak lanjut",
+        {"PapSmear": "abnormal 6 bln lalu"},
+        Vitals(120, 78, 86, 18, 36.8, 99),
+    )
+
+SCENARIOS = {
+    "A0": scenario_A0_Normal,
+    "A1": scenario_A1_PPH,
+    "A2": scenario_A2_Preeclampsia,
+    "A3": scenario_A3_MaternalSepsis,
+    "B1": scenario_B1_Prematurity,
+    "B2": scenario_B2_Asphyxia,
+    "B3": scenario_B3_NeonatalSepsis,
+    "C1": scenario_C1_GynSurgComp,
+    "C2": scenario_C2_PostOpInfection,
+    "C3": scenario_C3_DelayedGynCancer,
+}
+
+
+# --- Simulation & CDSS Logic (simplified) ---
+def drift_vitals(state: PatientState) -> PatientState:
+    v = state.vitals
+    clamp = lambda val, lo, hi: max(lo, min(hi, val))
+    drift_factor = 0 if state.scenario.startswith("A0") else 1
+    v.hr = clamp(v.hr + random.randint(-2, 2) * drift_factor, 40, 200)
+    v.sbp = clamp(v.sbp + random.randint(-2, 2) * drift_factor, 50, 220)
+    v.rr = clamp(v.rr + random.randint(-1, 1) * drift_factor, 8, 80)
+    state.vitals = v
+    return state
+
+
+# --- Rule-based fallback (no AI or AI disabled) ---
+
+def gemini_cdss(state: PatientState) -> str:
+    if not GEMINI_MODEL:
+        return f"[CDSS AI ERROR] {GEMINI_ERR}"
+    try:
+        v = state.vitals
+        prompt = f"CDSS for {state.scenario}. Vitals: SBP {v.sbp}/{v.dbp}, HR {v.hr}. Analyze risks, give concise steps in Indonesian."
+        return GEMINI_MODEL.generate_content(prompt).text or "[CDSS AI] No response."
+    except Exception as e:
+        return f"[CDSS AI error] {e}"
+
+
+# --- Plotting & Data Helpers ---
+def create_vital_plot(
+    df: pd.DataFrame, y_cols: List[str] | str, title: str, y_lim: List[int]
+):
+    """Creates a customized Plotly figure for a specific vital sign."""
+    # Create an empty plot if there is no data to prevent errors
+    if df.empty:
+        fig = px.line(title=title)
+    else:
+        fig = px.line(df, x="timestamp", y=y_cols, title=title, markers=True)
+        # Customize x-axis to show only first and last tick
+        if len(df) > 1:
+            fig.update_xaxes(
+                tickvals=[df["timestamp"].iloc[0], df["timestamp"].iloc[-1]]
+            )
+
+    # Apply standard layout settings
+    fig.update_layout(
+        height=250,
+        yaxis_range=y_lim,
+        margin=dict(t=40, b=10, l=10, r=10),  # Tighten margins
+    )
+    return fig
+
+def _row_from_state(ps: PatientState) -> Dict[str, Any]:
+    return {"timestamp": datetime.now(), "scenario": ps.scenario, **asdict(ps.vitals)}
+
+def prepare_df_for_display(df: pd.DataFrame) -> pd.DataFrame:
+    if df is None or df.empty:
+        return pd.DataFrame(
+            columns=[
+                "timestamp",
+                "scenario",
+                "sbp",
+                "dbp",
+                "hr",
+                "rr",
+                "temp_c",
+                "spo2",
+            ]
+        )
+    df_display = df.copy()
+    df_display["timestamp"] = pd.to_datetime(df_display["timestamp"])
+    df_display = df_display.sort_values("timestamp")
+    df_display["timestamp"] = df_display["timestamp"].dt.strftime("%Y-%m-%d %H:%M:%S")
+    return df_display
+
+def generate_all_plots(df: pd.DataFrame):
+    """Helper to generate all 5 plot figures from a dataframe."""
+    df_display = prepare_df_for_display(df)
+    bp_fig = create_vital_plot(
+        df_display,
+        y_cols=["sbp", "dbp"],
+        title="Blood Pressure (mmHg)",
+        y_lim=[40, 200],
+    )
+    hr_fig = create_vital_plot(
+        df_display, y_cols="hr", title="Heart Rate (bpm)", y_lim=[40, 200]
+    )
+    rr_fig = create_vital_plot(
+        df_display, y_cols="rr", title="Respiratory Rate (/min)", y_lim=[0, 70]
+    )
+    temp_fig = create_vital_plot(
+        df_display, y_cols="temp_c", title="Temperature (°C)", y_lim=[34, 42]
+    )
+    spo2_fig = create_vital_plot(
+        df_display, y_cols="spo2", title="Oxygen Saturation (%)", y_lim=[70, 101]
+    )
+    return df_display, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig
+
+
+# --- Gradio App Logic ---
+def process_and_update(
+    ps: PatientState, history_df: pd.DataFrame, historic_text: str, cdss_on: bool
+):
+    """Centralized function to process state, update history, and generate all UI component outputs."""
+    interpretation = gemini_cdss(ps) if cdss_on else rule_based_cdss(ps)
+    new_row = _row_from_state(ps)
+    history_df = pd.concat([history_df, pd.DataFrame([new_row])], ignore_index=True)
+
+    df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig = generate_all_plots(
+        history_df
+    )
+
+    return (
+        asdict(ps),
+        *state_to_panels(ps),
+        str(ps.labs),  # For labs_text
+        str(ps.labs),  # For labs_show
+        interpretation,
+        history_df,
+        df_for_table,
+        historic_text.strip(),
+        time.time(),
+        bp_fig,
+        hr_fig,
+        rr_fig,
+        temp_fig,
+        spo2_fig,
+    )
+
+def state_to_panels(state: PatientState) -> Tuple:
+    v = state.vitals
+    return (
+        state.scenario,
+        state.patient_type,
+        state.notes,
+        v.sbp,
+        v.dbp,
+        v.hr,
+        v.rr,
+        v.temp_c,
+        v.spo2,
+    )
+
+def inject_scenario(
+    tag: str, cdss_on: bool, history_df: pd.DataFrame, historic_text: str
+):
+    ps = SCENARIOS[tag]()
+    if historic_text:  # Add a newline if text already exists
+        historic_text += f"\n[{datetime.now().strftime('%H:%M:%S')}] Scenario Injected: {ps.scenario}"
+    else:
+        historic_text = (
+            f"[{datetime.now().strftime('%H:%M:%S')}] Scenario Injected: {ps.scenario}"
+        )
+    return process_and_update(ps, history_df, historic_text, cdss_on)
+
+def manual_edit(
+    sbp,
+    dbp,
+    hr,
+    rr,
+    temp_c,
+    spo2,
+    notes,
+    labs_text,
+    cdss_on,
+    patient_type,
+    current_state,
+    history_df,
+    historic_text,
+):
+    try:
+        labs = eval(labs_text)
+    except:
+        labs = {"raw": labs_text}
+    ps = PatientState(
+        current_state.get("scenario", "Manual"),
+        patient_type,
+        notes,
+        labs,
+        Vitals(int(sbp), int(dbp), int(hr), int(rr), float(temp_c), int(spo2)),
+    )
+    if ps.notes and ps.notes.strip():
+        historic_text += f"\n[{datetime.now().strftime('%H:%M:%S')}] {ps.notes}"
+    return process_and_update(ps, history_df, historic_text, cdss_on)
+
+def tick_timer(cdss_on, current_state, history_df, historic_text):
+    if not current_state:
+        return [gr.update()] * 22
+    ps = PatientState(**current_state)
+    ps.vitals = Vitals(**ps.vitals)
+    ps = drift_vitals(ps)
+    return process_and_update(ps, history_df, historic_text, cdss_on)
+
+def load_csv(file, history_df: pd.DataFrame):
+    try:
+        if file is not None:
+            df_new = pd.read_csv(file.name)
+            df_new["timestamp"] = pd.to_datetime(df_new["timestamp"])
+            history_df = (
+                pd.concat([history_df, df_new], ignore_index=True)
+                if not history_df.empty
+                else df_new
+            )
+    except Exception as e:
+        print(f"Error loading CSV: {e}")
+    df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig = generate_all_plots(
+        history_df
+    )
+    return history_df, df_for_table, bp_fig, hr_fig, rr_fig, temp_fig, spo2_fig
+
+def countdown_tick(last_tick_ts: float):
+    if not last_tick_ts:
+        return "Next update in —"
+    return f"Next update in {max(0, 30 - int(time.time() - last_tick_ts))}s"
+
+def simulator_ui():
+    with gr.TabItem("CDSS Simulator"):
+        with gr.Accordion("History, Trends, and Data Loading", open=True):
+            with gr.Row():
+                with gr.Tabs():
+                    with gr.Tab("Blood Pressure"):
+                        bp_plot = gr.Plot()
+                    with gr.Tab("Heart Rate"):
+                        hr_plot = gr.Plot()
+                    with gr.Tab("Respiration"):
+                        rr_plot = gr.Plot()
+                    with gr.Tab("Temperature"):
+                        temp_plot = gr.Plot()
+                    with gr.Tab("SpO₂"):
+                        spo2_plot = gr.Plot()
+    return bp_plot, hr_plot, rr_plot, temp_plot, spo2_plot

validator.py

@@ -0,0 +1,167 @@
+"""
+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