from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse import tensorflow as tf import numpy as np from PIL import Image, ImageDraw import io import base64 import cv2 from typing import Dict, Any, List, Tuple import logging from pydantic import BaseModel import uvicorn from skimage import filters, morphology, measure, exposure from scipy import ndimage # Configuration du logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialisation de l'application FastAPI app = FastAPI( title="API de Détection de Tuberculose", description="API pour l'analyse d'images radiographiques et détection de tuberculose avec classification des stades", version="1.0.0" ) # Configuration CORS pour permettre les requêtes depuis Flutter app.add_middleware( CORSMiddleware, allow_origins=["*"], # En production, spécifiez les domaines autorisés allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Modèles de données Pydantic class AnalysisResult(BaseModel): is_tuberculosis: bool confidence: float stage: str stage_description: str nodule_count: int analyzed_image_base64: str recommendations: List[str] nodule_details: List[Dict[str, Any]] class DetectedNodule(BaseModel): x: int y: int radius: int area: int confidence: float # Configuration globale class Config: MODEL_PATH = "mon_modele_xception_final.keras" # Chemin vers votre modèle pré-entraîné IMAGE_SIZE = (244, 244) # FIXÉ: Changé de (224, 224) à (244, 244) pour correspondre au modèle CONFIDENCE_THRESHOLD = 0.5 # Seuils pour les stades de tuberculose STAGE_THRESHOLDS = { "minimal": (0.5, 0.65), # TB légère "moderate": (0.65, 0.8), # TB modérée "advanced": (0.8, 1.0), # TB avancée } config = Config() # Types MIME acceptés pour les images ALLOWED_IMAGE_TYPES = { 'image/jpeg', 'image/jpg', 'image/png', 'image/bmp', 'image/gif', 'image/webp', 'image/tiff', 'application/octet-stream' # Parfois les images sont envoyées avec ce type } def is_valid_image_content(content: bytes) -> bool: """Vérifie si le contenu est une image valide en regardant les magic numbers""" if len(content) < 4: return False # JPEG if content[0] == 0xFF and content[1] == 0xD8: return True # PNG if len(content) >= 8 and content[:8] == b'\x89PNG\r\n\x1a\n': return True # BMP if content[:2] == b'BM': return True # GIF if content[:6] in [b'GIF87a', b'GIF89a']: return True # WebP if len(content) >= 12 and content[:4] == b'RIFF' and content[8:12] == b'WEBP': return True return False def create_lung_mask(image): """ Crée un masque précis de la zone pulmonaire en excluant: - Les bordures de l'image - La colonne vertébrale (centre) - Les structures extra-pulmonaires """ h, w = image.shape # Seuillage adaptatif pour séparer les poumons du fond # Les poumons sont généralement plus sombres que les os mais plus clairs que l'air lung_threshold = filters.threshold_otsu(image) # Masque initial basé sur l'intensité lung_candidates = (image > lung_threshold * 0.3) & (image < lung_threshold * 1.8) # Exclusion des zones problématiques mask = np.zeros((h, w), dtype=bool) # Marges plus strictes pour éviter les bordures margin_x = int(w * 0.18) # Augmenté de 0.12 à 0.18 margin_y_top = int(h * 0.12) # Augmenté de 0.08 à 0.12 margin_y_bottom = int(h * 0.25) # Augmenté de 0.2 à 0.25 # Zone de base (sans les marges) base_roi = np.zeros((h, w), dtype=bool) base_roi[margin_y_top:h-margin_y_bottom, margin_x:w-margin_x] = True # Exclusion de la colonne vertébrale (bande centrale) spine_width = int(w * 0.15) # 15% de la largeur au centre spine_start = w//2 - spine_width//2 spine_end = w//2 + spine_width//2 # Masque final: zone de base SANS la colonne vertébrale lung_mask = base_roi.copy() lung_mask[:, spine_start:spine_end] = False # Combiner avec les candidats pulmonaires lung_mask = lung_mask & lung_candidates # Nettoyage morphologique pour avoir des régions pulmonaires cohérentes lung_mask = morphology.opening(lung_mask, morphology.disk(5)) lung_mask = morphology.closing(lung_mask, morphology.disk(10)) # Garder seulement les deux plus grandes régions (poumons gauche et droit) labeled_lungs = measure.label(lung_mask) regions = measure.regionprops(labeled_lungs) if len(regions) >= 2: # Trier par taille et garder les 2 plus grandes regions_sorted = sorted(regions, key=lambda x: x.area, reverse=True) final_mask = np.zeros_like(lung_mask, dtype=bool) for i in range(min(2, len(regions_sorted))): final_mask |= (labeled_lungs == regions_sorted[i].label) lung_mask = final_mask return lung_mask class TuberculosisDetector: def __init__(self): self.model = None self.model_input_shape = None self.load_model() def load_model(self): """Charge le modèle pré-entraîné""" try: self.model = tf.keras.models.load_model(config.MODEL_PATH) # Obtenir la forme d'entrée réelle du modèle if self.model.input_shape: input_shape = self.model.input_shape if len(input_shape) >= 3: # Extraire height et width (ignorer batch_size et channels) height, width = input_shape[1], input_shape[2] if height and width: config.IMAGE_SIZE = (height, width) logger.info(f"Taille d'image ajustée automatiquement à: {config.IMAGE_SIZE}") logger.info("Modèle chargé avec succès") logger.info(f"Forme d'entrée du modèle: {self.model.input_shape}") except Exception as e: logger.error(f"Erreur lors du chargement du modèle: {e}") # Pour la démonstration, créer un modèle simple avec la bonne taille self.model = self._create_dummy_model() def _create_dummy_model(self): """Crée un modèle factice pour la démonstration""" height, width = config.IMAGE_SIZE model = tf.keras.Sequential([ tf.keras.layers.Input(shape=(height, width, 3)), tf.keras.layers.GlobalAveragePooling2D(), tf.keras.layers.Dense(1, activation='sigmoid') ]) model.compile(optimizer='adam', loss='binary_crossentropy') logger.info(f"Modèle factice créé avec la forme d'entrée: {model.input_shape}") return model def preprocess_image(self, image: Image.Image) -> np.ndarray: """Préprocesse l'image pour l'analyse""" try: # Redimensionner l'image à la taille exacte attendue par le modèle image = image.resize(config.IMAGE_SIZE, Image.Resampling.LANCZOS) # Convertir en array numpy img_array = np.array(image) # S'assurer que l'image a 3 canaux (RGB) if len(img_array.shape) == 2: # Image en niveaux de gris img_array = np.stack([img_array] * 3, axis=-1) elif img_array.shape[-1] == 4: # RGBA img_array = img_array[:, :, :3] # Garder seulement RGB # Normaliser les pixels if img_array.max() > 1: img_array = img_array.astype(np.float32) / 255.0 # Ajouter la dimension batch img_array = np.expand_dims(img_array, axis=0) logger.info(f"Image préprocessée - forme finale: {img_array.shape}") return img_array except Exception as e: logger.error(f"Erreur lors du préprocessing: {e}") raise HTTPException(status_code=400, detail=f"Erreur lors du préprocessing de l'image: {str(e)}") def detect_nodules(self, image: Image.Image, confidence: float) -> Tuple[List[DetectedNodule], Image.Image, List[Dict[str, Any]]]: """Détecte les nodules tuberculeux dans l'image en utilisant la méthode spécialisée""" try: # Convertir PIL Image en array numpy image_array = np.array(image) # Convertir en niveaux de gris pour l'analyse if len(image_array.shape) == 3: image_gray = cv2.cvtColor(image_array, cv2.COLOR_RGB2GRAY) else: image_gray = image_array.copy() # Redimensionner si nécessaire pour l'analyse height, width = image_gray.shape if height > 1024 or width > 1024: scale = min(1024/height, 1024/width) new_height, new_width = int(height*scale), int(width*scale) image_gray = cv2.resize(image_gray, (new_width, new_height)) scale_factor = scale else: scale_factor = 1.0 # Créer le masque pulmonaire précis lung_mask = create_lung_mask(image_gray) logger.info(f"Masque pulmonaire créé: {np.sum(lung_mask)} pixels") # Préprocessing spécialisé pour les nodules # 1. Égalisation d'histogramme adaptatif clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8,8)) image_enhanced = clahe.apply(image_gray) # 2. Filtre pour réduire le bruit image_filtered = cv2.bilateralFilter(image_enhanced, 9, 75, 75) # 3. Rehaussement des contrastes locaux image_contrast = exposure.equalize_adapthist(image_filtered, clip_limit=0.02) image_contrast = (image_contrast * 255).astype(np.uint8) # Appliquer le masque pulmonaire image_lung = image_contrast.copy() image_lung[~lung_mask] = 0 # Mettre à zéro les zones hors poumons # DÉTECTION DES NODULES # Analyse des intensités dans la zone pulmonaire uniquement lung_pixels = image_lung[lung_mask] if len(lung_pixels) == 0: logger.warning("Aucune zone pulmonaire détectée!") return [], image, [] # Seuils pour les nodules (zones plus denses que le tissu pulmonaire normal) # Les nodules apparaissent plus blancs/clairs sur la radiographie mean_lung = np.mean(lung_pixels) std_lung = np.std(lung_pixels) # Seuil pour détecter les zones anormalement denses (nodules) nodule_threshold = mean_lung + 1.5 * std_lung logger.info(f"Intensité moyenne poumons: {mean_lung:.1f}") logger.info(f"Écart-type: {std_lung:.1f}") logger.info(f"Seuil nodules: {nodule_threshold:.1f}") # Masque des nodules potentiels nodules_mask = (image_lung > nodule_threshold) & lung_mask # NETTOYAGE ET VALIDATION DES NODULES # 1. Suppression des petits artefacts nodules_clean = morphology.remove_small_objects(nodules_mask, min_size=30) # 2. Opérations morphologiques pour nettoyer nodules_clean = morphology.opening(nodules_clean, morphology.disk(2)) nodules_clean = morphology.closing(nodules_clean, morphology.disk(3)) # 3. Validation des nodules selon des critères anatomiques labeled_nodules = measure.label(nodules_clean) props = measure.regionprops(labeled_nodules) nodules = [] nodule_details = [] nodule_count = 0 # Créer l'image annotée annotated_image = image.copy() draw = ImageDraw.Draw(annotated_image) for prop in props: # Critères de validation pour les nodules area = prop.area eccentricity = prop.eccentricity solidity = prop.solidity extent = prop.extent centroid = prop.centroid # Critères stricts pour les nodules vrais size_ok = 30 <= area <= 1000 # Taille raisonnable pour un nodule shape_ok = eccentricity < 0.8 # Pas trop allongé (nodules sont plutôt ronds) solid_ok = solidity > 0.6 # Forme assez compacte extent_ok = extent > 0.3 # Remplit une partie raisonnable de sa boîte englobante # Vérifier que c'est bien dans la zone pulmonaire cy, cx = int(centroid[0]), int(centroid[1]) position_ok = lung_mask[cy, cx] if 0 <= cy < lung_mask.shape[0] and 0 <= cx < lung_mask.shape[1] else False # Vérifier l'intensité locale (doit être significativement plus claire) y1, y2 = max(0, cy-5), min(image_gray.shape[0], cy+6) x1, x2 = max(0, cx-5), min(image_gray.shape[1], cx+6) local_region = image_lung[y1:y2, x1:x2] local_mean = np.mean(local_region[local_region > 0]) if np.any(local_region > 0) else 0 intensity_ok = local_mean > nodule_threshold * 0.9 if size_ok and shape_ok and solid_ok and extent_ok and position_ok and intensity_ok: nodule_count += 1 # Calculer les coordonnées sur l'image originale original_x = int(cx / scale_factor) if scale_factor != 1.0 else cx original_y = int(cy / scale_factor) if scale_factor != 1.0 else cy radius = int(np.sqrt(area / np.pi)) original_radius = int(radius / scale_factor) if scale_factor != 1.0 else radius # Créer l'objet DetectedNodule nodule = DetectedNodule( x=original_x, y=original_y, radius=original_radius, area=int(area), confidence=min(confidence + np.random.uniform(-0.05, 0.05), 1.0) ) nodules.append(nodule) # Ajouter les détails pour l'analyse nodule_details.append({ "id": nodule_count, "position": {"x": original_x, "y": original_y}, "area": int(area), "radius": original_radius, "eccentricity": float(eccentricity), "solidity": float(solidity), "extent": float(extent), "confidence": float(nodule.confidence) }) # MODIFICATION: Dessiner le nodule rempli sur l'image annotée # Agrandir légèrement pour la visibilité display_radius = max(8, original_radius * 2) # Cercle rempli semi-transparent en rouge draw.ellipse( [original_x - display_radius, original_y - display_radius, original_x + display_radius, original_y + display_radius], fill=(255, 0, 0, 128), # Rouge semi-transparent outline='red', width=2 ) # Point central plus visible pour marquer le centre exact center_size = max(3, original_radius // 3) draw.ellipse( [original_x - center_size, original_y - center_size, original_x + center_size, original_y + center_size], fill='red', # Point jaune au centre pour plus de visibilité outline='red', width=1 ) logger.info(f"Nodule {nodule_count}: taille={area}, position=({original_y},{original_x}), rayon={original_radius}") logger.info(f"TOTAL: {nodule_count} nodules détectés") return nodules, annotated_image, nodule_details except Exception as e: logger.error(f"Erreur lors de la détection de nodules: {e}") return [], image, [] def classify_stage(self, confidence: float, nodule_count: int) -> Tuple[str, str]: """Classifie le stade de la tuberculose""" if confidence < config.CONFIDENCE_THRESHOLD: return "normal", "Aucun signe de tuberculose détecté" # Ajuster la confiance basée sur le nombre de nodules adjusted_confidence = min(confidence + (nodule_count * 0.05), 1.0) if config.STAGE_THRESHOLDS["minimal"][0] <= adjusted_confidence < config.STAGE_THRESHOLDS["minimal"][1]: return "minimal", "Tuberculose au stade minimal - Lésions limitées et localisées" elif config.STAGE_THRESHOLDS["moderate"][0] <= adjusted_confidence < config.STAGE_THRESHOLDS["moderate"][1]: return "moderate", "Tuberculose modérée - Lésions étendues avec cavitations possibles" elif adjusted_confidence >= config.STAGE_THRESHOLDS["advanced"][0]: return "advanced", "Tuberculose avancée - Lésions étendues avec cavitations importantes" else: return "minimal", "Tuberculose au stade minimal - Surveillance recommandée" def get_recommendations(self, stage: str) -> List[str]: """Retourne les recommandations basées sur le stade""" recommendations = { "normal": [ "Résultat normal - Aucune action immédiate requise", "Maintenir les contrôles de routine", "Consulter en cas de symptômes persistants" ], "minimal": [ "Consultation médicale recommandée dans les 48h", "Tests complémentaires nécessaires (expectoration, culture)", "Surveillance étroite de l'évolution", "Isolement préventif jusqu'à confirmation" ], "moderate": [ "Consultation médicale URGENTE requise", "Hospitalisation pour évaluation complète", "Début immédiat du traitement antituberculeux", "Isolement strict jusqu'à négativation", "Recherche de contacts et dépistage familial" ], "advanced": [ "URGENCE MÉDICALE - Hospitalisation immédiate", "Traitement antituberculeux intensif requis", "Isolement strict en milieu hospitalier", "Surveillance des complications pulmonaires", "Dépistage et traitement préventif des contacts", "Suivi nutritionnel et psychologique" ] } return recommendations.get(stage, recommendations["normal"]) # Instance du détecteur detector = TuberculosisDetector() @app.post("/analyze", response_model=AnalysisResult) async def analyze_xray(file: UploadFile = File(...)): """ Analyse une radiographie pulmonaire pour détecter la tuberculose """ try: logger.info(f"Fichier reçu: {file.filename}, Content-Type: {file.content_type}") # Lire le contenu du fichier content = await file.read() file_size = len(content) logger.info(f"Taille du fichier: {file_size} bytes") # Vérifier la taille du fichier (limiter à 10MB) if file_size > 10 * 1024 * 1024: # 10MB raise HTTPException(status_code=400, detail="Le fichier est trop volumineux (max 10MB)") # Vérifier si c'est vraiment une image en regardant le contenu if not is_valid_image_content(content): # Vérification du content_type comme fallback if file.content_type and file.content_type not in ALLOWED_IMAGE_TYPES: raise HTTPException( status_code=400, detail=f"Type de fichier non supporté: {file.content_type}. Types acceptés: {', '.join(ALLOWED_IMAGE_TYPES)}" ) # Essayer de lire l'image try: image = Image.open(io.BytesIO(content)) logger.info(f"Image chargée avec succès - Format: {image.format}, Taille: {image.size}, Mode: {image.mode}") except Exception as e: logger.error(f"Erreur lors de l'ouverture de l'image: {e}") raise HTTPException(status_code=400, detail=f"Format d'image non valide ou image corrompue: {str(e)}") # Convertir en RGB si nécessaire if image.mode != 'RGB': image = image.convert('RGB') logger.info(f"Image convertie en RGB") # Préprocesser l'image processed_image = detector.preprocess_image(image) # Prédiction avec le modèle try: prediction = detector.model.predict(processed_image, verbose=0) confidence = float(prediction[0][0]) logger.info(f"Prédiction réussie - confiance: {confidence}") except Exception as e: logger.error(f"Erreur lors de la prédiction: {e}") raise HTTPException(status_code=500, detail=f"Erreur lors de la prédiction: {str(e)}") # Déterminer si c'est de la tuberculose is_tuberculosis = confidence >= config.CONFIDENCE_THRESHOLD # Détecter les nodules si tuberculose nodules = [] annotated_image = image nodule_details = [] if is_tuberculosis: nodules, annotated_image, nodule_details = detector.detect_nodules(image, confidence) # Classifier le stade stage, stage_description = detector.classify_stage(confidence, len(nodules)) # Obtenir les recommandations recommendations = detector.get_recommendations(stage) # Convertir l'image annotée en base64 try: buffered = io.BytesIO() annotated_image.save(buffered, format="PNG") img_base64 = base64.b64encode(buffered.getvalue()).decode() except Exception as e: logger.error(f"Erreur lors de la conversion en base64: {e}") img_base64 = "" # Retourner les résultats result = AnalysisResult( is_tuberculosis=is_tuberculosis, confidence=confidence, stage=stage, stage_description=stage_description, nodule_count=len(nodules), analyzed_image_base64=img_base64, recommendations=recommendations, nodule_details=nodule_details ) logger.info(f"Analyse terminée - TB: {is_tuberculosis}, Confiance: {confidence}, Stade: {stage}, Nodules: {len(nodules)}") return result except HTTPException: raise except Exception as e: logger.error(f"Erreur lors de l'analyse: {e}") raise HTTPException(status_code=500, detail=f"Erreur lors de l'analyse: {str(e)}") @app.get("/health") async def health_check(): """Vérification de l'état de l'API""" return { "status": "healthy", "model_loaded": detector.model is not None, "expected_image_size": config.IMAGE_SIZE } @app.get("/model-info") async def get_model_info(): """Informations sur le modèle""" model_info = { "model_loaded": detector.model is not None, "image_size": config.IMAGE_SIZE, "confidence_threshold": config.CONFIDENCE_THRESHOLD, "stages": list(config.STAGE_THRESHOLDS.keys()), "allowed_content_types": list(ALLOWED_IMAGE_TYPES) } if detector.model is not None: model_info["model_input_shape"] = str(detector.model.input_shape) return model_info @app.post("/batch-analyze") async def batch_analyze(files: List[UploadFile] = File(...)): """ Analyse multiple de radiographies """ if len(files) > 10: # Limiter le nombre de fichiers raise HTTPException(status_code=400, detail="Maximum 10 fichiers autorisés") results = [] for file in files: try: # Réutiliser la logique d'analyse result = await analyze_xray(file) results.append({ "filename": file.filename, "result": result }) except Exception as e: results.append({ "filename": file.filename, "error": str(e) }) return {"results": results} if __name__ == "__main__": uvicorn.run( "main:app", host="0.0.0.0", port=8000, reload=True, log_level="info" )