api with gradido

app.py

@@ -11,6 +11,11 @@ from model import CombinedModel, ImageToTextProjector
 import pydicom
 import os
 import gc
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from typing import List
+import base64
+from fastapi.responses import JSONResponse
 
 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
@@ -42,26 +47,38 @@ image_transform = transforms.Compose([
 ])
 
 def dicom_to_image(file_bytes):
+    """Convert DICOM file bytes to PIL Image"""
     dicom_file = pydicom.dcmread(io.BytesIO(file_bytes))
     pixel_array = dicom_file.pixel_array.astype(np.float32)
     pixel_array = ((pixel_array - pixel_array.min()) / pixel_array.ptp()) * 255.0
     pixel_array = pixel_array.astype(np.uint8)
     return Image.fromarray(pixel_array).convert("RGB")
 
-def predict(files):
-    if not files:
+def process_images(file_data_list):
+    """Core image processing logic used by both Gradio and FastAPI"""
+    if not file_data_list:
         return "No images uploaded.", ""
 
     processed_imgs = []
-    for file_obj in files:
-        filename = file_obj.name.lower()
-        if filename.endswith((".dcm", ".ima")):
-            file_bytes = file_obj.read()
-            img = dicom_to_image(file_bytes)
-        else:
-            img = Image.open(file_obj).convert("RGB")
-        processed_imgs.append(img)
+    
+    for file_data in file_data_list:
+        filename = file_data.get('filename', '').lower()
+        file_content = file_data.get('content')
+        
+        try:
+            if filename.endswith((".dcm", ".ima")):
+                img = dicom_to_image(file_content)
+            else:
+                img = Image.open(io.BytesIO(file_content)).convert("RGB")
+            processed_imgs.append(img)
+        except Exception as e:
+            print(f"Error processing file {filename}: {e}")
+            continue
+
+    if not processed_imgs:
+        return "No valid images processed.", ""
 
+    # Sample frames for video model
     n_frames = 16
     if len(processed_imgs) >= n_frames:
         images_sampled = [
@@ -71,28 +88,180 @@ def predict(files):
     else:
         images_sampled = processed_imgs + [processed_imgs[-1]] * (n_frames - len(processed_imgs))
 
-    tensor_imgs = [image_transform(i) for i in images_sampled]
+    # Transform images to tensors
+    tensor_imgs = [image_transform(img) for img in images_sampled]
     input_tensor = torch.stack(tensor_imgs).permute(1, 0, 2, 3).unsqueeze(0).to(device)
 
+    # Model inference
     with torch.no_grad():
         class_logits, report, _ = combined_model(input_tensor)
         class_pred = torch.argmax(class_logits, dim=1).item()
         class_name = class_names[class_pred]
 
+    # Cleanup
     gc.collect()
     if torch.cuda.is_available():
         torch.cuda.empty_cache()
 
     return class_name, report[0] if report else "No report generated."
 
-# Gradio Blocks (100% reliable approach)
-# Replace your Blocks interface with this simpler Interface approach
+def predict_gradio(files):
+    """Gradio interface wrapper"""
+    if not files:
+        return "No images uploaded.", ""
+
+    file_data_list = []
+    for file_obj in files:
+        try:
+            file_content = file_obj.read() if hasattr(file_obj, 'read') else open(file_obj.name, 'rb').read()
+            file_data_list.append({
+                'filename': file_obj.name if hasattr(file_obj, 'name') else str(file_obj),
+                'content': file_content
+            })
+        except Exception as e:
+            print(f"Error reading file: {e}")
+            continue
+
+    return process_images(file_data_list)
+
+# Create FastAPI app
+app = FastAPI(
+    title="Phronesis ML API",
+    description="Medical Image Analysis API with Gradio Interface",
+    version="1.0.0"
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    """Root endpoint"""
+    return {
+        "message": "Phronesis ML API",
+        "status": "running",
+        "endpoints": {
+            "predict": "/predict",
+            "health": "/health",
+            "gradio": "/gradio"
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {
+        "status": "healthy",
+        "model_loaded": True,
+        "device": str(device)
+    }
+
+@app.post("/predict")
+async def predict_api(files: List[UploadFile] = File(...)):
+    """
+    API endpoint for medical image prediction
+    
+    Args:
+        files: List of uploaded image files (DICOM, JPG, PNG, etc.)
+    
+    Returns:
+        JSON response with predicted class and generated report
+    """
+    try:
+        if not files:
+            raise HTTPException(status_code=400, detail="No files uploaded")
+        
+        # Process uploaded files
+        file_data_list = []
+        for file in files:
+            try:
+                content = await file.read()
+                file_data_list.append({
+                    'filename': file.filename or 'unknown',
+                    'content': content
+                })
+            except Exception as e:
+                print(f"Error reading uploaded file {file.filename}: {e}")
+                continue
+        
+        if not file_data_list:
+            raise HTTPException(status_code=400, detail="No valid files processed")
+        
+        # Get predictions
+        predicted_class, generated_report = process_images(file_data_list)
+        
+        # Return results
+        return JSONResponse(content={
+            "status": "success",
+            "data": {
+                "predicted_class": predicted_class,
+                "generated_report": generated_report,
+                "processed_files": len(file_data_list)
+            }
+        })
+    
+    except HTTPException:
+        raise
+    except Exception as e:
+        print(f"Prediction error: {e}")
+        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request, exc):
+    """Global exception handler"""
+    return JSONResponse(
+        status_code=500,
+        content={
+            "status": "error",
+            "message": "Internal server error",
+            "detail": str(exc)
+        }
+    )
+
+# Create Gradio interface
 demo = gr.Interface(
-    fn=predict,
-    inputs=gr.File(file_count="multiple", file_types=[".dcm", ".jpg", ".jpeg", ".png", ".bmp"]),
-    outputs=[gr.Textbox(label="Predicted Class"), gr.Textbox(label="Generated Report")],
+    fn=predict_gradio,
+    inputs=gr.File(
+        file_count="multiple",
+        file_types=[".dcm", ".ima", ".jpg", ".jpeg", ".png", ".bmp"],
+        label="Upload Medical Images"
+    ),
+    outputs=[
+        gr.Textbox(label="Predicted Class"),
+        gr.Textbox(label="Generated Report", lines=5)
+    ],
     title="🩺 Phronesis Medical Report Generator",
-    description="Upload CT scan images to generate a medical report"
+    description="""
+    Upload CT scan images to generate a medical report and classification.
+    
+    **Supported formats:** DICOM (.dcm, .ima), JPEG, PNG, BMP
+    
+    **API Endpoint:** `/predict` (POST)
+    """,
+    examples=[],
+    allow_flagging="never"
 )
 
-demo.launch(share=True)
+# Mount Gradio app to FastAPI
+app = gr.mount_gradio_app(app, demo, path="/gradio")
+
+# Launch configuration
+if __name__ == "__main__":
+    import uvicorn
+    
+    # For local development
+    # uvicorn.run(app, host="0.0.0.0", port=7860)
+    
+    # For Hugging Face Spaces
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=True,
+        show_error=True
+    )

requirements.txt

@@ -14,3 +14,8 @@ scikit-learn==1.3.0
 tqdm==4.66.1
 sentencepiece==0.1.99
 pydicom==2.4.1
+
+
+
+uvicorn[standard]
+python-multipart