Create app.py

app.py

@@ -0,0 +1,285 @@
+import os
+import random
+import datetime
+import pandas as pd
+import gradio as gr
+
+ROENTGEN_DIR = "roentgen"
+PEDISYNTH_DIR = "pedisynth"
+RESULTS_DIR = "results"
+
+
+def load_pairs():
+    """Load matching image pairs from both folders, keyed by filename."""
+    ro_files = {f for f in os.listdir(ROENTGEN_DIR)
+                if os.path.isfile(os.path.join(ROENTGEN_DIR, f))}
+    pe_files = {f for f in os.listdir(PEDISYNTH_DIR)
+                if os.path.isfile(os.path.join(PEDISYNTH_DIR, f))}
+
+    common = sorted(ro_files & pe_files)
+    pairs = []
+
+    for fname in common:
+        base, _ext = os.path.splitext(fname)
+        parts = base.split()
+        # "{condition} {number}" -> condition is everything except last token
+        condition = " ".join(parts[:-1]) if len(parts) > 1 else base
+
+        pairs.append(
+            {
+                "filename": fname,
+                "condition": condition,
+                "roentgen_path": os.path.join(ROENTGEN_DIR, fname),
+                "pedisynth_path": os.path.join(PEDISYNTH_DIR, fname),
+            }
+        )
+    return pairs
+
+
+ALL_PAIRS = load_pairs()
+os.makedirs(RESULTS_DIR, exist_ok=True)
+
+
+def start_session(doctor_id):
+    """Initialize a new randomized session for this doctor."""
+    if not ALL_PAIRS:
+        status = "No matching image pairs found. Check your roentgen/ and pedisynth/ folders."
+        return (
+            None,  # img_a
+            None,  # img_b
+            "",    # condition text
+            "",    # progress text
+            None,  # state
+            None,  # results file
+            status,
+            gr.update(value=None),  # clear choice
+        )
+
+    if not doctor_id:
+        doctor_id = "anonymous"
+
+    # Randomize order of pairs
+    pairs = ALL_PAIRS.copy()
+    random.shuffle(pairs)
+
+    # Randomize which model appears on left (A) vs right (B)
+    session_pairs = []
+    for p in pairs:
+        if random.random() < 0.5:
+            left_model = "roentgen"
+            right_model = "pedisynth"
+            left_path = p["roentgen_path"]
+            right_path = p["pedisynth_path"]
+        else:
+            left_model = "pedisynth"
+            right_model = "roentgen"
+            left_path = p["pedisynth_path"]
+            right_path = p["roentgen_path"]
+
+        session_pairs.append(
+            {
+                **p,
+                "left_model": left_model,
+                "right_model": right_model,
+                "left_path": left_path,
+                "right_path": right_path,
+            }
+        )
+
+    state = {
+        "doctor_id": doctor_id,
+        "pairs": session_pairs,
+        "idx": 0,
+        "records": [],
+    }
+
+    first = session_pairs[0]
+    condition_text = f"**Condition:** {first['condition']}"
+    progress_text = f"Pair 1 of {len(session_pairs)}"
+    status = f"Session started for doctor: {doctor_id}"
+
+    return (
+        first["left_path"],
+        first["right_path"],
+        condition_text,
+        progress_text,
+        state,
+        None,  # no CSV yet
+        status,
+        gr.update(value=None),  # clear choice
+    )
+
+
+def next_pair(choice, state):
+    """Record the doctor's choice and move to the next pair."""
+    # If no active session
+    if state is None or not state.get("pairs"):
+        status = "No active session. Please start a session first."
+        return (
+            None,
+            None,
+            "",
+            "",
+            state,
+            None,
+            status,
+            gr.update(value=None),
+        )
+
+    idx = state["idx"]
+    pairs = state["pairs"]
+
+    # Ensure choice is made
+    if not choice:
+        status = "Please select A or B before proceeding."
+        cur_pair = pairs[idx]
+        condition_text = f"**Condition:** {cur_pair['condition']}"
+        progress_text = f"Pair {idx + 1} of {len(pairs)}"
+        return (
+            cur_pair["left_path"],
+            cur_pair["right_path"],
+            condition_text,
+            progress_text,
+            state,
+            None,
+            status,
+            gr.update(),  # keep current choice
+        )
+
+    cur_pair = pairs[idx]
+
+    chosen_side = choice
+    chosen_model = cur_pair["left_model"] if choice == "A" else cur_pair["right_model"]
+
+    record = {
+        "doctor_id": state["doctor_id"],
+        "timestamp_utc": datetime.datetime.utcnow().isoformat(),
+        "pair_index": idx,
+        "filename": cur_pair["filename"],
+        "condition": cur_pair["condition"],
+        "left_model": cur_pair["left_model"],
+        "right_model": cur_pair["right_model"],
+        "left_image_path": cur_pair["left_path"],
+        "right_image_path": cur_pair["right_path"],
+        "chosen_side": chosen_side,
+        "chosen_model": chosen_model,
+    }
+    state["records"].append(record)
+
+    # Advance index
+    state["idx"] += 1
+    idx = state["idx"]
+
+    # If we've reached the end, save CSV
+    if idx >= len(pairs):
+        df = pd.DataFrame(state["records"])
+        ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+        safe_doctor_id = "".join(
+            c for c in state["doctor_id"] if c.isalnum() or c in ("_", "-")
+        )
+        csv_name = f"session_{safe_doctor_id}_{ts}.csv"
+        csv_path = os.path.join(RESULTS_DIR, csv_name)
+        df.to_csv(csv_path, index=False)
+
+        status = (
+            f"Session complete. Saved {len(state['records'])} comparisons "
+            f"to `{csv_name}`."
+        )
+
+        # Optionally, reset session index or leave state as-is for audit
+        return (
+            None,             # img_a
+            None,             # img_b
+            "Session finished.",  # condition text
+            "",               # progress
+            state,
+            csv_path,         # file for download
+            status,
+            gr.update(value=None),
+        )
+
+    # Otherwise, show next pair
+    next_p = pairs[idx]
+    condition_text = f"**Condition:** {next_p['condition']}"
+    progress_text = f"Pair {idx + 1} of {len(pairs)}"
+    status = f"Recorded choice for pair {idx}."
+
+    return (
+        next_p["left_path"],
+        next_p["right_path"],
+        condition_text,
+        progress_text,
+        state,
+        None,
+        status,
+        gr.update(value=None),  # clear choice
+    )
+
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Synthetic Chest X-ray A/B Evaluation")
+
+    with gr.Row():
+        doctor_id = gr.Textbox(
+            label="Doctor ID (or initials)",
+            placeholder="e.g., dr_smith",
+        )
+        start_button = gr.Button("Start New Session", variant="primary")
+
+    state = gr.State()
+
+    with gr.Row():
+        condition_text = gr.Markdown("")
+        progress_text = gr.Markdown("")
+
+    with gr.Row():
+        img_a = gr.Image(label="Image A", interactive=False)
+        img_b = gr.Image(label="Image B", interactive=False)
+
+    choice = gr.Radio(
+        choices=["A", "B"],
+        label="Which image better matches the stated condition?",
+        interactive=True,
+    )
+
+    next_button = gr.Button("Next")
+
+    results_file = gr.File(
+        label="Download session CSV",
+        interactive=False,
+    )
+    status_text = gr.Markdown("")
+
+    # Wire callbacks
+    start_button.click(
+        fn=start_session,
+        inputs=doctor_id,
+        outputs=[
+            img_a,
+            img_b,
+            condition_text,
+            progress_text,
+            state,
+            results_file,
+            status_text,
+            choice,
+        ],
+    )
+
+    next_button.click(
+        fn=next_pair,
+        inputs=[choice, state],
+        outputs=[
+            img_a,
+            img_b,
+            condition_text,
+            progress_text,
+            state,
+            results_file,
+            status_text,
+            choice,
+        ],
+    )
+
+if __name__ == "__main__":
+    demo.launch()