Ce projet présente une solution complète de détection d'objets sur le bord (edge AI), illustrant un pipeline technique de bout en bout allant de l'entraînement du modèle de deep learning à son déploiement sur un dispositif embarqué. Le système utilise YOLOv5 comme algorithme de détection principal, entraîné et déployé via le framework PyTorch, et déployé avec succès sur la plateforme de calcul embarquée Raspberry Pi 5 pour une détection en temps réel. Sur le plan technique, le projet a développé une chaîne d'outils complète de traitement des données, incluant un module d'acquisition d'images basé sur PiCamera2, un outil d'annotation interactif au format YOLO, ainsi qu'un pipeline d'entraînement de modèle intelligent. L'architecture du système est conçue de manière modulaire, avec un contrôle matériel via l'interface GPIO et une intégration d'une interface graphique Tkinter pour une expérience utilisateur conviviale.

Caractéristiques du système

• Détection intelligente : Basée sur l'algorithme de deep learning YOLOv5, identifiant avec précision les absences de plants dans les sillons de plantation.
• Alarme sonore et visuelle : Déclenchement automatique d'un buzzer et de LED en cas de détection d'absence.
• Interface conviviale : Fournit une interface graphique intuitive pour faciliter les opérations de l'utilisateur.
• Processus complet : Une solution intégrée de l'acquisition et l'annotation des données jusqu'à l'entraînement du modèle.
• Coût maîtrisé : Basé sur la plateforme matérielle Raspberry Pi, avec un coût contrôlé.

Architecture technique

Plate-forme matérielle :

• Contrôleur principal : Carte de développement Raspberry Pi 5 (processeur ARM haute performance).
• Capteur visuel : Caméra OV5647 (5 Mégapixels).
• Dispositif d'alarme : Buzzer actif + LED RGB.
• Connexion : Contrôle via broches GPIO.

Pile technologique principale :

• Framework de deep learning : PyTorch + YOLOv5
• Vision par ordinateur : OpenCV
• Contrôle matériel : RPi.GPIO + gpiozero
• Interface graphique : Tkinter
• Enterface caméra : PiCamera2

Détails de l'implémentation

1. Collecte et annnotation des données

1.1 Module de collecte de données (data\_collector.py)

Le collecteur de données contrôle la caméra Raspberry Pi via l'interface PiCamera2 pour capturer des images RGB en résolution 640x480 en temps réel.

class CollecteurDonnees:
    def __init__(self, repertoire_sortie="dataset"):
        # Création des répertoires de classification
        os.makedirs(os.path.join(repertoire_sortie, "absence"), exist_ok=True)
        os.makedirs(os.path.join(repertoire_sortie, "plant"), exist_ok=True)
        
        # Initialisation de la configuration de la caméra
        self.cam = Picamera2()
        self.cam.configure(self.cam.create_preview_configuration(
            main={"format": 'RGB888', "size": (640, 480)}))
        self.cam.start()

    def sauvegarder_image(self, image, etiquet):
        # Génération d'un nom de fichier horodaté pour garantir l'unicité
        horodatage = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
        nom_fichier = f"{horodatage}.jpg"
        chemin_sauvegarde = os.path.join(self.repertoire_sortie, etiquet, nom_fichier)
        cv2.imwrite(chemin_sauvegarde, image)

Fonctionnalités clés : prévisualisation en temps réel via OpenCV, contrôle par touches (m pour un échantillon de plant manquant, s pour un plant normal), nommage automatique avec horodatage, stockage classifié dans des répertoires.

1.2 Outil d'annotation de données (data\_labeler.py)

L'outil d'annotation met en œuvre une fonctionnalité complète d'annotation au format YOLO.

