save new rule

.gitignore

@@ -0,0 +1 @@
+__pycache__

cdss.py

@@ -380,6 +380,107 @@ def countdown_tick(last_tick_ts: float):
     return f"Next update in {max(0, 30 - int(time.time() - last_tick_ts))}s"
 
 
+import json
+import ast
+
+import graphviz
+
+def parse_rules(return_json=False):
+    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 hasattr(body_item.test, 'left') and hasattr(body_item.test.left, 'id') and body_item.test.left.id == 'state':
+                    patient_type = body_item.test.comparators[0].s
+                    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'):
+                                        alert = item.value.args[0].s
+                                rules[patient_type].append({"conditions": conditions, "alert": alert})
+    
+    if return_json:
+        return rules
+
+    dot = graphviz.Digraph(comment='Rule Based CDSS')
+    dot.attr(rankdir='TB', splines='ortho')
+
+    for patient_type, rule_list in rules.items():
+        with dot.subgraph(name=f'cluster_{patient_type}') as c:
+            c.attr(label=patient_type, style='filled', color='lightgrey')
+            for i, rule in enumerate(rule_list):
+                c.node(f'{patient_type}_{i}_cond', rule['conditions'], shape='box')
+                c.node(f'{patient_type}_{i}_alert', rule['alert'], shape='plaintext')
+                c.edge(f'{patient_type}_{i}_cond', f'{patient_type}_{i}_alert')
+
+    return dot
+
+def save_rules(rules_json):
+    rules = json.loads(rules_json)
+    
+    # Read the original file
+    with open("rules.py", "r") as f:
+        tree = ast.parse(f.read())
+
+    # Find the function definition
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == "rule_based_cdss":
+            # Clear the existing body
+            node.body = []
+
+            # Add the initial variable assignments
+            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])
+
+            # Add the rules
+            for patient_type, rule_list in rules.items():
+                if_patient_type_body = []
+                for rule in rule_list:
+                    if_rule = ast.parse(f"if {rule['conditions']}:\n    alerts.append(\"{rule['alert']}\")").body[0]
+                    if_patient_type_body.append(if_rule)
+
+                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)
+
+            # Add the return statement
+            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)\"").body[0])
+
+    # Unparse the modified AST
+    new_code = ast.unparse(tree)
+
+    # Write the new code back to the file
+    with open("rules.py", "w") as f:
+        f.write(new_code)
+
+    return "Rules saved successfully."
+
+def generate_and_display_rules():
+    dot = parse_rules()
+    dot.render('rules_visualization', format='svg', cleanup=True)
+    return 'rules_visualization.svg'
+
 # --- Build UI ---
 with gr.Blocks(
     css=".gradio-container { max-width: 1200px !important; margin: auto !important; }"
@@ -389,21 +490,84 @@ with gr.Blocks(
     history_df = gr.State(pd.DataFrame())
     historic_text = gr.State("")
     last_tick_ts = gr.State(time.time())
-
     interpretation = gr.Textbox(label="CDSS Interpretation", lines=2, interactive=False)
-    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.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.Row():
+                with gr.Column(scale=2):
+                    with gr.Accordion("Inject New Scenario (resets chart)", open=True):
+                        with gr.Row():
+                            btn_A0 = gr.Button("A0: Normal", elem_classes="small-btn")
+                            btn_A1 = gr.Button("A1: PPH", elem_classes="small-btn")
+                            btn_A2 = gr.Button("A2: Preeklampsia", elem_classes="small-btn")
+                            btn_A3 = gr.Button("A3: Sepsis", elem_classes="small-btn")
+                        with gr.Row():
+                            btn_B1 = gr.Button("B1: Prematuritas", elem_classes="small-btn")
+                            btn_B2 = gr.Button("B2: Asfiksia", elem_classes="small-btn")
+                            btn_B3 = gr.Button("B3: Sepsis", elem_classes="small-btn")
+                        with gr.Row():
+                            btn_C1 = gr.Button("C1: Bedah Komplikasi", elem_classes="small-btn")
+                            btn_C2 = gr.Button(
+                                "C2: Infeksi Pasca-Bedah", elem_classes="small-btn"
+                            )
+                            btn_C3 = gr.Button("C3: Kanker", elem_classes="small-btn")
+                    notes = gr.Textbox(label="Catatan Klinis", lines=2)
+                    labs_text = gr.Textbox(label="Lab (dict or text)", value="{}")
+                    labs_show = gr.Textbox(label="Labs (Parsed)", interactive=False)
+                with gr.Column(scale=1):
+                    with gr.Group():
+                        patient_type_radio = gr.Radio(
+                            ["Mother", "Neonate", "Gyn"], label="Patient Type", value="Mother"
+                        )
+                        sbp = gr.Number(label="SBP")
+                        dbp = gr.Number(label="DBP")
+                        hr = gr.Number(label="HR")
+                        rr = gr.Number(label="RR")
+                        temp_c = gr.Number(label="Temp (°C)")
+                        spo2 = gr.Number(label="SpO₂ (%)")
+                    apply_manual = gr.Button("Apply Manual Edits", variant="secondary")
+            with gr.Row():
+                csv_file = gr.File(label="Load CSV to Graph")
+                df_view = gr.Dataframe(label="History Table", wrap=True, interactive=False)
+                cdss_toggle = gr.Checkbox(value=False, label="With CDSS (Gemini)")
+                scenario_lbl = gr.Textbox(label="Active Scenario", interactive=False)
+                countdown_lbl = gr.Label()
+                historic_box = gr.Textbox(label="Historic Text", lines=12, interactive=False)
+
+        with gr.TabItem("Rule Visualizer"):
+            gr.Markdown("## CDSS Rule Visualizer and Editor")
+    
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("### Current Rules Visualization")
+                    rules_display = gr.Image(interactive=False)
+                with gr.Column():
+                    gr.Markdown("### Edit Rules (JSON)")
+                    rules_editor = gr.Textbox(lines=20, label="Edit Rules in JSON format")
+                    save_button = gr.Button("Save Rules")
+
+            def update_editor_and_rules():
+                return generate_and_display_rules(), json.dumps(parse_rules(return_json=True), indent=2)
+
+            demo.load(update_editor_and_rules, None, [rules_display, rules_editor])
+            save_button.click(save_rules, inputs=rules_editor, outputs=gr.Textbox()).then(
+                update_editor_and_rules, None, [rules_display, rules_editor]
+            )
+
 
     with gr.Row():
         with gr.Column(scale=2):

requirements.txt

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

rules.py

@@ -1,84 +1,30 @@
 from models import PatientState
 
-
 def rule_based_cdss(state: PatientState) -> str:
     v = state.vitals
     labs = state.labs
     alerts = []
-
-    if state.patient_type == "Mother":
-        # PPH-like
-        if (
-            v.sbp < 100
-            and v.hr > 110
-            and ("Hb" in labs and labs["Hb"] <= 9 or state.scenario.startswith("A1"))
-        ):
-            alerts.append(
-                "PPH suspected: lakukan uterotonik, massage uterus, siapkan transfusi, monitoring 5 menit."
-            )
-        # Preeclampsia
-        if v.sbp >= 160 or v.dbp >= 110 or state.scenario.startswith("A2"):
-            if labs.get("Proteinuria") in (
-                "2+",
-                "3+",
-                "4+",
-            ) or state.scenario.startswith("A2"):
-                alerts.append(
-                    "Preeklampsia berat: MgSO4 loading + maintenance, kontrol TD, pertimbangkan terminasi ≥34 minggu."
-                )
-        # Maternal sepsis
-        if v.temp_c >= 38.5 and v.sbp <= 100 or state.scenario.startswith("A3"):
-            alerts.append(
-                "Sepsis maternal: kultur darah & luka, antibiotik spektrum luas dalam 1 jam, monitor MAP & urine."
-            )
-
-    if state.patient_type == "Neonate":
-        # Prematurity/BBLR
-        if (
-            labs.get("BB", 9999) < 2500
-            or labs.get("UsiaGestasi_mgg", 99) < 37
-            or state.scenario.startswith("B1")
-        ):
-            if v.temp_c < 36.5 or v.spo2 < 92:
-                alerts.append(
-                    "Prematur/BBLR: inkubator/skin-to-skin, monitor SpO2 & suhu, pertimbangkan CPAP jika SpO2<92%."
-                )
-        # Asphyxia
-        if v.spo2 < 90 or v.hr < 100 or state.scenario.startswith("B2"):
-            alerts.append(
-                "Asfiksia: resusitasi neonatal sesuai NRP, ventilasi tekanan positif, evaluasi setiap 30 detik."
-            )
-        # Neonatal sepsis
-        if (
-            v.temp_c >= 38.0
-            and labs.get("CRP", 0) >= 10
-            or state.scenario.startswith("B3")
-        ):
-            alerts.append(
-                "Sepsis neonatal: kultur darah, antibiotik empiris NICU, monitor tanda vital ketat."
-            )
-
-    if state.patient_type == "Gyn":
-        # Post-op complication (ureter)
-        if state.scenario.startswith("C1") or labs.get("UrineOutput_ml_hr", 9999) < 20:
-            alerts.append(
-                "Komplikasi pasca bedah (curiga ureter): evaluasi segera, USG/CT urografi, konsult urologi."
-            )
-        # Post-op infection
-        if v.temp_c >= 38.0 and (
-            state.scenario.startswith("C2") or labs.get("Luka") == "bengkak+kemerahan"
-        ):
-            alerts.append(
-                "Infeksi pasca-bedah: kultur luka, antibiotik, pertimbangkan debridement."
-            )
-        # Delayed cancer follow-up
-        if state.scenario.startswith("C3") or labs.get("PapSmear", "").startswith(
-            "abnormal"
-        ):
-            alerts.append(
-                "Pap smear abnormal tanpa follow-up: lakukan kolposkopi & biopsi, aktifkan notifikasi rujukan."
-            )
-
+    if state.patient_type == 'Mother':
+        if v.sbp < 100 and v.hr > 110 and ('Hb' in labs and labs['Hb'] <= 9 or state.scenario.startswith('A1')):
+            alerts.append('PPH suspected: lakukan uterotonik, massage uterus, siapkan transfusi, monitoring 5 menit.')
+        if v.sbp >= 160 or v.dbp >= 110 or state.scenario.startswith('A2'):
+            alerts.append('no alert')
+        if v.temp_c >= 38.5 and v.sbp <= 100 or state.scenario.startswith('A3'):
+            alerts.append('Sepsis maternal: kultur darah & luka, antibiotik spektrum luas dalam 1 jam, monitor MAP & urine.')
+    if state.patient_type == 'Neonate':
+        if labs.get('BB', 9999) < 2500 or labs.get('UsiaGestasi_mgg', 99) < 37 or state.scenario.startswith('B1'):
+            alerts.append('Prematurity/BBLR')
+        if v.spo2 < 90 or v.hr < 100 or state.scenario.startswith('B2'):
+            alerts.append('Asfiksia: resusitasi neonatal sesuai NRP, ventilasi tekanan positif, evaluasi setiap 30 detik.')
+        if v.temp_c >= 38.0 and labs.get('CRP', 0) >= 10 or state.scenario.startswith('B3'):
+            alerts.append('Sepsis neonatal: kultur darah, antibiotik empiris NICU, monitor tanda vital ketat.')
+    if state.patient_type == 'Gyn':
+        if state.scenario.startswith('C1') or labs.get('UrineOutput_ml_hr', 9999) < 20:
+            alerts.append('Komplikasi pasca bedah (curiga ureter): evaluasi segera, USG/CT urografi, konsult urologi.')
+        if v.temp_c >= 38.0 and (state.scenario.startswith('C2') or labs.get('Luka') == 'bengkak+kemerahan'):
+            alerts.append('Infeksi pasca-bedah: kultur luka, antibiotik, pertimbangkan debridement.')
+        if state.scenario.startswith('C3') or labs.get('PapSmear', '').startswith('abnormal'):
+            alerts.append('Pap smear abnormal tanpa follow-up: lakukan kolposkopi & biopsi, aktifkan notifikasi rujukan.')
     if not alerts:
-        return "Tidak ada alert prioritas tinggi. Lanjutkan pemantauan dan dokumentasi."
-    return "\n- ".join(["ALERT:"] + alerts)
+        return 'Tidak ada alert prioritas tinggi. Lanjutkan pemantauan dan dokumentasi.'
+    return '\n- '.join(['ALERT:'] + alerts)

visualizer.py

@@ -0,0 +1,184 @@
+import gradio as gr
+import json
+import ast
+import pandas as pd
+
+
+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_dataframes(df_mother, df_neonate, df_gyn):
+    dfs = {"Mother": df_mother, "Neonate": df_neonate, "Gyn": df_gyn}
+    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."
+
+
+with gr.Blocks() as demo:
+    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"),
+            )
+            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"),
+            )
+            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"),
+            )
+            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_dataframes, inputs=[df_mother, df_neonate, df_gyn], outputs=status_textbox
+    )
+
+if __name__ == "__main__":
+    demo.launch()