from __future__ import annotations import base64 import io import os import warnings from pathlib import Path import cv2 import numpy as np from fastapi import FastAPI, File, Form, Header, HTTPException, UploadFile from fastapi.responses import HTMLResponse from fastapi.middleware.cors import CORSMiddleware from huggingface_hub import hf_hub_download os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "3") warnings.filterwarnings("ignore") import tensorflow as tf tf.get_logger().setLevel("ERROR") HF_REPO = "jvrfer/medical-ai-model" IMAGE_SIZE = (384, 384) THRESHOLD = 0.35 GRADCAM_LAYER = "top_conv" from model import RandomBrightness from dataset import load_inference_tensors_from_bytes from gradcam import make_gradcam_plus_plus_heatmap, overlay_heatmap from utils import load_json API_KEY = os.environ.get("MEDICAL_AI_API_KEY") if not API_KEY: raise RuntimeError("MEDICAL_AI_API_KEY no configurada") print("Descargando modelo...") model_path = hf_hub_download(repo_id=HF_REPO, filename="best_model_phase2.keras") class_names_path = hf_hub_download(repo_id=HF_REPO, filename="class_names.json") CUSTOM_OBJECTS = {"RandomBrightness": RandomBrightness} print("Cargando modelo...") MODEL = tf.keras.models.load_model(model_path, custom_objects=CUSTOM_OBJECTS) CLASS_NAMES = load_json(class_names_path)["class_names"] print(f"Modelo cargado. Clases: {CLASS_NAMES}") LEGAL_WARNING = ( "Advertencia: esta salida es solo apoyo informatico y no constituye diagnostico medico. " "Debe ser interpretada por personal de salud calificado." ) app = FastAPI(title="Medical AI API", version="1.0.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) def validate_key(api_key: str | None): if not api_key: raise HTTPException(401, "Falta header X-API-Key") if api_key != API_KEY: raise HTTPException(403, "API key invalida") def predict_bytes(filename: str, payload: bytes) -> dict: _, processed = load_inference_tensors_from_bytes(filename, payload, image_size=IMAGE_SIZE) batch = np.expand_dims(processed, axis=0) probs = MODEL.predict(batch, verbose=0)[0] results = {cls: float(p) for cls, p in zip(CLASS_NAMES, probs)} top_class = max(results, key=results.get) return {"diagnosis": top_class, "confidence": results[top_class], "all_probabilities": results} @app.get("/health") def health(): return {"status": "healthy"} @app.post("/predict") async def predict(file: UploadFile = File(...), x_api_key: str | None = Header(default=None)): validate_key(x_api_key) payload = await file.read() if not payload: raise HTTPException(400, "Archivo vacio") try: result = predict_bytes(file.filename or "image", payload) except Exception as e: raise HTTPException(400, f"Error: {e}") return {**result, "threshold_used": THRESHOLD, "legal_warning": LEGAL_WARNING} @app.post("/predict-with-gradcam") async def predict_gradcam(file: UploadFile = File(...), x_api_key: str | None = Header(default=None)): validate_key(x_api_key) payload = await file.read() if not payload: raise HTTPException(400, "Archivo vacio") try: original, processed = load_inference_tensors_from_bytes( file.filename or "image", payload, image_size=IMAGE_SIZE ) except Exception as e: raise HTTPException(400, f"Error: {e}") batch = np.expand_dims(processed, axis=0) probs = MODEL.predict(batch, verbose=0)[0] results = {cls: float(p) for cls, p in zip(CLASS_NAMES, probs)} top_class = max(results, key=results.get) gradcam_b64 = None try: heatmap = make_gradcam_plus_plus_heatmap(MODEL, batch, layer_name=GRADCAM_LAYER) overlay, _ = overlay_heatmap(original, heatmap) _, buf = cv2.imencode(".png", cv2.cvtColor(overlay, cv2.COLOR_RGB2BGR)) gradcam_b64 = base64.b64encode(buf).decode() except Exception: pass return { "diagnosis": top_class, "confidence": results[top_class], "all_probabilities": results, "gradcam_b64": gradcam_b64, "threshold_used": THRESHOLD, "legal_warning": LEGAL_WARNING, } WEB_HTML = """ Medical AI

Medical AI

Analisis de radiografias de torax

Arrastra una imagen aqui o haz clic para seleccionar

JPG, PNG, WEBP, BMP, DICOM

Procesando imagen...

Mapa de calor GradCAM++

""" @app.get("/", response_class=HTMLResponse) def web_ui(): return HTMLResponse(WEB_HTML.replace("{api_key}", API_KEY if API_KEY else ""))