application2/lib/api/main.py

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"
    )