class AnnotateurDonnees:
    def rappel_souris(self, evenement, x, y, flags, param):
        """Fonction de rappel pour dessiner des rectangles de détection."""
        if evenement == cv2.EVENT_LBUTTONDOWN:
            self.en_dessin = True
            self.point_depart = (x, y)
        elif evenement == cv2.EVENT_LBUTTONUP and self.en_dessin:
            self.en_dessin = False
            self.point_fin = (x, y)
            # Dessiner le rectangle et sauvegarder l'annotation
            cv2.rectangle(self.image_courante, self.point_depart, self.point_fin, (0, 255, 0), 2)
            self.sauvegarder_annotation()

    def sauvegarder_annotation(self):
        """Sauvegarde l'annotation au format YOLO."""
        # Calcul des coordonnées normalisées
        h, w = self.image_courante.shape[:2]
        x_centre = ((x1 + x2) / 2) / w
        y_centre = ((y1 + y2) / 2) / h
        largeur = (x2 - x1) / w
        hauteur = (y2 - y1) / h
        
        # Écriture dans le fichier d'étiquettes au format YOLO
        with open(chemin_etiquette, 'a') as f:
            f.write(f"{self.id_classe} {x_centre:.6f} {y_centre:.6f} {largeur:.6f} {hauteur:.6f}\n")

Caractéristiques clés : annotation interactive par glisser-déposer de la souris, bascule de classe avec les touches 0/1 (plant manquant / plant normal), affectation au jeu d'entraînement/validation avec les touches t/v, génération automatique de fichiers d'annotation aux coordonnées normalisées au format YOLO.

2. Modèle de deep learning

2.1 Entraînement du modèle (train\_model.py)

Utilisation de YOLOv5 comme algorithme de détection d'objets. Le module d'entraînement implémente des optimisations spécifiques pour la Raspberry Pi.

def entrainer_modele(fichier_config, poids="yolov5s.pt", taille_image=640, taille_lot=16, epoques=100):
    """Fonction d'entraînement intelligente, adaptation automatique à l'environnement matériel."""
    
    # Détection de l'environnement Raspberry Pi
    est_raspberry_pi = False
    if os.path.exists('/proc/device-tree/model'):
        with open('/proc/device-tree/model', 'r') as f:
            est_raspberry_pi = 'raspberry pi' in f.read().lower()
    
    # Optimisation des paramètres pour Raspberry Pi
    if est_raspberry_pi:
        if taille_lot > 8:
            taille_lot = 8  # Réduction de la taille du lot pour éviter le débordement mémoire
        if taille_image > 416:
            taille_image = 416  # Réduction de la résolution de l'image pour améliorer la vitesse
    
    # Construction de la commande d'entraînement
    commande_entraînement = [
        sys.executable, f"{repertoire_yolov5}/train.py",
        "--img", str(taille_image),
        "--batch", str(taille_lot),
        "--epochs", str(epoques),
        "--data", fichier_config,
        "--weights", poids,
        "--cache"  # Activation du cache pour améliorer la vitesse d'entraînement
    ]
    
    # Exécution de l'entraînement
    subprocess.run(commande_entraînement, check=False)

2.2 Configuration du jeu de données (dataset/seedling.yaml)

Le fichier de configuration d'entraînement YOLOv5 définit la structure du jeu de données.

path: /home/pc/monitorage_plants/dataset
train: images/train
val: images/val
nc: 2
names: ["absence", "plant"]

