openclassroom-ai/oc-p09-street-vision-poc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

website color mapping update

Browse source
This commit is contained in:
Grégoire Mureau 2025-09-04 10:15:38 +02:00
parent 9186bfd1c5
commit 830facaf41
3 changed files with 0 additions and 1738 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

755
app_2/main.py
View file

@ -1,755 +0,0 @@
import os
import sys
import traceback
import tempfile
import shutil
from datetime import datetime
from typing import Optional, List
from pathlib import Path
import mlflow
from fastapi import FastAPI, HTTPException, File, UploadFile
from fastapi.responses import FileResponse
from pydantic import BaseModel
import numpy as np
import pickle
import base64
import matplotlib.pyplot as plt
import matplotlib.image as mpimg
# Configuration MLflow avec variables d'environnement
def configure_mlflow():
"""Configure MLflow avec les variables d'environnement"""
print("=== DEBUG VARIABLES D'ENVIRONNEMENT ===")
mlflow_tracking_uri = os.getenv("MLFLOW_TRACKING_URI", "").strip()
mlflow_s3_endpoint_url = os.getenv("MLFLOW_S3_ENDPOINT_URL", "").strip()
aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID", "").strip()
aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY", "").strip()
print(f"MLFLOW_TRACKING_URI: {mlflow_tracking_uri}")
print(f"MLFLOW_S3_ENDPOINT_URL: {mlflow_s3_endpoint_url}")
print(f"AWS_ACCESS_KEY_ID: {aws_access_key_id}")
print(f"AWS_SECRET_ACCESS_KEY: {'***' if aws_secret_access_key else None} ({'set' if aws_secret_access_key else 'not set'})")
print("============================================")
if not mlflow_tracking_uri:
raise ValueError("MLFLOW_TRACKING_URI not found in environment variables")
mlflow.set_tracking_uri(mlflow_tracking_uri)
print(f"MLflow Tracking URI: {mlflow_tracking_uri}")
if not mlflow_s3_endpoint_url or not aws_access_key_id or not aws_secret_access_key:
raise ValueError("Variables d'environnement MLflow manquantes (S3_ENDPOINT, ACCESS_KEY, SECRET_KEY)")
os.environ["MLFLOW_S3_ENDPOINT_URL"] = mlflow_s3_endpoint_url
os.environ["AWS_ACCESS_KEY_ID"] = aws_access_key_id
os.environ["AWS_SECRET_ACCESS_KEY"] = aws_secret_access_key
print(f"MLflow S3 Endpoint: {mlflow_s3_endpoint_url}")
print("✅ Identifiants AWS configurés")
print("✅ Configuration MLflow terminée")
# Import des fonctions utils
sys.path.append('.')
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from utils.utils import (
load_cityscapes_config,
load_model,
run_inference_and_visualize,
colorize_mask
)
# Configuration
MODEL_NAME = "StreetsSegmentation"
SAMPLE_IMAGES_DIR = "./notebooks/content/data/test_images_sample"
TRAIN_IMAGES_DIR = "./notebooks/content/data/train_images_sample"
CITYSCAPES_CONFIG_PATH = "./cityscapes_config.json"
TEMP_UPLOAD_DIR = "/tmp/uploads"
# Modèles de données mis à jour
class ModelInfo(BaseModel):
name: str
run_id: str
encoder_name: str
input_size: tuple
rank: int
architecture: Optional[str] = None
test_mean_iou: Optional[float] = None
test_accuracy: Optional[float] = None
class_ious: Optional[dict] = None
class ComparisonResult(BaseModel):
success: bool
message: str
model1_info: ModelInfo
model2_info: ModelInfo
image_path: Optional[str] = None
mask_path: Optional[str] = None
model1_stats: Optional[dict] = None
model2_stats: Optional[dict] = None
model1_inference_time: float
model2_inference_time: float
speed_comparison: dict
ground_truth_available: bool = False
figure_data: Optional[str] = None
class ImageInfo(BaseModel):
filename: str
display_name: str
has_ground_truth: bool
class AvailableImagesResponse(BaseModel):
images: List[ImageInfo]
total_count: int
class PreprocessingRequest(BaseModel):
filename: str
apply_augmentation: bool = False
class PreprocessingResult(BaseModel):
success: bool
message: str
original_image: Optional[str] = None
normalized_image: Optional[str] = None
augmented_image: Optional[str] = None
transformations_applied: Optional[list] = None
class SampleImageRequest(BaseModel):
filename: str
# Variables globales pour les deux modèles
best_model = None
second_best_model = None
best_model_info = {}
second_best_model_info = {}
mapping_config = None
def get_model_detailed_info(run_id, mapping_config):
"""Récupère les informations détaillées d'un modèle depuis MLflow"""
client = mlflow.tracking.MlflowClient()
try:
run = mlflow.get_run(run_id)
params = run.data.params
metrics = run.data.metrics
# Architecture depuis le paramètre model_architecture
architecture = params.get("model_architecture", "Unknown").split('_')[0]
# Métriques de performance
test_mean_iou = metrics.get("test_mean_iou", None)
test_accuracy = metrics.get("test_accuracy", None)
# IoU par classe depuis les métriques class_X_iou
class_ious = {}
if mapping_config and 'group_names' in mapping_config:
for i, class_name in enumerate(mapping_config['group_names']):
metric_key = f"class_{i}_iou"
if metric_key in metrics:
class_ious[class_name] = round(metrics[metric_key], 3)
return {
"architecture": architecture,
"test_mean_iou": test_mean_iou,
"test_accuracy": test_accuracy,
"class_ious": class_ious
}
except Exception as e:
print(f"⚠️ Erreur lors de la récupération des infos détaillées: {e}")
return {
"architecture": "Unknown",
"test_mean_iou": None,
"test_accuracy": None,
"class_ious": {}
}
def initialize_models():
"""Initialise les deux meilleurs modèles au démarrage"""
global best_model, second_best_model, best_model_info, second_best_model_info, mapping_config
try:
print("🔄 Configuration de MLflow...")
configure_mlflow()
print("🔄 Chargement de la configuration Cityscapes...")
mapping_config = load_cityscapes_config(CITYSCAPES_CONFIG_PATH, verbose=False)
print("✅ Configuration Cityscapes chargée")
print("🔄 Chargement du meilleur modèle...")
best_model, best_run_id, best_encoder_name, best_img_size = load_model(
experiment_name="OC Projet 9",
metric="test_mean_iou",
top_n=1
)
# Récupérer les infos détaillées du meilleur modèle
best_detailed_info = get_model_detailed_info(best_run_id, mapping_config)
best_model_info = {
"name": "Meilleur modèle",
"run_id": best_run_id,
"encoder_name": best_encoder_name,
"input_size": best_img_size,
"rank": 1,
"architecture": best_detailed_info["architecture"],
"test_mean_iou": best_detailed_info["test_mean_iou"],
"test_accuracy": best_detailed_info["test_accuracy"],
"class_ious": best_detailed_info["class_ious"]
}
print("🔄 Chargement du deuxième meilleur modèle...")
second_best_model, second_run_id, second_encoder_name, second_img_size = load_model(
experiment_name="OC Projet 9",
metric="test_mean_iou",
top_n=2
)
# Récupérer les infos détaillées du deuxième modèle
second_detailed_info = get_model_detailed_info(second_run_id, mapping_config)
second_best_model_info = {
"name": "Deuxième modèle",
"run_id": second_run_id,
"encoder_name": second_encoder_name,
"input_size": second_img_size,
"rank": 2,
"architecture": second_detailed_info["architecture"],
"test_mean_iou": second_detailed_info["test_mean_iou"],
"test_accuracy": second_detailed_info["test_accuracy"],
"class_ious": second_detailed_info["class_ious"]
}
os.makedirs(TEMP_UPLOAD_DIR, exist_ok=True)
except Exception as e:
print(f"❌ Erreur lors de l'initialisation: {e}")
traceback.print_exc()
raise e
# Initialisation de l'API
app = FastAPI(
title="Multi-Class Segmentation Comparison API",
description="API de comparaison de segmentation sémantique avec deux modèles",
version="2.0.0"
)
@app.on_event("startup")
async def startup_event():
try:
initialize_models()
print("🚀 API démarrée avec succès - Modèles chargés!")
except Exception as e:
print(f"⚠️ ATTENTION: Impossible de charger les modèles au démarrage: {e}")
@app.get("/")
def root():
return {
"message": "Multi-Class Segmentation Comparison API - Online",
"models_loaded": best_model is not None and second_best_model is not None,
"best_model": best_model_info if best_model else None,
"second_best_model": second_best_model_info if second_best_model else None,
"num_classes": mapping_config['num_classes'] if mapping_config else None
}
@app.get("/health")
def health_check():
if best_model is None or second_best_model is None or mapping_config is None:
raise HTTPException(status_code=503, detail="Models not loaded")
return {
"status": "healthy",
"models_loaded": True,
"timestamp": datetime.utcnow().isoformat()
}
@app.get("/models", response_model=dict)
def get_models_info():
if best_model is None or second_best_model is None:
raise HTTPException(status_code=503, detail="Models not loaded")
return {
"best_model": best_model_info,
"second_best_model": second_best_model_info,
"num_classes": mapping_config['num_classes'],
"class_names": mapping_config['group_names'],
"class_colors": mapping_config['group_colors'].tolist()
}
@app.get("/dataset-info")
def get_dataset_info():
"""Retourne les informations sur le dataset"""
return {
"name": "Cityscapes Dataset",
"description": "Référence internationale pour l'évaluation des algorithmes de segmentation sémantique en environnement urbain",
"total_images": 25000,
"resolution": "2048×1024 pixels",
"cities": 50,
"countries": ["Allemagne", "Suisse"],
"splits": {
"train": {"images": 2975, "usage": "Entraînement avec annotations fines"},
"validation": {"images": 500, "usage": "Validation avec annotations de référence"},
"test": {"images": 1525, "usage": "Test (annotations non-publiques)"},
"coarse": {"images": 20000, "usage": "Entraînement avec annotations grossières"}
},
"project_splits": {
"train": {"images": 2380, "percentage": 80, "source": "Cityscapes train"},
"validation": {"images": 595, "percentage": 20, "source": "Cityscapes train"},
"test": {"images": 500, "percentage": 100, "source": "Cityscapes val"}
},
"meta_classes": mapping_config['group_names'] if mapping_config else [],
"num_classes": mapping_config['num_classes'] if mapping_config else 0
}
@app.get("/train-images", response_model=AvailableImagesResponse)
def get_train_images():
"""Retourne la liste des images d'entraînement disponibles"""
try:
original_dir = Path(TRAIN_IMAGES_DIR) / "original"
mask_dir = Path(TRAIN_IMAGES_DIR) / "mask"
if not original_dir.exists():
raise HTTPException(status_code=404, detail="Train images directory not found")
images = []
for img_path in original_dir.glob("*leftImg8bit.png"):
# Construire le nom du masque correspondant
base_name = img_path.stem.replace("_leftImg8bit", "")
mask_name = f"{base_name}_gtFine_labelIds.png"
mask_path = mask_dir / mask_name
# Créer un nom d'affichage plus convivial
display_name = base_name.replace("_", " ").title()
images.append(ImageInfo(
filename=img_path.name,
display_name=display_name,
has_ground_truth=mask_path.exists()
))
# Trier par nom pour un affichage cohérent
images.sort(key=lambda x: x.filename)
return AvailableImagesResponse(
images=images,
total_count=len(images)
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error listing train images: {str(e)}")
@app.get("/train-image/{filename}")
def get_train_image(filename: str):
"""Sert une image d'entraînement pour prévisualisation"""
try:
original_dir = Path(TRAIN_IMAGES_DIR) / "original"
image_path = original_dir / filename
if not image_path.exists():
raise HTTPException(status_code=404, detail=f"Train image {filename} not found")
# Vérifier que c'est bien un fichier image attendu
if not filename.endswith('_leftImg8bit.png'):
raise HTTPException(status_code=400, detail="Invalid image filename format")
return FileResponse(
image_path,
media_type="image/png",
headers={"Content-Disposition": f"inline; filename={filename}"}
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error serving train image: {str(e)}")
@app.get("/class-repartition")
def get_class_repartition():
"""Sert le graphique de répartition des classes"""
try:
graph_path = Path(TRAIN_IMAGES_DIR) / "class_repartition.png"
if not graph_path.exists():
raise HTTPException(status_code=404, detail="Class repartition graph not found")
return FileResponse(
graph_path,
media_type="image/png",
headers={"Content-Disposition": "inline; filename=class_repartition.png"}
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error serving class repartition graph: {str(e)}")
@app.post("/preprocess-train-image", response_model=PreprocessingResult)
def preprocess_train_image(request: PreprocessingRequest):
"""Prétraite une image d'entraînement et retourne les versions transformées"""
if mapping_config is None:
raise HTTPException(status_code=503, detail="Configuration not loaded")
try:
original_dir = Path(TRAIN_IMAGES_DIR) / "original"
image_path = original_dir / request.filename
if not image_path.exists():
raise HTTPException(status_code=404, detail=f"Train image {request.filename} not found")
# Importer les fonctions nécessaires
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import io
import base64
from utils.utils import get_preprocessing_fn, apply_random_augmentation
# Charger l'image originale
original_pil = Image.open(image_path).convert('RGB')
original_np = np.array(original_pil)
# Appliquer le preprocessing standard (resize + normalisation)
preprocess_fn = get_preprocessing_fn(img_size=(512, 512), encoder=None)
normalized_tensor, _ = preprocess_fn(original_np, None)
# Convertir le tensor en image pour affichage
normalized_np = normalized_tensor.permute(1, 2, 0).numpy()
# Dénormaliser pour l'affichage (tensor était normalisé 0-1)
normalized_display = (normalized_np * 255).astype(np.uint8)
# Créer les images en base64 pour l'envoi
def img_to_base64(img_array):
img_pil = Image.fromarray(img_array)
buffer = io.BytesIO()
img_pil.save(buffer, format='PNG')
return base64.b64encode(buffer.getvalue()).decode()
# Image originale redimensionnée pour comparaison
original_resized = np.array(original_pil.resize((512, 512)))
result = PreprocessingResult(
success=True,
message="Preprocessing completed successfully",
original_image=img_to_base64(original_resized),
normalized_image=img_to_base64(normalized_display),
transformations_applied=["Resize (512x512)", "Normalization (0-1)"]
)
# Appliquer l'augmentation si demandée
if request.apply_augmentation:
augmented_np, transforms_applied = apply_random_augmentation(original_np)
# Appliquer aussi le preprocessing à l'image augmentée
augmented_tensor, _ = preprocess_fn(augmented_np, None)
augmented_display = (augmented_tensor.permute(1, 2, 0).numpy() * 255).astype(np.uint8)
result.augmented_image = img_to_base64(augmented_display)
result.transformations_applied.extend(transforms_applied)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"Preprocessing error: {str(e)}")
@app.get("/sample-images", response_model=AvailableImagesResponse)
def get_sample_images():
try:
original_dir = Path(SAMPLE_IMAGES_DIR) / "original"
mask_dir = Path(SAMPLE_IMAGES_DIR) / "mask"
if not original_dir.exists():
raise HTTPException(status_code=404, detail="Sample images directory not found")
images = []
for img_path in original_dir.glob("*leftImg8bit.png"):
base_name = img_path.stem.replace("_leftImg8bit", "")
mask_name = f"{base_name}_gtFine_labelIds.png"
mask_path = mask_dir / mask_name
display_name = base_name.replace("_", " ").title()
images.append(ImageInfo(
filename=img_path.name,
display_name=display_name,
has_ground_truth=mask_path.exists()
))
images.sort(key=lambda x: x.filename)
return AvailableImagesResponse(images=images, total_count=len(images))
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error listing sample images: {str(e)}")
@app.get("/sample-image/{filename}")
def get_sample_image(filename: str):
try:
original_dir = Path(SAMPLE_IMAGES_DIR) / "original"
image_path = original_dir / filename
if not image_path.exists():
raise HTTPException(status_code=404, detail=f"Sample image {filename} not found")
if not filename.endswith('_leftImg8bit.png'):
raise HTTPException(status_code=400, detail="Invalid image filename format")
return FileResponse(
image_path,
media_type="image/png",
headers={"Content-Disposition": f"inline; filename={filename}"}
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error serving image: {str(e)}")
@app.post("/compare-sample", response_model=ComparisonResult)
def compare_sample_image(request: SampleImageRequest):
if best_model is None or second_best_model is None or mapping_config is None:
raise HTTPException(status_code=503, detail="Models not loaded")
try:
original_dir = Path(SAMPLE_IMAGES_DIR) / "original"
image_path = original_dir / request.filename
if not image_path.exists():
raise HTTPException(status_code=404, detail=f"Sample image {request.filename} not found")
base_name = image_path.stem.replace("_leftImg8bit", "")
mask_name = f"{base_name}_gtFine_labelIds.png"
mask_path = Path(SAMPLE_IMAGES_DIR) / "mask" / mask_name
mask_path_str = str(mask_path) if mask_path.exists() else None
# Inférence modèle 1 avec sa résolution spécifique
pred1, t1, fig1 = run_inference_and_visualize(
image_path=str(image_path),
mask_path=mask_path_str,
model=best_model,
encoder_name=None, # Normalisé de la même manière
img_size=best_model_info["input_size"], # Utilise la résolution du modèle
mapping_config=mapping_config,
save_dir=None, show_stats=False, show_plot=False
)
# Inférence modèle 2 avec sa résolution spécifique
pred2, t2, fig2 = run_inference_and_visualize(
image_path=str(image_path),
mask_path=mask_path_str,
model=second_best_model,
encoder_name=None, # Normalisé de la même manière
img_size=second_best_model_info["input_size"], # Utilise la résolution du modèle
mapping_config=mapping_config,
save_dir=None, show_stats=False, show_plot=False
)
# Créer une figure combinée avec les deux modèles
# Nombre de colonnes : 3 si pas de GT, 4 si GT disponible
num_cols = 3 if not (mask_path_str and os.path.exists(mask_path_str)) else 4
fig, axes = plt.subplots(2, num_cols, figsize=(5*num_cols, 8))
# Récupérer les images depuis les figures individuelles
axes1 = fig1.get_axes()
axes2 = fig2.get_axes()
# Ligne 1 : Modèle 1
for i in range(min(len(axes1), num_cols)):
ax_src = axes1[i]
ax_dst = axes[0, i]
# Copier toutes les images de l'axe source
for img_obj in ax_src.get_images():
ax_dst.imshow(
img_obj.get_array(),
cmap=img_obj.get_cmap(),
alpha=img_obj.get_alpha(),
vmin=img_obj.get_clim()[0] if img_obj.get_clim() else None,
vmax=img_obj.get_clim()[1] if img_obj.get_clim() else None
)
# Copier le titre avec nom du modèle
original_title = ax_src.get_title()
if i == 0:
new_title = "Image originale"
elif "Prédit" in original_title:
new_title = f"Prédiction {best_model_info['architecture']} - {best_model_info['encoder_name']}"
else:
new_title = original_title
ax_dst.set_title(new_title)
ax_dst.axis('off')
# Ligne 2 : Modèle 2
for i in range(min(len(axes2), num_cols)):
ax_src = axes2[i]
ax_dst = axes[1, i]
# Copier toutes les images de l'axe source
for img_obj in ax_src.get_images():
ax_dst.imshow(
img_obj.get_array(),
cmap=img_obj.get_cmap(),
alpha=img_obj.get_alpha(),
vmin=img_obj.get_clim()[0] if img_obj.get_clim() else None,
vmax=img_obj.get_clim()[1] if img_obj.get_clim() else None
)
# Copier le titre avec nom du modèle
original_title = ax_src.get_title()
if i == 0:
new_title = "Image originale"
elif "Prédit" in original_title:
new_title = f"Prédiction {second_best_model_info['architecture']} - {second_best_model_info['encoder_name']}"
else:
new_title = original_title
ax_dst.set_title(new_title)
ax_dst.axis('off')
plt.tight_layout()
# Fermer les figures individuelles pour libérer la mémoire
plt.close(fig1)
plt.close(fig2)
fig_b64 = base64.b64encode(pickle.dumps(fig)).decode()
# Stats
def calc(mask):
uc, cnts = np.unique(mask, return_counts=True)
stats = {}
for cid, c in zip(uc, cnts):
if cid < len(mapping_config['group_names']):
stats[mapping_config['group_names'][cid]] = {
"pixels": int(c),
"percentage": round((c/mask.size)*100, 1)
}
return stats
stats1, stats2 = calc(pred1), calc(pred2)
faster = best_model_info['encoder_name'] if t1 < t2 else second_best_model_info['encoder_name']
speedup = round(max(t1, t2) / min(t1, t2), 2)
speed_comp = {"faster_model": faster, "speedup_ratio": speedup, "time_difference_ms": abs(t1-t2)*1000}
return ComparisonResult(
success=True, message="Comparison completed successfully",
model1_info=ModelInfo(**best_model_info), model2_info=ModelInfo(**second_best_model_info),
image_path=str(image_path), mask_path=mask_path_str,
model1_stats=stats1, model2_stats=stats2,
model1_inference_time=t1, model2_inference_time=t2,
speed_comparison=speed_comp, ground_truth_available=mask_path.exists(),
figure_data=fig_b64
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Comparison error: {str(e)}")
@app.post("/compare-upload", response_model=ComparisonResult)
def compare_uploaded_image(file: UploadFile = File(...)):
if best_model is None or second_best_model is None or mapping_config is None:
raise HTTPException(status_code=503, detail="Models not loaded")
allowed_types = ["image/jpeg", "image/jpg", "image/png"]
if file.content_type not in allowed_types:
raise HTTPException(status_code=400, detail=f"Invalid file type. Allowed: {', '.join(allowed_types)}")
temp_file_path = None
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
temp_filename = f"upload_{timestamp}_{file.filename}"
temp_file_path = os.path.join(TEMP_UPLOAD_DIR, temp_filename)
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Inférences avec résolutions spécifiques
pred1, t1, fig1 = run_inference_and_visualize(
image_path=temp_file_path, mask_path=None,
model=best_model, encoder_name=None, # Normalisé de la même manière
img_size=best_model_info["input_size"], mapping_config=mapping_config,
save_dir=None, show_stats=False, show_plot=False,
)
pred2, t2, fig2 = run_inference_and_visualize(
image_path=temp_file_path, mask_path=None,
model=second_best_model, encoder_name=None, # Normalisé de la même manière
img_size=second_best_model_info["input_size"], mapping_config=mapping_config,
save_dir=None, show_stats=False, show_plot=False,
)
# Figure combinée pour uploads (pas de GT, donc 2 colonnes seulement)
fig, axes = plt.subplots(2, 2, figsize=(10, 8))
# Récupérer les images depuis les figures individuelles
axes1 = fig1.get_axes()
axes2 = fig2.get_axes()
# Ligne 1 : Modèle 1 (Image originale + Prédiction)
for i in range(min(len(axes1), 2)):
ax_src = axes1[i]
ax_dst = axes[0, i]
# Copier toutes les images de l'axe source
for img_obj in ax_src.get_images():
ax_dst.imshow(
img_obj.get_array(),
cmap=img_obj.get_cmap(),
alpha=img_obj.get_alpha(),
vmin=img_obj.get_clim()[0] if img_obj.get_clim() else None,
vmax=img_obj.get_clim()[1] if img_obj.get_clim() else None
)
# Copier le titre avec nom du modèle
original_title = ax_src.get_title()
if i == 0:
new_title = "Image originale"
else:
new_title = f"Prédiction {best_model_info['architecture']} - {best_model_info['encoder_name']}"
ax_dst.set_title(new_title)
ax_dst.axis('off')
# Ligne 2 : Modèle 2 (Image originale + Prédiction)
for i in range(min(len(axes2), 2)):
ax_src = axes2[i]
ax_dst = axes[1, i]
# Copier toutes les images de l'axe source
for img_obj in ax_src.get_images():
ax_dst.imshow(
img_obj.get_array(),
cmap=img_obj.get_cmap(),
alpha=img_obj.get_alpha(),
vmin=img_obj.get_clim()[0] if img_obj.get_clim() else None,
vmax=img_obj.get_clim()[1] if img_obj.get_clim() else None
)
# Copier le titre avec nom du modèle
original_title = ax_src.get_title()
if i == 0:
new_title = "Image originale"
else:
new_title = f"Prédiction {second_best_model_info['architecture']} - {second_best_model_info['encoder_name']}"
ax_dst.set_title(new_title)
ax_dst.axis('off')
plt.tight_layout()
# Fermer les figures individuelles pour libérer la mémoire
plt.close(fig1)
plt.close(fig2)
fig_b64 = base64.b64encode(pickle.dumps(fig)).decode()
# Stats
def calc(mask):
uc, cnts = np.unique(mask, return_counts=True)
stats = {}
for cid, c in zip(uc, cnts):
if cid < len(mapping_config['group_names']):
stats[mapping_config['group_names'][cid]] = {
"pixels": int(c),
"percentage": round((c/mask.size)*100, 1)
}
return stats
stats1, stats2 = calc(pred1), calc(pred2)
faster = best_model_info['encoder_name'] if t1 < t2 else second_best_model_info['encoder_name']
speedup = round(max(t1, t2) / min(t1, t2), 2)
speed_comp = {"faster_model": faster, "speedup_ratio": speedup, "time_difference_ms": abs(t1-t2)*1000}
return ComparisonResult(
success=True, message="Comparison completed successfully",
model1_info=ModelInfo(**best_model_info), model2_info=ModelInfo(**second_best_model_info),
image_path=temp_file_path,
model1_stats=stats1, model2_stats=stats2,
model1_inference_time=t1, model2_inference_time=t2,
speed_comparison=speed_comp, ground_truth_available=False,
figure_data=fig_b64
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Comparison error: {str(e)}")
finally:
if temp_file_path and os.path.exists(temp_file_path):
try: os.remove(temp_file_path)
except Exception: pass
@app.get("/metrics")
def get_metrics():
return {
"models_loaded": best_model is not None and second_best_model is not None,
"best_model_info": best_model_info,
"second_best_model_info": second_best_model_info,
"num_classes": mapping_config['num_classes'] if mapping_config else None,
"timestamp": datetime.utcnow().isoformat()
}

70
app_2/requirements.txt
View file

@ -1,70 +0,0 @@
# =============================================================================
# Multi-Class Segmentation API - Requirements
# =============================================================================
# ===== CORE WEB FRAMEWORK =====
fastapi==0.115.14
uvicorn==0.35.0
starlette==0.46.2
pydantic==2.11.7
pydantic_core==2.33.2
python-multipart==0.0.20 # uploads FastAPI
# ===== STREAMLIT =====
streamlit==1.46.1
altair==5.5.0 # utilisé par Streamlit
blinker==1.9.0 # dépendance Streamlit
# ===== HTTP CLIENTS =====
requests==2.32.4
httpx==0.28.1 # utilisé en interne par FastAPI
httpcore==1.0.9
h11==0.16.0
# ===== MACHINE LEARNING CORE (PyTorch) =====
torch==2.8.0 # version stable PyTorch
torchvision==0.23.0 # utile pour les modèles pré-entraînés
torchaudio==2.8.0 # optionnel (mais recommandé pour compatibilité PyTorch)
segmentation-models-pytorch==0.5.0
albumentations==2.0.8
# version PyTorch de ToTensor
albumentations[imgaug,opencv]
numpy>=1.24.0,<2.0.0
scipy==1.16.0
scikit-learn==1.7.0
# ===== COMPUTER VISION =====
pillow==11.3.0
opencv-python-headless==4.9.0.80
scikit-image==0.25.2
imageio==2.37.0
# ===== DATA SCIENCE =====
pandas==2.3.0
matplotlib==3.10.3
networkx==3.5
tifffile==2025.6.11
lazy_loader==0.4
# ===== MLFLOW & MODEL REGISTRY =====
mlflow==3.2.0
mlflow-skinny==3.2.0
mlflow-tracing==3.2.0
# ===== AWS/S3 SUPPORT =====
boto3==1.39.2
botocore==1.39.2
s3transfer==0.13.0
# ===== UTILITIES =====
click==8.2.1
tqdm==4.67.1
joblib==1.5.1
python-dateutil
packaging
typing_extensions==4.14.0
# ===== MATPLOTLIB ADDONS =====
cycler==0.12.1
fonttools==4.58.5

913
app_2/streamlit_app.py
View file

@ -1,913 +0,0 @@
import os
import sys
import requests
import streamlit as st
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import io
import base64
import pickle
# Configuration de l'API
API_BASE_URL = os.getenv("MULTISEG_API_BASE_URL", "http://127.0.0.1:8080/api")
# Import des fonctions utils pour l'affichage local
sys.path.append('/app')
sys.path.append('.')
from utils.utils import colorize_mask
# Configuration de la page
st.set_page_config(
page_title="Street Vision - Segmentation Comparison",
page_icon="🚗",
layout="wide"
)
# CSS personnalisé mis à jour
st.markdown("""
<style>
.main-header {
background: linear-gradient(90deg, #667eea 0%, #764ba2 100%);
padding: 2rem;
border-radius: 10px;
margin-bottom: 2rem;
color: white;
text-align: center;
}
.comparison-card {
background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
padding: 1.5rem;
border-radius: 10px;
border-left: 4px solid #667eea;
margin: 1rem 0;
color: #2c3e50;
font-weight: 500;
}
.speed-winner {
background: linear-gradient(135deg, #d4edda 0%, #c3e6cb 100%);
border-left-color: #28a745;
}
.speed-slower {
background: linear-gradient(135deg, #f8d7da 0%, #f1b0b7 100%);
border-left-color: #dc3545;
}
.metric-card {
background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);
padding: 1rem;
border-radius: 8px;
border-left: 4px solid #667eea;
margin: 0.5rem 0;
color: #2c3e50;
font-weight: 500;
}
.dataset-info {
background: #f8f9fa;
padding: 1.5rem;
border-radius: 10px;
margin: 1rem 0;
border-left: 4px solid #17a2b8;
}
.preprocessing-card {
background: linear-gradient(135deg, #e3f2fd 0%, #bbdefb 100%);
padding: 1rem;
border-radius: 8px;
margin: 0.5rem 0;
border-left: 4px solid #2196f3;
}
</style>
""", unsafe_allow_html=True)
# En-tête principal
st.markdown("""
<div class="main-header">
<h1>🚗 Street Vision - Comparaison de Modèles</h1>
<p>Évaluation comparative SegFormer vs FPN sur le dataset Cityscapes</p>
</div>
""", unsafe_allow_html=True)
# Fonctions utilitaires
def get_models_info():
"""Récupère les informations des deux modèles"""
try:
response = requests.get(f"{API_BASE_URL}/models")
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur lors de la récupération des infos modèles: {response.status_code}")
return None
except Exception as e:
st.error(f"Erreur de connexion à l'API: {e}")
return None
def get_dataset_info():
"""Récupère les informations sur le dataset"""
try:
response = requests.get(f"{API_BASE_URL}/dataset-info")
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur lors de la récupération des infos dataset: {response.status_code}")
return None
except Exception as e:
st.error(f"Erreur de connexion à l'API: {e}")
return None
def get_train_images():
"""Récupère la liste des images d'entraînement"""
try:
response = requests.get(f"{API_BASE_URL}/train-images")
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur lors de la récupération des images: {response.status_code}")
return None
except Exception as e:
st.error(f"Erreur de connexion à l'API: {e}")
return None
def get_class_repartition():
"""Récupère le graphique de répartition des classes"""
try:
response = requests.get(f"{API_BASE_URL}/class-repartition", timeout=10)
if response.status_code == 200:
return response.content
else:
return None
except Exception as e:
st.warning(f"Graphique de répartition non disponible: {e}")
return None
def preprocess_train_image(filename, apply_augmentation=False):
"""Lance le preprocessing sur une image d'entraînement"""
try:
data = {
"filename": filename,
"apply_augmentation": apply_augmentation
}
response = requests.post(
f"{API_BASE_URL}/preprocess-train-image",
json=data,
headers={"Content-Type": "application/json"}
)
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur API: {response.status_code} - {response.text}")
return None
except Exception as e:
st.error(f"Erreur lors du preprocessing: {e}")
return None
def get_sample_images():
"""Récupère la liste des images d'exemple"""
try:
response = requests.get(f"{API_BASE_URL}/sample-images")
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur lors de la récupération des images: {response.status_code}")
return None
except Exception as e:
st.error(f"Erreur de connexion à l'API: {e}")
return None
def compare_sample_image(filename):
"""Lance une comparaison sur une image d'exemple"""
try:
data = {"filename": filename}
response = requests.post(
f"{API_BASE_URL}/compare-sample",
json=data,
headers={"Content-Type": "application/json"}
)
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur API: {response.status_code} - {response.text}")
return None
except Exception as e:
st.error(f"Erreur lors de la comparaison: {e}")
return None
def compare_uploaded_image(uploaded_file):
"""Lance une comparaison sur une image uploadée"""
try:
files = {"file": (uploaded_file.name, uploaded_file.getvalue(), uploaded_file.type)}
response = requests.post(f"{API_BASE_URL}/compare-upload", files=files)
if response.status_code == 200:
return response.json()
else:
st.error(f"Erreur API: {response.status_code} - {response.text}")
return None
except Exception as e:
st.error(f"Erreur lors de la comparaison: {e}")
return None
def display_comparison_results(result, models_info):
"""Affiche les résultats de comparaison entre les deux modèles"""
if not result or not result['success']:
st.error("Aucun résultat de comparaison disponible")
return
# Informations des modèles
st.markdown("### 🏆 Modèles Comparés")
col1, col2 = st.columns(2)
with col1:
model1 = result['model1_info']
# Formater les métriques séparément
miou1 = f"{model1.get('test_mean_iou'):.4f}" if model1.get('test_mean_iou') is not None else 'N/A'
acc1 = f"{model1.get('test_accuracy'):.4f}" if model1.get('test_accuracy') is not None else 'N/A'
st.markdown(f"""
<div class="comparison-card">
<h4>🥇 Modèle #1</h4>
<strong>{model1.get('architecture', 'N/A')} - {model1['encoder_name']}</strong><br>
📐 Taille: {model1['input_size']}<br>
📊 mIoU: {miou1}<br>
🎯 Accuracy: {acc1}
</div>
""", unsafe_allow_html=True)
with col2:
model2 = result['model2_info']
# Formater les métriques séparément
miou2 = f"{model2.get('test_mean_iou'):.4f}" if model2.get('test_mean_iou') is not None else 'N/A'
acc2 = f"{model2.get('test_accuracy'):.4f}" if model2.get('test_accuracy') is not None else 'N/A'
st.markdown(f"""
<div class="comparison-card">
<h4>🥈 Modèle #2</h4>
<strong>{model2.get('architecture', 'N/A')} - {model2['encoder_name']}</strong><br>
📐 Taille: {model2['input_size']}<br>
📊 mIoU: {miou2}<br>
🎯 Accuracy: {acc2}
</div>
""", unsafe_allow_html=True)
# Comparaison des temps d'inférence
st.markdown("### ⏱️ Comparaison des Performances")
speed_comp = result['speed_comparison']
model1_time = result['model1_inference_time'] * 1000
model2_time = result['model2_inference_time'] * 1000
col1, col2, col3 = st.columns(3)
with col1:
winner_class = "speed-winner" if speed_comp['faster_model'] == model1['encoder_name'] else "speed-slower"
st.markdown(f"""
<div class="comparison-card {winner_class}">
<h5>🥇 {model1.get('architecture', 'N/A')} - {model1['encoder_name']}</h5>
<strong>{model1_time:.1f} ms</strong>
</div>
""", unsafe_allow_html=True)
with col2:
st.markdown(f"""
<div class="comparison-card">
<h5>⚡ Comparaison</h5>
<strong>{speed_comp['faster_model']}</strong><br>
est {speed_comp['speedup_ratio']}x plus rapide<br>
<small>Différence: {speed_comp['time_difference_ms']:.1f} ms</small>
</div>
""", unsafe_allow_html=True)
with col3:
winner_class = "speed-winner" if speed_comp['faster_model'] == model2['encoder_name'] else "speed-slower"
st.markdown(f"""
<div class="comparison-card {winner_class}">
<h5>🥈 {model2.get('architecture', 'N/A')} - {model2['encoder_name']}</h5>
<strong>{model2_time:.1f} ms</strong>
</div>
""", unsafe_allow_html=True)
# Affichage des segmentations
if result.get('figure_data'):
st.markdown("### 🎨 Résultats de Segmentation")
try:
fig_bytes = base64.b64decode(result['figure_data'])
fig = pickle.loads(fig_bytes)
st.pyplot(fig)
plt.close(fig)
except Exception as e:
st.error(f"Erreur lors de l'affichage: {e}")
# Comparaison des statistiques
st.markdown("### 📊 Répartition des classes par modèle")
col1, col2 = st.columns(2)
with col1:
if result.get('model1_stats'):
display_model_stats_chart(result['model1_stats'], models_info, f"{model1.get('architecture', 'N/A')} - {model1['encoder_name']}")
with col2:
if result.get('model2_stats'):
display_model_stats_chart(result['model2_stats'], models_info, f"{model2.get('architecture', 'N/A')} - {model2['encoder_name']}")
# Analyse comparative des classes
if result.get('model1_stats') and result.get('model2_stats'):
display_class_comparison(result['model1_stats'], result['model2_stats'], model1, model2, models_info)
def display_model_stats_chart(stats, models_info, title):
"""Affiche les statistiques d'un modèle sous forme de graphique"""
if not stats:
st.warning("Aucune statistique disponible")
return
# Utiliser l'ordre des classes du modèle pour la cohérence
if models_info and 'class_names' in models_info:
# Ordonner selon l'ordre du modèle
ordered_class_names = [name for name in models_info['class_names'] if name in stats]
else:
# Fallback : ordre alphabétique
ordered_class_names = sorted(stats.keys())
percentages = [stats[name]['percentage'] for name in ordered_class_names]
# Récupérer les couleurs des classes
colors = []
if models_info and 'class_colors' in models_info and 'class_names' in models_info:
model_class_names = models_info['class_names']
model_class_colors = models_info['class_colors']
for class_name in ordered_class_names:
try:
idx = model_class_names.index(class_name)
color = model_class_colors[idx]
colors.append([c/255.0 for c in color])
except (ValueError, IndexError):
colors.append([0.5, 0.5, 0.5])
else:
colors = plt.cm.Set3(np.linspace(0, 1, len(ordered_class_names)))
# Créer le graphique
fig, ax = plt.subplots(figsize=(8, 5))
bars = ax.bar(ordered_class_names, percentages, color=colors, alpha=0.8, edgecolor='white', linewidth=1)
ax.set_xlabel('Classes de Segmentation', fontsize=10, fontweight='bold')
ax.set_ylabel('Pourcentage (%)', fontsize=10, fontweight='bold')
ax.set_title(title, fontsize=11, fontweight='bold', pad=15)
# Rotation des labels selon le nombre de classes
if len(ordered_class_names) <= 4:
plt.xticks(rotation=0, ha='center', fontsize=9)
elif len(ordered_class_names) <= 6:
plt.xticks(rotation=30, ha='right', fontsize=8)
else:
plt.xticks(rotation=45, ha='right', fontsize=8)
plt.yticks(fontsize=9)
# Valeurs sur les barres
for bar, percentage in zip(bars, percentages):
height = bar.get_height()
ax.text(bar.get_x() + bar.get_width()/2., height + max(percentages) * 0.01,
f'{percentage:.1f}%',
ha='center', va='bottom', fontsize=8, fontweight='bold')
plt.tight_layout()
ax.grid(axis='y', alpha=0.2, linestyle='--')
ax.set_axisbelow(True)
# Style des axes
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
# Couleur de fond
fig.patch.set_facecolor('white')
ax.set_facecolor('#fafafa')
st.pyplot(fig)
plt.close(fig)
def display_class_comparison(stats1, stats2, model1, model2, models_info):
"""Affiche une comparaison détaillée des classes entre les deux modèles"""
st.markdown("### 🔍 Analyse Comparative par Classe")
# Utiliser l'ordre des classes du modèle pour la cohérence
if models_info and 'class_names' in models_info:
# Combiner toutes les classes présentes en conservant l'ordre du modèle
all_classes_in_stats = set(stats1.keys()) | set(stats2.keys())
ordered_all_classes = [name for name in models_info['class_names'] if name in all_classes_in_stats]
else:
# Fallback : ordre alphabétique
all_classes = set(stats1.keys()) | set(stats2.keys())
ordered_all_classes = sorted(all_classes)
comparison_data = []
for class_name in ordered_all_classes:
pct1 = stats1.get(class_name, {}).get('percentage', 0)
pct2 = stats2.get(class_name, {}).get('percentage', 0)
diff = pct1 - pct2
comparison_data.append({
'class': class_name,
'model1_pct': pct1,
'model2_pct': pct2,
'difference': diff
})
# Créer un graphique comparatif des barres
st.markdown("**📊 Comparaison des classes entre les modèles:**")
# Préparation des données
class_names = [data['class'] for data in comparison_data]
model1_values = [data['model1_pct'] for data in comparison_data]
model2_values = [data['model2_pct'] for data in comparison_data]
# Récupérer les couleurs des classes
colors = []
if models_info and 'class_colors' in models_info and 'class_names' in models_info:
model_class_names = models_info['class_names']
model_class_colors = models_info['class_colors']
for class_name in class_names:
try:
idx = model_class_names.index(class_name)
color = model_class_colors[idx]
hex_color = f"#{color[0]:02x}{color[1]:02x}{color[2]:02x}"
colors.append(hex_color)
except (ValueError, IndexError):
colors.append("#999999")
else:
colors = ['#999999'] * len(class_names)
# Créer le graphique comparatif
fig, ax = plt.subplots(figsize=(12, 7))
x = np.arange(len(class_names)) # Positions des barres
width = 0.35 # Largeur des barres
bars1 = ax.bar(x - width/2, model1_values, width, label=f"{model1.get('architecture', 'N/A')} - {model1['encoder_name']}",
color=colors, alpha=0.8, edgecolor='white')
# Barres hachurées pour le modèle 2
bars2 = ax.bar(x + width/2, model2_values, width, label=f"{model2.get('architecture', 'N/A')} - {model2['encoder_name']}",
color=colors, alpha=0.6, edgecolor='white', hatch='///')
ax.set_xlabel('Classes', fontsize=11, fontweight='bold')
ax.set_ylabel('Pourcentage (%)', fontsize=11, fontweight='bold')
ax.set_title('Comparaison des classes entre les deux modèles', fontsize=13, fontweight='bold', pad=15)
ax.set_xticks(x)
# Rotation des labels selon le nombre de classes
if len(class_names) <= 4:
ax.set_xticklabels(class_names, fontsize=10)
elif len(class_names) <= 6:
ax.set_xticklabels(class_names, rotation=30, ha='right', fontsize=9)
else:
ax.set_xticklabels(class_names, rotation=45, ha='right', fontsize=9)
ax.legend(loc='upper right', fontsize=10)
ax.grid(axis='y', alpha=0.2, linestyle='--')
ax.set_axisbelow(True)
# Style des axes
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
# Valeurs sur les barres
for bar1, bar2, val1, val2 in zip(bars1, bars2, model1_values, model2_values):
height1 = bar1.get_height()
height2 = bar2.get_height()
ax.text(bar1.get_x() + bar1.get_width()/2., height1 + 0.5,
f'{val1:.1f}%', ha='center', va='bottom', fontsize=8, fontweight='bold')
ax.text(bar2.get_x() + bar2.get_width()/2., height2 + 0.5,
f'{val2:.1f}%', ha='center', va='bottom', fontsize=8, fontweight='bold')
plt.tight_layout()
st.pyplot(fig)
plt.close(fig)
# Tableau détaillé avec un beau format
st.markdown("**🔎 Détails des différences par classe:**")
# Préparer les données pour le tableau
table_data = []
for i, data in enumerate(comparison_data):
class_color = colors[i]
diff_value = data['difference']
# Classification de la différence
if abs(diff_value) < 1:
diff_category = "Similaire"
diff_emoji = "🟢"
elif abs(diff_value) < 5:
diff_category = "Différence modérée"
diff_emoji = "🟡"
else:
diff_category = "Différence importante"
diff_emoji = "🔴"
table_data.append({
"Classe": data['class'],
f"🥇 {model1.get('architecture', 'N/A')[:15]} (%)": f"{data['model1_pct']:.1f}%",
f"🥈 {model2.get('architecture', 'N/A')[:15]} (%)": f"{data['model2_pct']:.1f}%",
"Différence": f"{diff_value:+.1f}%",
"Catégorie": f"{diff_emoji} {diff_category}"
})
# Créer le DataFrame et l'afficher
df = pd.DataFrame(table_data)
# Utiliser st.dataframe avec style personnalisé
st.dataframe(
df,
use_container_width=True,
hide_index=True,
column_config={
"Classe": st.column_config.TextColumn("🎯 Classe", width="medium"),
f"🥇 {model1.get('architecture', 'N/A')[:15]} (%)": st.column_config.TextColumn(width="medium"),
f"🥈 {model2.get('architecture', 'N/A')[:15]} (%)": st.column_config.TextColumn(width="medium"),
"Différence": st.column_config.TextColumn("📊 Différence", width="small"),
"Catégorie": st.column_config.TextColumn("📈 Évaluation", width="medium")
}
)
# Onglets principaux
main_tab1, main_tab2 = st.tabs(["📊 Présentation Jeu de données Cityscapes", "🔄 Comparaison de performances"])
# ===== ONGLET 1: PRÉSENTATION DATASET =====
with main_tab1:
st.markdown("## 📊 Analyse Exploratoire - Dataset Cityscapes")
# Informations générales sur le dataset
dataset_info = get_dataset_info()
if dataset_info:
# Section méthodologique
st.markdown("""
### 🎯 Note Méthodologique : Preuve de concept SegFormer vs FPN
#### Dataset retenu
**Cityscapes Dataset : Référence en Segmentation Urbaine**
Le **Cityscapes Dataset** constitue la référence internationale pour l'évaluation des algorithmes de segmentation sémantique en environnement urbain. Développé par l'Université de Tübingen et Mercedes-Benz, ce dataset comprend **25,000 images** haute résolution (2048×1024 pixels) capturées dans 50 villes allemandes et suisses.
""")
# Composition du dataset
col1, col2 = st.columns(2)
with col1:
st.markdown("**🗂️ Composition du dataset :**")
splits_info = dataset_info.get('splits', {})
for split_name, split_info in splits_info.items():
st.write(f"• **{split_name.title()}** : {split_info['images']} images - {split_info['usage']}")
with col2:
st.markdown("**🔬 Utilisation pour ce projet :**")
project_splits = dataset_info.get('project_splits', {})
for split_name, split_info in project_splits.items():
st.write(f"• **{split_name.title()}** : {split_info['images']} images ({split_info['percentage']}%) - {split_info['source']}")
# Architecture des classes
st.markdown("### 🏗️ Architecture de classes hiérarchique")
st.markdown("""
Le dataset organise 34 classes détaillées en **8 méta-classes** pour la segmentation :
| Méta-classe | Description | Importance |
|-------------|-------------|------------|
| **Flat** | Route, trottoir, parkings | Navigation de base |
| **Human** | Piéton, cycliste | **Sécurité critique** |
| **Vehicle** | Voiture, camion, bus, train | Obstacles dynamiques |
| **Construction** | Bâtiment, mur, clôture, pont | Structure urbaine |
| **Object** | Poteau, feu, panneau | Signalétique |
| **Nature** | Végétation, terrain | Environnement |
| **Sky** | Ciel | Contexte |
| **Void** | Zones non-définies | Exclusion |
""")
# Graphique de répartition des classes
st.markdown("### 📊 Répartition des classes dans le dataset d'entraînement")
class_repartition_data = get_class_repartition()
if class_repartition_data:
try:
class_repartition_image = Image.open(io.BytesIO(class_repartition_data))
st.image(
class_repartition_image,
caption="Distribution des classes dans les images d'entraînement Cityscapes",
use_container_width=True
)
except Exception as e:
st.warning("Impossible d'afficher le graphique de répartition des classes")
else:
st.info("Graphique de répartition des classes non disponible")
# Spécificités techniques
st.markdown("""
### ⚙️ Spécificités techniques
- **Résolutions d'acquisition** : Données natives 2048×1024 pixels
- **Conditions d'acquisition** : Conditions météo bonnes à moyennes, pas de conditions adverses (pluie, neige)
- **Diversité géographique** : Villes allemandes et suisses (architectures variées)
- **Annotations pixel-perfect** : Masques de segmentation précis au pixel près
Ce dataset permet d'évaluer la capacité des modèles à comprendre les scènes urbaines complexes, élément fondamental pour les applications de conduite autonome et de surveillance urbaine intelligente.
""")
# Section preprocessing des images
st.markdown("---")
st.markdown("### 🔧 Exemple de Preprocessing d'Images")
# Récupérer les images d'entraînement disponibles
train_data = get_train_images()
if train_data and train_data['images']:
# Sélecteur d'image
image_options = {img['display_name']: img['filename'] for img in train_data['images']}
selected_display_name = st.selectbox(
"Choisissez une image d'entraînement pour voir les transformations:",
options=list(image_options.keys()),
help=f"{train_data['total_count']} images disponibles"
)
selected_filename = image_options[selected_display_name]
# Options de preprocessing
col1, col2 = st.columns(2)
with col1:
show_augmentation = st.checkbox(
"Inclure l'augmentation aléatoire",
help="Affiche un exemple d'augmentation de données utilisée pendant l'entraînement"
)
with col2:
if st.button("🔧 Appliquer le Preprocessing", type="primary"):
with st.spinner("Application des transformations..."):
result = preprocess_train_image(selected_filename, show_augmentation)
if result and result['success']:
st.success("Preprocessing terminé!")
# Affichage des images transformées
st.markdown("#### Résultats des transformations :")
cols = st.columns(3 if show_augmentation else 2)
# Image originale
with cols[0]:
st.markdown("**Image Originale (redimensionnée)**")
if result.get('original_image'):
original_img_data = base64.b64decode(result['original_image'])
original_img = Image.open(io.BytesIO(original_img_data))
st.image(original_img, use_container_width=True)
# Image normalisée
with cols[1]:
st.markdown("**Après Normalisation**")
if result.get('normalized_image'):
normalized_img_data = base64.b64decode(result['normalized_image'])
normalized_img = Image.open(io.BytesIO(normalized_img_data))
st.image(normalized_img, use_container_width=True)
# Image avec augmentation (si demandée)
if show_augmentation and result.get('augmented_image'):
with cols[2]:
st.markdown("**Avec Augmentation Aléatoire**")
augmented_img_data = base64.b64decode(result['augmented_image'])
augmented_img = Image.open(io.BytesIO(augmented_img_data))
st.image(augmented_img, use_container_width=True)
# Explications des transformations
if result.get('transformations_applied'):
st.markdown("#### 📋 Transformations appliquées :")
for i, transform in enumerate(result['transformations_applied'], 1):
st.markdown(f"{i}. {transform}")
# Explications détaillées
with st.expander("ℹ️ Détails des transformations"):
st.markdown("""
**Normalisation standard :**
- Redimensionnement à 512×512 pixels
- Normalisation des valeurs de pixels de [0-255] vers [0-1]
- Cette normalisation uniforme est appliquée à tous les modèles (paramètre encoder non utilisé)
**Augmentation de données (si activée) :**
- Transformations aléatoires appliquées pendant l'entraînement
- Permet d'améliorer la robustesse et la généralisation du modèle
- Chaque image d'entraînement peut être transformée différemment à chaque époque
""")
else:
st.error("Erreur lors du preprocessing")
else:
st.warning("Aucune image d'entraînement disponible pour la démonstration")
# ===== ONGLET 2: COMPARAISON DE PERFORMANCES =====
with main_tab2:
# Variables de session pour les résultats
if "sample_comparison_result" not in st.session_state:
st.session_state.sample_comparison_result = None
if "upload_comparison_result" not in st.session_state:
st.session_state.upload_comparison_result = None
def reset_sample_comparison():
st.session_state.sample_comparison_result = None
def reset_upload_comparison():
st.session_state.upload_comparison_result = None
# Sous-onglets pour la comparaison
sub_tab1, sub_tab2 = st.tabs(["🏙️ Images Test Cityscapes", "📤 Upload Personnel"])
# SUB-TAB 1: Images Test Cityscapes
with sub_tab1:
st.markdown("## 🏙️ Comparaison sur Images Test Cityscapes")
st.markdown("Comparez les performances des deux modèles sur des images Cityscapes avec masques de vérité.")
# Récupérer les images d'exemple
sample_data = get_sample_images()
if sample_data and sample_data['images']:
# Sélecteur d'image
image_options = {img['display_name']: img['filename'] for img in sample_data['images']}
selected_display_name = st.selectbox(
"Choisissez une image:",
options=list(image_options.keys()),
help=f"{sample_data['total_count']} images disponibles"
)
selected_filename = image_options[selected_display_name]
# Afficher les infos de l'image sélectionnée
selected_img_info = next(img for img in sample_data['images'] if img['filename'] == selected_filename)
col1, col2 = st.columns([2, 1])
with col1:
st.info(f"📁 **Fichier:** {selected_filename}")
with col2:
ground_truth_status = "✅ Disponible" if selected_img_info['has_ground_truth'] else "❌ Non disponible"
st.info(f"🎯 **Vérité terrain:** {ground_truth_status}")
# Prévisualisation de l'image sélectionnée
if selected_filename:
try:
image_url = f"{API_BASE_URL}/sample-image/{selected_filename}"
response = requests.get(image_url, timeout=5)
if response.status_code == 200:
sample_image = Image.open(io.BytesIO(response.content))
st.image(
sample_image,
caption=f"Aperçu: {selected_display_name}",
use_container_width=True
)
except:
st.info("📷 Aperçu non disponible - L'image sera chargée lors de la comparaison")
# Bouton de comparaison
if st.button("🔄 Lancer la Comparaison", key="compare_sample", type="primary"):
if selected_filename:
with st.spinner("Comparaison en cours..."):
result = compare_sample_image(selected_filename)
if result and result['success']:
st.session_state.sample_comparison_result = result
st.success("✅ Comparaison terminée!")
st.rerun()
else:
st.error("❌ Erreur lors de la comparaison")
else:
st.warning("Aucune image d'exemple disponible")
# Affichage des résultats de comparaison
if st.session_state.sample_comparison_result:
st.markdown("---")
models_info = get_models_info()
display_comparison_results(st.session_state.sample_comparison_result, models_info)
# Bouton de reset
if st.button("🔄 Nouvelle Comparaison", key="reset_sample", type="secondary"):
reset_sample_comparison()
st.rerun()
# SUB-TAB 2: Upload Personnel
with sub_tab2:
st.markdown("## 📤 Comparaison sur Image Personnelle")
st.markdown("Uploadez une image pour comparer les performances des deux modèles.")
uploaded_file = st.file_uploader(
"Sélectionnez une image:",
type=['jpg', 'jpeg', 'png'],
help="Taille maximale recommandée: 5MB"
)
if uploaded_file is not None:
# Afficher l'image uploadée
image = Image.open(uploaded_file)
st.image(image, caption=f"Image uploadée: {uploaded_file.name}", use_container_width=True)
# Informations sur le fichier
file_size_mb = len(uploaded_file.getvalue()) / 1024 / 1024
st.info(f"📁 **Fichier:** {uploaded_file.name} | 📊 **Taille:** {file_size_mb:.2f} MB")
# Bouton de comparaison
if st.button("🔄 Lancer la Comparaison", key="compare_upload", type="primary"):
with st.spinner("Comparaison en cours..."):
result = compare_uploaded_image(uploaded_file)
if result and result['success']:
st.session_state.upload_comparison_result = result
st.success("✅ Comparaison terminée!")
st.rerun()
else:
st.error("❌ Erreur lors de la comparaison")
# Affichage des résultats de comparaison
if st.session_state.upload_comparison_result:
st.markdown("---")
models_info = get_models_info()
display_comparison_results(st.session_state.upload_comparison_result, models_info)
# Bouton de reset
if st.button("🔄 Nouvelle Comparaison", key="reset_upload", type="secondary"):
reset_upload_comparison()
st.rerun()
# Sidebar avec informations des modèles
with st.sidebar:
st.markdown("## 🔧 Informations des Modèles")
models_info = get_models_info()
if models_info:
st.success("✅ Modèles chargés")
# Modèle 1
best_model = models_info.get('best_model', {})
st.markdown("**🥇 Meilleur Modèle:**")
st.write(f"**Architecture:** {best_model.get('architecture', 'N/A')}")
st.write(f"**Encoder:** {best_model.get('encoder_name', 'N/A')}")
st.write(f"**Résolution:** {best_model.get('input_size', 'N/A')}")
# Modèle 2
second_model = models_info.get('second_best_model', {})
st.markdown("**🥈 Deuxième Modèle:**")
st.write(f"**Architecture:** {second_model.get('architecture', 'N/A')}")
st.write(f"**Encoder:** {second_model.get('encoder_name', 'N/A')}")
st.write(f"**Résolution:** {second_model.get('input_size', 'N/A')}")
else:
st.error("❌ Modèles non disponibles")
st.markdown("---")
st.markdown("## 🎯 Classes Segmentées (IoU)")
# Affichage avec carrés colorés et IoU
if models_info and 'class_names' in models_info and 'class_colors' in models_info:
class_names = models_info['class_names']
class_colors = models_info['class_colors']
# Récupérer les IoU des modèles
best_ious = best_model.get('class_ious', {}) if models_info else {}
second_ious = second_model.get('class_ious', {}) if models_info else {}
for i, (class_name, color) in enumerate(zip(class_names, class_colors)):
color_hex = f"#{color[0]:02x}{color[1]:02x}{color[2]:02x}"
# Construire la chaîne IoU
iou_str = ""
if best_ious.get(class_name) is not None and second_ious.get(class_name) is not None:
best_iou = best_ious[class_name]
second_iou = second_ious[class_name]
iou_str = f" (🥇{best_iou:.2f} - 🥈{second_iou:.2f})"
elif best_ious.get(class_name) is not None:
best_iou = best_ious[class_name]
iou_str = f" (🥇{best_iou:.2f})"
elif second_ious.get(class_name) is not None:
second_iou = second_ious[class_name]
iou_str = f" (🥈{second_iou:.2f})"
st.markdown(f"""
<div style="
display: flex;
align-items: center;
gap: 0.5rem;
margin: 0.3rem 0;
font-size: 0.85rem;
">
<div style="
width: 14px;
height: 14px;
background-color: {color_hex};
border-radius: 2px;
border: 1px solid rgba(0,0,0,0.2);
flex-shrink: 0;
"></div>
<span style="font-weight: 500;">{i}. {class_name}{iou_str}</span>
</div>
""", unsafe_allow_html=True)
# Footer
st.markdown("---")
st.markdown("""
<div style="text-align: center; color: #666; padding: 1rem;">
🚗 <strong>Street Vision - Comparaison de Modèles</strong> -
Évaluation comparative de modèles de segmentation sémantique<br>
<small>Développé avec FastAPI, Streamlit et PyTorch</small>
</div>
""", unsafe_allow_html=True)