Update app.py

app.py

@@ -1,7 +1,275 @@
+import os
+import io
+import json
+import math
+import numpy as np
+import pandas as pd
 import gradio as gr
+from PIL import Image
 
-def greet(name):
-    return "Hello " + name + "!!"
+# Imaging (Chest X-ray) — TorchXRayVision
+import torch
+import torch.nn.functional as F
+import torchxrayvision as xrv
+from torchvision import transforms
+
+# ECG / Signals — HeartPy
+import heartpy as hp
+
+# Optional DICOM support
+try:
+    import pydicom
+    HAS_PYDICOM = True
+except Exception:
+    HAS_PYDICOM = False
+
+# -----------------------------
+# Imaging Agent (Chest X-ray)
+# -----------------------------
+_IMAGING_DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+# Pretrained DenseNet on multiple datasets (free)
+_IMAGING_MODEL = xrv.models.DenseNet(weights="densenet121-res224-all").to(_IMAGING_DEVICE)
+_IMAGING_MODEL.eval()
+_PATHOLOGY_LIST = _IMAGING_MODEL.pathologies  # list of labels
+
+# Preprocess pipeline for CXRs
+_img_resize = transforms.Compose([
+    xrv.datasets.XRayCenterCrop(),
+    xrv.datasets.XRayResizer(224)
+])
+
+def _load_cxr_image(filepath: str) -> Image.Image:
+    # Accept: DICOM, jpg, png
+    ext = os.path.splitext(filepath)[1].lower()
+    if ext in [".dcm", ".dicom"] and HAS_PYDICOM:
+        ds = pydicom.dcmread(filepath)
+        arr = ds.pixel_array.astype(np.float32)
+        # Normalize and to PIL
+        arr = arr - arr.min()
+        if arr.max() > 0:
+            arr = arr / arr.max()
+        arr = (arr * 255.0).clip(0, 255).astype(np.uint8)
+        return Image.fromarray(arr).convert("L")
+    else:
+        return Image.open(filepath).convert("L")
+
+def imaging_agent(image_path: str):
+    if not image_path:
+        return "No image provided.", None
+
+    try:
+        img = _load_cxr_image(image_path)
+        # to numpy [1,1,H,W] normalized
+        arr = np.array(img).astype(np.float32)
+        if arr.max() > 1.0:
+            arr /= 255.0
+        arr = xrv.datasets.normalize(arr, 4096)  # safe normalize
+        arr = _img_resize(arr)
+        arr = torch.from_numpy(arr).unsqueeze(0).unsqueeze(0).to(_IMAGING_DEVICE)
+
+        with torch.no_grad():
+            preds = _IMAGING_MODEL(arr)[0]  # shape [num_labels]
+            probs = torch.sigmoid(preds).detach().cpu().numpy().tolist()
+
+        # Top 5 findings
+        top = sorted(zip(_PATHOLOGY_LIST, probs), key=lambda x: x[1], reverse=True)[:5]
+        lines = [f"{name}: {p*100:.1f}%" for name, p in top]
+        table = {name: round(p, 4) for name, p in zip(_PATHOLOGY_LIST, probs)}
+        return "🖼️ Imaging Agent (Chest X-ray)\n" + "\n".join(lines), json.dumps(table, indent=2)
+    except Exception as e:
+        return f"Imaging agent error: {e}", None
+
+# -----------------------------
+# Signal Agent (ECG)
+# -----------------------------
+def _read_signal(file_obj) -> np.ndarray:
+    # Expect CSV/TSV or plain text with one column
+    try:
+        file_obj.seek(0)
+        df = pd.read_csv(file_obj)
+    except Exception:
+        file_obj.seek(0)
+        try:
+            df = pd.read_csv(file_obj, sep="\t")
+        except Exception:
+            file_obj.seek(0)
+            # Plain text: one value per line
+            vals = [float(x.strip()) for x in file_obj.read().decode("utf-8").splitlines() if x.strip()]
+            return np.asarray(vals, dtype=np.float32)
+    # Prefer first numeric column
+    for col in df.columns:
+        if pd.api.types.is_numeric_dtype(df[col]):
+            return df[col].dropna().to_numpy(dtype=np.float32)
+    # Fallback: try to coerce first column
+    return pd.to_numeric(df.iloc[:, 0], errors="coerce").dropna().to_numpy(dtype=np.float32)
+
+def signal_agent(file, sample_rate_hz: float):
+    if file is None:
+        return "No signal file uploaded."
+    if not sample_rate_hz or sample_rate_hz <= 0:
+        return "Please provide a valid sampling rate (Hz)."
+
+    try:
+        sig = _read_signal(file)
+        # Basic safety checks
+        if len(sig) < sample_rate_hz * 5:
+            return "Signal too short. Provide at least 5 seconds of data."
+        # HeartPy processing
+        wd, m = hp.process(sig, sample_rate=sample_rate_hz)
+        bpm = m.get('bpm', float('nan'))
+        rmssd = m.get('rmssd', float('nan'))
+        ibi = m.get('ibi', float('nan'))
+        sdnn = m.get('sdnn', float('nan'))
+
+        # Simple flags (not medical diagnoses)
+        flags = []
+        if not math.isnan(bpm):
+            if bpm < 50: flags.append("Bradycardia risk (low BPM)")
+            if bpm > 110: flags.append("Tachycardia risk (high BPM)")
+        if not math.isnan(sdnn) and sdnn > 100:
+            flags.append("High variability — possible irregular rhythm")
+        if not math.isnan(rmssd) and rmssd > 80:
+            flags.append("Elevated RMSSD — irregularity suspicion")
+
+        summary = [
+            "💓 Signal Agent (ECG-like biosignal)",
+            f"BPM: {bpm:.1f}" if not math.isnan(bpm) else "BPM: N/A",
+            f"SDNN: {sdnn:.1f} ms" if not math.isnan(sdnn) else "SDNN: N/A",
+            f"RMSSD: {rmssd:.1f} ms" if not math.isnan(rmssd) else "RMSSD: N/A",
+            f"IBI: {ibi:.1f} ms" if not math.isnan(ibi) else "IBI: N/A",
+            "Flags: " + (", ".join(flags) if flags else "none")
+        ]
+        return "\n".join(summary)
+    except Exception as e:
+        return f"Signal agent error: {e}"
+
+# -----------------------------
+# Lab Agent (text parsing)
+# -----------------------------
+# Simple thresholds (illustrative only; NOT medical advice)
+_LAB_THRESHOLDS = {
+    "glucose": {"unit": "mg/dL", "high": 126, "low": 70},
+    "hemoglobin": {"unit": "g/dL", "low": 12.0},
+    "spo2": {"unit": "%", "low": 92},
+    "ldl": {"unit": "mg/dL", "high": 160},
+    "hdl": {"unit": "mg/dL", "low": 40},
+    "triglycerides": {"unit": "mg/dL", "high": 200"},
+    "creatinine": {"unit": "mg/dL", "high": 1.3},
+}
+
+import re
+def _extract_labs(text: str):
+    # match patterns like "glucose: 180 mg/dL" or "glucose 180"
+    results = {}
+    for line in text.splitlines():
+        line_l = line.lower()
+        matches = re.findall(r'([a-z\%\d\/]+)\s*[:=]?\s*([\-]?\d+\.?\d*)\s*([a-z%\/]+)?', line_l)
+        for label, val, unit in matches:
+            label = label.strip().replace("%", "spo2") if label.strip() == "o2" else label.strip()
+            if label in _LAB_THRESHOLDS:
+                try:
+                    v = float(val)
+                    results[label] = {"value": v, "unit": unit or _LAB_THRESHOLDS[label]["unit"]}
+                except:
+                    pass
+    return results
+
+def lab_agent(lab_text: str):
+    if not lab_text or not lab_text.strip():
+        return "No lab text provided."
+    labs = _extract_labs(lab_text)
+    if not labs:
+        return "Could not parse lab values. Use lines like 'glucose: 180 mg/dL'."
+
+    lines = ["🧪 Lab Agent"]
+    flags = []
+    for k, v in labs.items():
+        value = v["value"]
+        unit = v["unit"]
+        thr = _LAB_THRESHOLDS.get(k, {})
+        status = "ok"
+        if "high" in thr and value > thr["high"]:
+            status = "high"
+            flags.append(f"{k} high")
+        if "low" in thr and value < thr["low"]:
+            status = "low"
+            flags.append(f"{k} low")
+        lines.append(f"{k.capitalize()}: {value} {unit} → {status}")
+    if flags:
+        lines.append("Flags: " + ", ".join(flags))
+    else:
+        lines.append("Flags: none")
+    return "\n".join(lines)
+
+# -----------------------------
+# Coordinator
+# -----------------------------
+def coordinator(imaging_txt: str, signal_txt: str, lab_txt: str):
+    parts = []
+    if imaging_txt and "Imaging Agent" in imaging_txt:
+        parts.append(imaging_txt)
+    if signal_txt and "Signal Agent" in signal_txt:
+        parts.append(signal_txt)
+    if lab_txt and "Lab Agent" in lab_txt:
+        parts.append(lab_txt)
+
+    assessment = []
+    if imaging_txt and ("pneumonia" in imaging_txt.lower() or "infiltration" in imaging_txt.lower()):
+        assessment.append("Possible pulmonary involvement")
+    if signal_txt and ("tachycardia" in signal_txt.lower() or "irregular" in signal_txt.lower()):
+        assessment.append("Cardiac rhythm irregularity risk")
+    if lab_txt and ("glucose" in lab_txt.lower() and "high" in lab_txt.lower()):
+        assessment.append("Hyperglycemia risk")
+
+    summary = "📋 Coordinator Summary\n"
+    summary += "\n\n".join(parts) if parts else "No agent outputs."
+    if assessment:
+        summary += "\n\n🧭 Integrated Assessment: " + "; ".join(assessment)
+    summary += "\n\n⚠️ Disclaimer: This demo is not a medical device. Do not use for diagnosis. Consult a qualified clinician."
+    return summary
+
+# -----------------------------
+# Gradio UI
+# -----------------------------
+with gr.Blocks(theme="soft") as demo:
+    gr.Markdown("# 🏥 AI Diagnostics Agents (Demo)")
+    gr.Markdown(
+        "Upload a chest X-ray (PNG/JPG/DICOM), an ECG-like signal (CSV/TSV or text with one value per line), "
+        "and/or paste lab values. Each agent analyzes its modality; the coordinator fuses results.\n\n"
+        "⚠️ **Not for clinical use.**"
+    )
+
+    with gr.Row():
+        with gr.Column():
+            img_in = gr.Image(type="filepath", label="Chest X-ray (PNG/JPG or DICOM)")
+            imaging_out = gr.Textbox(label="Imaging Agent (Top findings)")
+            imaging_raw = gr.Code(label="Imaging Probabilities (JSON)", language="json")
+        with gr.Column():
+            ecg_in = gr.File(label="ECG / Biosignal (CSV/TSV or txt)")
+            sr = gr.Number(label="Sampling Rate (Hz)", value=250, precision=1)
+            signal_out = gr.Textbox(label="Signal Agent Summary")
+        with gr.Column():
+            lab_in = gr.Textbox(lines=12, label="Lab Results (e.g., 'glucose: 180 mg/dL')")
+            lab_out = gr.Textbox(label="Lab Agent Summary")
+
+    run_btn = gr.Button("Run All Agents")
+    coord_out = gr.Textbox(label="Coordinator Summary", lines=14)
+
+    def run_imaging(image_path):
+        txt, raw = imaging_agent(image_path) if image_path else ("No image provided.", None)
+        return txt, raw
+
+    def run_signal(file, rate):
+        return signal_agent(file, rate)
+
+    def run_lab(text):
+        return lab_agent(text)
+
+    run_btn.click(run_imaging, inputs=img_in, outputs=[imaging_out, imaging_raw])\
+           .then(run_signal, inputs=[ecg_in, sr], outputs=signal_out)\
+           .then(run_lab, inputs=lab_in, outputs=lab_out)\
+           .then(coordinator, inputs=[imaging_out, signal_out, lab_out], outputs=coord_out)
 
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
 demo.launch()
+