Create app.py

app.py

@@ -0,0 +1,94 @@
+# app.py
+import io
+import faiss
+import torch
+import pandas as pd
+
+from PIL import Image
+from fastapi import FastAPI, File, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from huggingface_hub import hf_hub_download
+from transformers import CLIPProcessor, CLIPModel
+
+# ---------- FastAPI app ----------
+app = FastAPI()
+
+# Allow React frontend to call this API
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],      # later you can restrict to your frontend domain
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ---------- Load index + metadata + model at startup ----------
+
+REPO_ID = "saad003/Red01"  # your dataset repo
+
+print("Downloading FAISS index & metadata...")
+INDEX_PATH = hf_hub_download(
+    repo_id=REPO_ID,
+    filename="radiology_index.faiss",
+    repo_type="dataset",
+)
+META_PATH = hf_hub_download(
+    repo_id=REPO_ID,
+    filename="radiology_metadata.csv",
+    repo_type="dataset",
+)
+
+print("Loading FAISS index...")
+index = faiss.read_index(INDEX_PATH)
+
+print("Loading metadata CSV...")
+metadata = pd.read_csv(META_PATH)
+
+print("Loading CLIP model...")
+MODEL_NAME = "flaviagiammarino/pubmed-clip-vit-base-patch32"
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+clip_model = CLIPModel.from_pretrained(MODEL_NAME).to(device)
+clip_processor = CLIPProcessor.from_pretrained(MODEL_NAME)
+clip_model.eval()
+
+print("Backend ready ✅")
+
+
+# ---------- Helper: search by image ----------
+
+def _search_similar_by_image(image: Image.Image, k: int = 5):
+    inputs = clip_processor(images=image, return_tensors="pt").to(device)
+    with torch.no_grad():
+        feats = clip_model.get_image_features(**inputs)
+
+    feats = feats / feats.norm(p=2, dim=-1, keepdim=True)
+    feats = feats.cpu().numpy().astype("float32")
+
+    D, I = index.search(feats, k)
+
+    rows = metadata.iloc[I[0]].copy()
+    rows["score"] = D[0]
+    # Only send useful columns
+    return rows[["ID", "split", "img_path", "caption", "concepts_manual", "score"]]
+
+
+# ---------- API endpoint ----------
+
+@app.post("/search_by_image")
+async def search_by_image(file: UploadFile = File(...), k: int = 5):
+    # read image from request
+    content = await file.read()
+    image = Image.open(io.BytesIO(content)).convert("RGB")
+
+    results_df = _search_similar_by_image(image, k=k)
+    results = results_df.to_dict(orient="records")
+
+    return JSONResponse({"results": results})
+
+
+@app.get("/")
+def root():
+    return {"status": "ok", "message": "Radiology retrieval API"}