Create app.py

app.py

@@ -0,0 +1,181 @@
+import os
+import io
+import base64
+import json
+import datetime
+import importlib.util
+from typing import List, Optional
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse, FileResponse, HTMLResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+from PIL import Image
+
+
+def _load_rmmm_module():
+    """Dynamically load the project's RMMM module so we can reuse model
+    loading and inference code from `XRaySwinGen-RMMM/app.py`.
+    """
+    root = os.path.dirname(__file__)
+    module_path = os.path.normpath(os.path.join(root, "XRaySwinGen-RMMM", "app.py"))
+    spec = importlib.util.spec_from_file_location("rmmm_module", module_path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+# Load module at startup (this will run top-level initialization in that file)
+RMMM = _load_rmmm_module()
+
+
+app = FastAPI(title="XRaySwinGen - RMMM API for Static Frontend")
+
+# Serve the static frontend contained in XRaySwinGen-RMMM/frontend
+frontend_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), "XRaySwinGen-RMMM", "frontend"))
+app.mount("/", StaticFiles(directory=frontend_dir, html=True), name="frontend")
+
+
+class ReportRequest(BaseModel):
+    image_base64: Optional[str] = None
+    filename: Optional[str] = None
+    source_id: Optional[str] = None
+
+
+def pil_to_data_url(img: Image.Image, fmt: str = "PNG") -> str:
+    buf = io.BytesIO()
+    img.save(buf, format=fmt)
+    b = base64.b64encode(buf.getvalue()).decode("ascii")
+    return f"data:image/{fmt.lower()};base64,{b}"
+
+
+@app.get("/api/v1/samples")
+async def get_samples(limit: int = 24):
+    """Return a list of sample images (base64) found in XRaySwinGen-RMMM/images"""
+    images_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), "XRaySwinGen-RMMM", "images"))
+    items = []
+    if not os.path.isdir(images_dir):
+        return JSONResponse({"items": []})
+
+    files = [f for f in os.listdir(images_dir) if f.lower().endswith((".jpg", ".jpeg", ".png"))]
+    files = sorted(files)[:limit]
+    for fname in files:
+        path = os.path.join(images_dir, fname)
+        try:
+            with Image.open(path) as im:
+                data_url = pil_to_data_url(im.convert("RGB"), fmt="PNG")
+            item = {
+                "id": os.path.splitext(fname)[0],
+                "title": os.path.splitext(fname)[0],
+                "image_base64": data_url,
+            }
+            items.append(item)
+        except Exception:
+            continue
+
+    return JSONResponse({"items": items})
+
+
+@app.post("/api/v1/report")
+async def create_report(req: ReportRequest):
+    """Generate a report from an uploaded image or a sample. The frontend
+    expects a JSON response containing fields used by the static UI.
+    """
+    try:
+        # Determine image source
+        pil_img = None
+        selected_image_path = ""
+
+        if req.source_id:
+            # Try to find a matching file in images/ by id
+            images_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), "XRaySwinGen-RMMM", "images"))
+            candidates = [f for f in os.listdir(images_dir) if os.path.splitext(f)[0] == req.source_id]
+            if candidates:
+                selected_image_path = os.path.join(images_dir, candidates[0])
+                pil_img = Image.open(selected_image_path).convert("RGB")
+
+        if req.image_base64 and not pil_img:
+            # image_base64 may be a data URL
+            data = req.image_base64
+            if "," in data:
+                data = data.split(",", 1)[1]
+            raw = base64.b64decode(data)
+            pil_img = Image.open(io.BytesIO(raw)).convert("RGB")
+
+        if pil_img is None:
+            return JSONResponse({"error": "No valid image provided"}, status_code=400)
+
+        # Run inference using functions from the loaded module
+        ai_report = None
+        ground_truth = ""
+        metrics_html = ""
+
+        if hasattr(RMMM, "inference_image_pipe_with_state"):
+            ai_report, ground_truth, metrics_html = RMMM.inference_image_pipe_with_state(pil_img, selected_image_path)
+        else:
+            ai_report = RMMM.inference_torch_model_fast(pil_img)
+            if hasattr(RMMM, "get_ground_truth_from_filename"):
+                ground_truth = RMMM.get_ground_truth_from_filename(selected_image_path)
+
+        # Annotated image
+        annotated_pil = None
+        if hasattr(RMMM, "annotate_image"):
+            try:
+                annotated_pil = RMMM.annotate_image(pil_img.copy(), ai_report or "")
+            except Exception:
+                annotated_pil = pil_img
+        else:
+            annotated_pil = pil_img
+
+        # Explanation
+        explanation = ""
+        detailed = ""
+        step_html = ""
+        if hasattr(RMMM, "explain_findings"):
+            try:
+                explanation, detailed, step_html = RMMM.explain_findings(ai_report or "", ground_truth or "")
+            except Exception:
+                explanation = ""
+
+        # Compute numeric metrics if possible
+        numeric_metrics = None
+        if hasattr(RMMM, "calculate_evaluation_metrics") and ground_truth:
+            try:
+                numeric_metrics = RMMM.calculate_evaluation_metrics(ai_report or "", ground_truth or "")
+                numeric_metrics = {k: v for k, v in numeric_metrics.items() if k in ("bleu4_score", "rougeL_f")}
+            except Exception:
+                numeric_metrics = None
+
+        # Build small findings/insights from explanation (best-effort)
+        insights = []
+        findings = {}
+        if explanation:
+            for line in explanation.split("\n")[:6]:
+                if line.strip():
+                    insights.append(line.strip())
+
+        # Convert annotated image to data URL
+        annotated_data_url = pil_to_data_url(annotated_pil, fmt="PNG")
+
+        now = datetime.datetime.utcnow().isoformat() + "Z"
+
+        response = {
+            "created_at": now,
+            "report_text": ai_report or "",
+            "annotated_image": annotated_data_url,
+            "insights": insights,
+            "findings": findings,
+            "metrics": numeric_metrics,
+            "status_chain": [
+                {"title": "Uploaded", "detail": req.filename or req.source_id or "uploaded image"},
+                {"title": "Preprocessing", "detail": "Resizing, normalization"},
+                {"title": "Inference", "detail": "RMMM model inference executed"},
+                {"title": "Postprocessing", "detail": "Decoding and annotations"},
+            ],
+            "source_id": req.source_id or os.path.splitext(req.filename or "")[0],
+        }
+
+        return JSONResponse(response)
+
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)