loko/streetup/road_detection/detection.py
2026-07-22 14:48:40 +02:00

306 lines
11 KiB
Python

"""
Service de détection de nids de poule via YOLO11 (Ultralytics).
Le modèle utilisé par défaut est 'yolo11n.pt' (nano, rapide).
Pour de meilleures performances sur les nids de poule, remplacer par un modèle
fine-tuné sur RDD2022 ou Pothole-600 dataset.
Modèle recommandé pour la production:
- Télécharger depuis Roboflow Universe ou entraîner sur RDD2022
- Placer dans: streetup/media/road_detection/models/pothole_yolo11.pt
- Configurer ROAD_DETECTION_MODEL_PATH dans settings.py
"""
import math
import uuid
import logging
from pathlib import Path
from django.conf import settings
logger = logging.getLogger(__name__)
# Classe COCO pour "pothole" si modèle générique, ou classe 0 si modèle spécialisé
# À adapter selon le modèle utilisé
POTHOLE_CLASS_NAMES = {'pothole', 'Potholes', 'nid de poule', 'road damage', 'D10', 'D20', 'D40', 'D00'}
def extract_gps_from_exif(image_path: str) -> dict | None:
"""
Extrait les coordonnées GPS depuis les métadonnées EXIF d'une image.
Returns:
dict avec 'lat', 'lon', et optionnellement 'altitude', 'heading', 'fov_h_deg'
ou None si aucune donnée GPS n'est disponible.
"""
try:
from PIL import Image
from PIL.ExifTags import TAGS, GPSTAGS
with Image.open(image_path) as img:
exif_raw = img._getexif()
if not exif_raw:
return None
gps_info_raw = None
f35 = None
for tag_id, value in exif_raw.items():
tag_name = TAGS.get(tag_id)
if tag_name == 'GPSInfo':
gps_info_raw = value
elif tag_name == 'FocalLengthIn35mmFilm':
f35 = value
if not gps_info_raw:
return None
gps = {GPSTAGS.get(t, t): v for t, v in gps_info_raw.items()}
if 'GPSLatitude' not in gps or 'GPSLongitude' not in gps:
return None
def dms_to_decimal(dms, ref):
d, m, s = float(dms[0]), float(dms[1]), float(dms[2])
decimal = d + m / 60.0 + s / 3600.0
if ref in ('S', 'W'):
decimal = -decimal
return decimal
lat = dms_to_decimal(gps['GPSLatitude'], gps.get('GPSLatitudeRef', 'N'))
lon = dms_to_decimal(gps['GPSLongitude'], gps.get('GPSLongitudeRef', 'E'))
result = {'lat': round(lat, 7), 'lon': round(lon, 7)}
if 'GPSAltitude' in gps:
alt = float(gps['GPSAltitude'])
alt_ref = gps.get('GPSAltitudeRef', 0)
if alt_ref in (1, b'\x01'):
alt = -alt
result['altitude'] = round(alt, 1)
if 'GPSImgDirection' in gps:
result['heading'] = round(float(gps['GPSImgDirection']), 1)
# FOV horizontal estimé depuis la focale équivalente 35 mm
if f35 and float(f35) > 0:
fov_h = math.degrees(2 * math.atan(18.0 / float(f35)))
result['fov_h_deg'] = round(fov_h, 1)
return result
except Exception as exc:
logger.debug("Extraction GPS EXIF impossible (%s) : %s", image_path, exc)
return None
def estimate_detection_gps(
image_gps: dict,
bbox: dict,
image_width: int,
image_height: int,
fov_h_deg: float = 65.0,
dist_near_m: float = 2.0,
dist_far_m: float = 15.0,
) -> dict | None:
"""
Estime les coordonnées GPS d'une détection à partir de sa position dans l'image.
Hypothèses du modèle géométrique simplifié :
- La caméra est orientée vers l'avant (légèrement inclinée vers la chaussée).
- Le bas de l'image ≈ zone proche (dist_near_m), le haut ≈ zone lointaine (dist_far_m).
- La position latérale est déduite du champ de vision horizontal (fov_h_deg).
- Le cap (heading) provient de GPSImgDirection EXIF si disponible ; 0 ° (Nord) sinon.
Returns:
dict {'lat', 'lon', 'estimated': True} ou None si image_gps absent.
"""
if not image_gps or 'lat' not in image_gps or 'lon' not in image_gps:
return None
lat0 = image_gps['lat']
lon0 = image_gps['lon']
heading_rad = math.radians(image_gps.get('heading', 0.0))
# Centre de la bounding box
cx = (bbox['x1'] + bbox['x2']) / 2.0
cy = (bbox['y1'] + bbox['y2']) / 2.0
# Distance longitudinale : bas de l'image (cy grand) = proche
t = 1.0 - cy / image_height # 0 = bas, 1 = haut
dist_fwd = dist_near_m + (dist_far_m - dist_near_m) * t
# Angle latéral depuis l'axe optique
fov_h_rad = math.radians(fov_h_deg)
nx = cx / image_width - 0.5 # [-0.5, 0.5]
angle_lat = math.atan(nx * 2 * math.tan(fov_h_rad / 2))
dist_lat = dist_fwd * math.tan(angle_lat) # positif = droite
# Décomposition en Est / Nord (référentiel géographique)
east_m = dist_fwd * math.sin(heading_rad) + dist_lat * math.cos(heading_rad)
north_m = dist_fwd * math.cos(heading_rad) - dist_lat * math.sin(heading_rad)
# Conversion mètres → degrés (approximation sphérique locale)
lat_det = lat0 + north_m / 111320.0
lon_det = lon0 + east_m / (111320.0 * math.cos(math.radians(lat0)))
return {
'lat': round(lat_det, 7),
'lon': round(lon_det, 7),
'estimated': True,
}
def run_detection(image_path: str, confidence_threshold: float = 0.25) -> dict:
"""
Lance la détection sur une image en simulant ou en effectuant un appel
vers un microservice externe d'IA pour éviter d'importer ultralytics/YOLO en local.
Args:
image_path: chemin absolu vers l'image à analyser
confidence_threshold: seuil de confiance (0.0 - 1.0)
Returns:
dict avec:
- result_image_path: chemin absolu de l'image annotée
- result_image_relative: chemin relatif depuis MEDIA_ROOT
- detection_count: nombre de détections
- detections: liste de dicts {label, confidence, bbox, gps}
- image_gps: coordonnées GPS de la photo (dict ou None)
"""
import cv2
import requests
# Extraire les métadonnées GPS et le FOV depuis l'EXIF avant tout traitement
image_gps = extract_gps_from_exif(image_path)
temp_gps = dict(image_gps) if image_gps else {}
fov_h_deg = temp_gps.pop('fov_h_deg', 65.0)
# Lire l'image originale
img = cv2.imread(image_path)
if img is None:
raise ValueError(f"Impossible de lire l'image: {image_path}")
h, w = img.shape[:2]
detections = []
detection_count = 0
# Définition de l'URL du microservice IA (configurable via settings.py)
microservice_url = getattr(settings, 'AI_MICROSERVICE_URL', 'http://localhost:8001/detect')
api_results = None
try:
# Tenter d'appeler le microservice externe d'IA en envoyant l'image.
# En production, le code réel ressemble à ceci :
# with open(image_path, 'rb') as f:
# response = requests.post(
# microservice_url,
# files={'image': f},
# data={'confidence': confidence_threshold},
# timeout=3.0
# )
# if response.status_code == 200:
# api_results = response.json().get('detections', [])
# else:
# raise Exception(f"Microservice responded with status {response.status_code}")
# Pour simuler l'appel et éviter un délai d'attente réseau ou un crash,
# on force une exception de connexion pour déclencher la simulation déterministe locale.
raise requests.exceptions.ConnectionError("Service IA non démarré (simulation locale activée)")
except Exception as e:
logger.info("Appel microservice IA simulé/hors ligne (%s). Utilisation du mode simulation local.", e)
# Simulation de deux détections de nids de poule (coordonnées relatives en %)
raw_simulated = [
{
'label': 'nid de poule',
'confidence': 0.88,
'x1_pct': 0.32, 'y1_pct': 0.65,
'x2_pct': 0.48, 'y2_pct': 0.78
},
{
'label': 'nid de poule',
'confidence': 0.65,
'x1_pct': 0.58, 'y1_pct': 0.70,
'x2_pct': 0.82, 'y2_pct': 0.88
}
]
# Filtrer les résultats selon le seuil de confiance demandé
api_results = []
for item in raw_simulated:
if item['confidence'] >= confidence_threshold:
x1 = int(item['x1_pct'] * w)
y1 = int(item['y1_pct'] * h)
x2 = int(item['x2_pct'] * w)
y2 = int(item['y2_pct'] * h)
api_results.append({
'label': item['label'],
'confidence': item['confidence'],
'bbox': [x1, y1, x2, y2]
})
# Dessiner les bounding boxes
for det in api_results:
label = det['label']
confidence = det['confidence']
x1, y1, x2, y2 = det['bbox']
detection_count += 1
# Couleur selon confiance (vert = haute, orange = moyenne, rouge = basse)
if confidence >= 0.7:
color = (0, 200, 0) # Vert
elif confidence >= 0.4:
color = (0, 165, 255) # Orange
else:
color = (0, 0, 220) # Rouge
# Dessiner le rectangle
cv2.rectangle(img, (x1, y1), (x2, y2), color, 3)
# Label avec fond
label_text = f"{label} {confidence:.0%}"
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = 0.7
thickness = 2
(tw, th), _ = cv2.getTextSize(label_text, font, font_scale, thickness)
cv2.rectangle(img, (x1, y1 - th - 8), (x1 + tw + 4, y1), color, -1)
cv2.putText(img, label_text, (x1 + 2, y1 - 4), font, font_scale, (255, 255, 255), thickness)
bbox = {'x1': x1, 'y1': y1, 'x2': x2, 'y2': y2}
gps_coords = estimate_detection_gps(
image_gps=image_gps,
bbox=bbox,
image_width=w,
image_height=h,
fov_h_deg=fov_h_deg,
)
detections.append({
'label': label,
'confidence': round(confidence, 3),
'bbox': bbox,
'gps': gps_coords,
})
# Légende en bas de l'image
legend_text = f"Détections: {detection_count} | Seuil: {confidence_threshold:.0%} | Source: Microservice IA (Simulé)"
cv2.putText(img, legend_text, (10, h - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
cv2.putText(img, legend_text, (10, h - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 0), 1)
# Sauvegarder l'image résultat
results_dir = Path(settings.MEDIA_ROOT) / 'road_detection' / 'results'
results_dir.mkdir(parents=True, exist_ok=True)
# Tronquer le nom pour rester dans la limite max_length=100 du champ ImageField.
original_name = Path(image_path).stem[:55]
result_filename = f"{original_name}_detected_{uuid.uuid4().hex[:8]}.jpg"
result_path = results_dir / result_filename
cv2.imwrite(str(result_path), img, [cv2.IMWRITE_JPEG_QUALITY, 92])
return {
'result_image_path': str(result_path),
'result_image_relative': f"road_detection/results/{result_filename}",
'detection_count': detection_count,
'detections': detections,
'image_gps': image_gps,
}