| | from fastapi import FastAPI, UploadFile, File, HTTPException, Request,Query |
| | import io |
| | from PIL import Image |
| | from io import BytesIO |
| | from pydantic import BaseModel |
| | from typing import Union |
| | from io import BytesIO |
| | import base64 |
| | import logging |
| | import logging |
| | from app.model import load_model, predict_with_model |
| | |
| | logging.basicConfig( |
| | level=logging.DEBUG, |
| | format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" |
| | ) |
| |
|
| | logger = logging.getLogger(__name__) |
| |
|
| |
|
| |
|
| | app = FastAPI() |
| |
|
| |
|
| | @app.on_event("startup") |
| | def load_models_once(): |
| | _ = load_model() |
| |
|
| | class ImagePayload(BaseModel): |
| | image: str |
| | @app.post("/predict") |
| | async def predict(request: Request, |
| | file: UploadFile = File(None), |
| | payload: Union[ImagePayload, None] = None, |
| | show_heatmap: bool = Query(False, description="Afficher la heatmap"), |
| | ): |
| |
|
| | logger.info("🔁 Requête reçue") |
| | logger.info(f"✅ Show heatmap : {show_heatmap}") |
| | |
| | try: |
| | |
| | if file is not None: |
| | image_bytes = await file.read() |
| | logger.debug("✅ Image reçue via multipart :", file.filename, len(image_bytes), "octets") |
| |
|
| | |
| | elif await request.json(): |
| | body = await request.json() |
| | if "image" not in body: |
| | raise HTTPException(status_code=422, detail="Champ 'image' manquant.") |
| | image_base64 = body["image"] |
| | image_bytes = base64.b64decode(image_base64) |
| | logger.debug("✅ Image décodée depuis base64 :", len(image_bytes), "octets") |
| |
|
| | else: |
| | logger.info("⚠️ Aucune image reçue") |
| | raise HTTPException(status_code=400, detail="Format de requête non supporté.") |
| |
|
| | |
| | logger.debug("🔍 Appel du vote multi-modèles...") |
| | model_config = load_model()[0] |
| | prediction = predict_with_model(model_config, image_bytes, show_heatmap) |
| |
|
| | |
| | return prediction |
| |
|
| | except Exception as e: |
| | logger.error("❌ Une erreur s'est produite", exc_info=True) |
| | raise HTTPException(status_code=500, detail=str(e)) |
| |
|
| |
|
| | @app.get("/health") |
| | def health_check(): |
| | return {"status": "ok"} |
| |
|