Avantages techniques : architecture allégée (modèle YOLOv5s avec seulement 7.2M de paramètres, adapté aux périphériques de bord), entraînement de bout en bout (prise en charge du finetuning à partir d'un modèle pré-entraîné), optimisation automatique (ajustement automatique des paramètres d'entraînement selon l'environnement matériel), mécanisme de mise en cache (cache du jeu de données après le premier chargement pour améliorer l'efficacité de l'entraînement).

3. Système de surveillance en temps réel

3.1 Algorithme de détection principal (seedling\_monitor.py)

Le cœur du système de surveillance en temps réel est le moteur d'inférence YOLOv5, implémentant un pipeline de détection complet.

class MoniteurPlants:
    def __init__(self, chemin_modele, seuil_confiance=0.5):
        """Initialisation du système de surveillance."""
        # Initialisation de la caméra
        self.cam = Picamera2()
        self.cam.configure(self.cam.create_preview_configuration(
            main={"format": 'RGB888', "size": (640, 480)}))
        self.cam.start()
        
        # Chargement du modèle et recherche intelligente du chemin
        chemin_yolov5 = os.path.join(os.getcwd(), 'yolov5')
        sys.path.insert(0, chemin_yolov5)
        from models.experimental import attempt_load
        self.modele = attempt_load(chemin_modele, device='cpu')
        
        # Initialisation des périphériques GPIO
        self.buzzer = OutputDevice(17)
        self.led_rouge = OutputDevice(22)
        self.led_verte = OutputDevice(27)
        self.led_bleue = OutputDevice(23)

    def detecter_absence_plants(self, image):
        """Algorithme de détection principal."""
        # Prétraitement de l'image : conversion RGB en Tensor, normalisation
        tensor_image = torch.from_numpy(image).float()
        tensor_image = tensor_image.permute(2, 0, 1).unsqueeze(0) / 255.0
        
        # Inférence du modèle YOLOv5
        with torch.no_grad():
            resultats = self.modele(tensor_image)
        
        # Analyse des résultats et post-traitement
        detections = []
        if isinstance(resultats, tuple):
            prediction = resultats[0]
        else:
            prediction = resultats
            
        if prediction is not None and len(prediction) > 0:
            boites = prediction[0]
            for boite in boites:
                boite_np = boite.cpu().numpy()
                x1, y1, x2, y2, confiance, classe = boite_np[:6]
                
                if confiance >= self.seuil_confiance:
                    detections.append({
                        'xmin': float(x1), 'ymin': float(y1),
                        'xmax': float(x2), 'ymax': float(y2),
                        'confiance': float(confiance), 'classe': int(classe)
                    })
        
        # Statistiques des détections d'absence (classe 0 = absence)
        absence_detectee = [d for d in detections if d['classe'] == 0]
        return len(absence_detectee) > 0, resultats

3.2 Système d'alarme intelligent

Un mécanisme d'alarme multithread garantit l'exécution concurrente de la détection et de l'alarme.

def demarrer_alarme(self):
    """Déclenchement de l'alarme sonore et visuelle."""
    global alarme_active, thread_alarme
    if not alarme_active:
        alarme_active = True
        thread_alarme = Thread(target=self._boucle_alarme)
        thread_alarme.daemon = True
        thread_alarme.start()

def _boucle_alarme(self):
    """Boucle d'alarme s'exécutant dans un thread séparé."""
    while alarme_active:
        # Activation synchrone du buzzer et de la LED rouge
        self.buzzer.on()
        self.led_rouge.on()
        time.sleep(0.3)
        self.buzzer.off()
        self.led_rouge.off()
        time.sleep(0.3)

def afficher_resultats(self, image, resultats):
    """Visualisation des résultats de détection."""
    image_annotee = image.copy()
    # Dessin des cadres de détection, rouge pour absence, vert pour plant normal
    for detection in detections:
        couleur = (0, 0, 255) if detection['classe'] == 0 else (0, 255, 0)
        cv2.rectangle(image_annotee, 
                     (int(detection['xmin']), int(detection['ymin'])),
                     (int(detection['xmax']), int(detection['ymax'])), 
                     couleur, 2)
        # Ajout de l'étiquette de confiance
        etiquette = f"Classe {detection['classe']}: {detection['confiance']:.2f}"
        cv2.putText(image_annotee, etiquette, 
                   (int(detection['xmin']), int(detection['ymin']) - 10),
                   cv2.FONT_HERSHEY_SIMPLEX, 0.5, couleur, 2)
    return image_annotee

Caractéristiques techniques clés : traitement en temps réel (capture et traitement d'images de la caméra avec une latence inférieure à 100ms), architecture multithread (séparation des threads de détection et d'alarme pour garantir une réponse rapide), seuil adaptatif (seuil de confiance ajustable pour équilibre entre précision et rappel de la détection), gestion d'état (gestion intelligente de l'état de l'alarme pour éviter les déclenchements répétés).

4. Interface graphique utilisateur

4.1 Conception de l'interface principale (seedling\_ui.py)

Une interface multi-onglets construite avec Tkinter, fournissant un flux de travail complet.

class InterfaceMoniteurPlants:
    def __init__(self, racine):
        """Initialisation de l'interface utilisateur."""
        self.racine = racine
        self.racine.title("Système de surveillance intelligent d'absence de plants")
        self.racine.geometry("600x500")
        
        # Création du contrôle à onglets
        controle_onglets = ttk.Notebook(cadre_principal)
        
        # Trois modules fonctionnels
        self.onglet_donnees = ttk.Frame(controle_onglets)      # Traitement des données
        self.onglet_entraînement = ttk.Frame(controle_onglets) # Entraînement du modèle
        self.onglet_surveillance = ttk.Frame(controle_onglets) # Surveillance en temps réel
        
        controle_onglets.add(self.onglet_donnees, text="Traitement des données")
        controle_onglets.add(self.onglet_entraînement, text="Entraînement du modèle")
        controle_onglets.add(self.onglet_surveillance, text="Surveillance d'absence")

4.2 Gestion intelligente du chemin du modèle

Le système peut rechercher et identifier automatiquement le modèle d'entraînement le plus récent.

def rafraîchir_chemin_modele(self):
    """Recherche intelligente du modèle d'entraînement le plus récent."""
    try:
        repertoire_entraînement = os.path.join("yolov5", "runs", "train")
        if os.path.exists(repertoire_entraînement):
            # Obtention de tous les répertoires d'expérimentation
            repertoires_exp = [d for d in os.listdir(repertoire_entraînement) 
                       if os.path.isdir(os.path.join(repertoire_entraînement, d))]
            if repertoires_exp:
                # Tri par date de modification, obtention du répertoire le plus récent
                repertoire_le_plus_recent = sorted(repertoires_exp, 
                    key=lambda x: os.path.getmtime(os.path.join(repertoire_entraînement, x)), 
                    reverse=True)[0]
                chemin_poids = os.path.join(repertoire_entraînement, repertoire_le_plus_recent, "weights", "best.pt")
                if os.path.exists(chemin_poids):
                    self.chemin_modele.set(chemin_poids)
                    messagebox.showinfo("Succès", f"Modèle le plus récent trouvé : {repertoire_le_plus_recent}")
    except Exception as e:
        messagebox.showerror("Erreur", f"Erreur lors de la recherche du modèle : {str(e)}")

5. Conception de l'interface matérielle

5.1 Configuration des broches GPIO

Interface matérielle soigneusement conçue pour garantir la stabilité et l'extensibilité.

# Définition des broches GPIO
BROCHE_BUZZER = 17    # Buzzer - Broche supportant le PWM
BROCHE_ROUGE = 22     # LED RGB rouge - Sortie haut niveau
BROCHE_VERTE = 27     # LED RGB verte - Sortie haut niveau
BROCHE_BLEUE = 23     # LED RGB bleue - Sortie haut niveau

# Initialisation des périphériques
self.buzzer = OutputDevice(BROCHE_BUZZER)
self.led_rouge = OutputDevice(BROCHE_ROUGE)
self.led_verte = OutputDevice(BROCHE_VERTE)
self.led_bleue = OutputDevice(BROCHE_BLEUE)

Points forts de la conception matérielle : protection des circuits (toutes les sorties GPIO sont protégées par des résistances limitatrices de courant), standardisation de l'interface (conforme au standard GPIO de Raspberry Pi pour faciliter le débogage matériel), conception modulaire (encapsulation de la logique de contrôle matériel pour faciliter l'extension à d'autres capteurs), considération de faible consommation (paramétrage raisonnable des temps de sommeil pour réduire la consommation énergétique du système).

Optimisation des performances et détails techniques

1. Optimisation de l'inférence du modèle

1.1 Optimisation du traitement par lots dynamique

def optimiser_traitement_par_lots(self, file_images):
    """Optimisation de la vitesse d'inférence par traitement par lots dynamique."""
    if len(file_images) > 1:
        # Traitement par lots de plusieurs images
        lot_tensor = torch.stack([self.pretraiter(image) for image in file_images])
        with torch.no_grad():
            resultats_lot = self.modele(lot_tensor)
        return self.analyser_resultats_lot(resultats_lot)
    else:
        # Traitement d'une seule image
        return self.detecter_absence_plants(file_images[0])

1.2 Accélération par quantification du modèle

def quantifier_modele(self, chemin_modele):
    """Quantification du modèle pour réduire l'empreinte mémoire et améliorer la vitesse d'inférence."""
    modele = torch.load(chemin_modele, map_location='cpu')
    modele_quantifie = torch.quantization.quantize_dynamic(
        modele, {torch.nn.Linear, torch.nn.Conv2d}, dtype=torch.qint8)
    return modele_quantifie

2. Optimisation de l'architecture système

2.1 Modèle producteur-consommateur

class ProcesseurImages:
    def __init__(self):
        self.file_images = queue.Queue(maxsize=5)
        self.file_resultats = queue.Queue(maxsize=5)
    
    def producteur_images(self):
        """Thread de capture d'images."""
        while self.actif:
            image = self.cam.capture_array()
            if not self.file_images.full():
                self.file_images.put(image)
    
    def consommateur_images(self):
        """Thread de traitement d'images."""
        while self.actif:
            if not self.file_images.empty():
                image = self.file_images.get()
                resultat = self.detecter_absence_plants(image)
                self.file_resultats.put(resultat)

3. Optimisation de l'adaptation matérielle

3.1 Optimisations spécifiques à Raspberry Pi

def configurer_optimisation_raspberry_pi(self):
    """Configuration des optimisations spécifiques à Raspberry Pi."""
    # Ajustement de la fréquence CPU
    os.system("echo performance | sudo tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor")
    
    # Optimisation de l'allocation mémoire GPU
    os.system("sudo raspi-config nonint do_memory_split 128")
    
    # Désactivation des services non nécessaires
    os.system("sudo systemctl disable bluetooth")
    
    # Définit l'affinité des threads
    os.sched_setaffinity(0, {0, 1, 2, 3})  # Utilisation de tous les cœurs CPU

4. Mécanismes de gestion des errreurs et de tolérance aux pannes

4.1 Mécanisme de redémarrage intelligent

class MoniteurSysteme:
    def __init__(self):
        self.compteur_erreurs = 0
        self.max_erreurs = 5
        self.seuil_redemarrage = 10
    
    def gerer_erreur(self, erreur):
        """Gestion intelligente des erreurs."""
        self.compteur_erreurs += 1
        print(f"Erreur détectée : {erreur}, compteur d'erreurs : {self.compteur_erreurs}")
        
        if self.compteur_erreurs >= self.max_erreurs:
            self.redemarrer_systeme_detection()
            self.compteur_erreurs = 0
    
    def redemarrer_systeme_detection(self):
        """Redémarrage du système de détection."""
        print("Redémarrage du système de détection...")
        self.cam.arreter()
        time.sleep(2)
        self.cam.demarrer()
        # Rechargement du modèle
        self.modele = self.charger_modele(self.chemin_modele)

Concepts algorithmiques clés

1. Pipeline de détection YOLOv5

def pipeline_detection(self, image):
    """Pipeline de détection complet."""
    # 1. Prétraitement de l'image
    image_traitee = self.pretraiter(image)
    
    # 2. Inférence du modèle
    with torch.no_grad():
        predictions = self.modele(image_traitee)
    
    # 3. Post-traitement
    detections = self.posttraiter(predictions)
    
    # 4. Logique métier
    nombre_absences = sum(1 for det in detections if det['classe'] == 0)
    confiance_moyenne = np.mean([det['confiance'] for det in detections])
    
    return {
        'absence_detectee': nombre_absences > 0,
        'nombre_absences': nombre_absences,
        'confiance_moyenne': confiance_moyenne,
        'detections': detections
    }

2. Algorithme de seuil adaptatif

class SeuilAdaptatif:
    def __init__(self, seuil_initial=0.5):
        self.seuil = seuil_initial
        self.historique = []
        self.taux_faux_positifs = 0.0
    
    def mettre_a_jour_seuil(self, verite_terrain, predictions):
        """Ajustement dynamique du seuil basé sur les performances historiques."""
        # Calcul de la précision et du rappel avec le seuil actuel
        precision, rappel = self.calculer_metriques(verite_terrain, predictions)
        
        # Si le taux de faux positifs est trop élevé, augmenter le seuil
        if precision < 0.8:
            self.seuil = min(0.9, self.seuil + 0.05)
        # Si le taux de détection manquée est trop élevé, réduire le seuil
        elif rappel < 0.8:
            self.seuil = max(0.3, self.seuil - 0.05)