commentaire main.py

reachy_mini_hello_world/main.py

@@ -14,16 +14,22 @@ from fastapi import UploadFile, File
 from fastapi.responses import FileResponse
 import yt_dlp
 from reachy_mini import ReachyMini, ReachyMiniApp
+
+# Import des scripts internes pour l'analyse, le rejeu et la génération de mouvements
 from scripts.play_json import analyser_video, rejouer_mouvements, extraire_donnees_visage
 from scripts.generation_mouvement import trouver_emotion, calculer_cibles_antennes, adoucir_mouvement, calculer_danse_antennes
 
+# CONFIGURATION DES CHEMINS ET DOSSIERS DU PROJET
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
 sys.path.append(str(PROJECT_ROOT))
 
-DATA_DIR = PROJECT_ROOT / "data"
-MODELS_DIR = PROJECT_ROOT / "models"
+DATA_DIR = PROJECT_ROOT / "data"        # Dossier de stockage des vidéos et JSON de mouvements
+MODELS_DIR = PROJECT_ROOT / "models"    # Dossier pour les modèles d'IA (MediaPipe)
 MODEL = MODELS_DIR / "face_landmarker_v2_with_blendshapes.task"
 
+# TÉLÉCHARGEMENT AUTOMATIQUE DU MODÈLE DE VISION (Sécurité / Zéro Configuration)
+# Si le fichier de détection de visage n'est pas présent localement,
+# l'application le télécharge automatiquement depuis les serveurs de Google.
 if not MODEL.exists():
     import urllib.request
     try:
@@ -37,6 +43,7 @@ if not MODEL.exists():
     except Exception as e:
         print(f"Erreur lors du téléchargement automatique du modèle : {e}")
 
+# Fonction utilitaire pour vérifier si une vidéo a déjà été analysée (fichiers prêts)
 def est_prete(filename):
     stem = Path(filename).stem
     return (
@@ -45,32 +52,36 @@ def est_prete(filename):
     )
 
 class ReachyMiniHelloWorld(ReachyMiniApp):
-    custom_app_url: str | None = "http://0.0.0.0:8080"
-    request_media_backend: str | None = "no_media"
+    # Configuration du serveur web embarqué
+    custom_app_url: str | None = "http://0.0.0.0:8080" # URL de notre dashboard de contrôle
+    request_media_backend: str | None = "no_media"     # Indique à Reachy qu'on gère nous-mêmes la caméra
 
-    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):
+    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):        
+        # État du pipeline de traitement de fichiers vidéo locaux/YouTube
         status = {
-            "robot_busy": False,
-            "analysis_busy": False,
-            "analysis_file": None,
-            "analysis_progress": 0,
-            "selected_video": None,
-            "playback_state": "playing",
-            "playback_time_sync": 0.0,
-            "playback_sync_flag": False,
-            "stop_requested": False
+            "robot_busy": False,          # True si le robot est en train de rejouer des mouvements
+            "analysis_busy": False,       # True si MediaPipe est en train d'analyser une vidéo
+            "analysis_file": None,        # Nom de la vidéo en cours d'analyse
+            "analysis_progress": 0,       # Barre de progression de l'analyse (0-100 %)
+            "selected_video": None,       # Vidéo actuellement sélectionnée
+            "playback_state": "playing",  # État de lecture (playing / paused)
+            "playback_time_sync": 0.0,    # Temps de lecture synchronisé avec le navigateur
+            "playback_sync_flag": False,  # Flag pour forcer la resynchronisation temporelle
+            "stop_requested": False       # True si l'utilisateur a cliqué sur "Stop"
         }
 
+        # État du pipeline de suivi en temps réel (Webcam Live)
         live = {
-            "active": False,
-            "detector": None,
-            "emotion": "Neutre",
-            "emotion_validee": "Neutre",
-            "emotion_en_cours": "Neutre",
-            "confirmations": 0,
-            "chronometre_emotion": 0.0,
-            "dernier_temps_visage": 0.0,
-            "temps_debut": 0.0,
+            "active": False,              # True si le mode live par webcam est actif
+            "detector": None,             # Instance active de l'IA FaceLandmarker de MediaPipe
+            "emotion": "Neutre",          # Émotion actuellement affichée par le robot
+            "emotion_validee": "Neutre",  # Émotion stable après confirmation temporelle
+            "emotion_en_cours": "Neutre", # Émotion détectée sur la frame actuelle
+            "confirmations": 0,           # Compteur de frames pour stabiliser l'émotion détectée
+            "chronometre_emotion": 0.0,   # Timestamp du début de l'émotion actuelle (pour l'animation)
+            "dernier_temps_visage": 0.0,  # Dernier instant où un visage a été détecté
+            "temps_debut": 0.0,           # Heure système au lancement du mode live
+            # État lissé des articulations (pitch, yaw, roll pour la tête et les antennes)
             "etat_robot": {
                 "tete_p": 0.0, "tete_y": 0.0, "tete_r": 0.0,
                 "ant_g_p": 0.0, "ant_g_r": 0.0,
@@ -78,6 +89,9 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             }
         }
 
+        # PIPELINES DE TRAITEMENT EN ARRIÈRE-PLAN 
+
+        # Pipeline 1 : Analyse complète d'un fichier vidéo en tâche de fond
         def pipeline_analyse(filename, delai_danse=2.0):
             status["stop_requested"] = False
             status["analysis_busy"] = True
@@ -86,6 +100,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             
             try:
                 video_cible = DATA_DIR / filename
+                # Appelle notre fonction de traitement d'image et d'extraction de blendshapes
                 analyser_video(str(video_cible), str(MODEL), status=status, delai_danse=delai_danse)
             except Exception as e:
                 pass
@@ -95,6 +110,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                 status["analysis_progress"] = 0
                 status["stop_requested"] = False
 
+        # Pipeline 2 : Rejeu physique des mouvements sur le robot
         def pipeline_replay():
             status["stop_requested"] = False
             status["robot_busy"] = True
@@ -104,6 +120,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             try:
                 stem = Path(status["selected_video"]).stem
                 movement_path = DATA_DIR / f"{stem}_movement.json"
+                # Envoie les consignes angulaires lissées et synchronisées au robot en temps réel
                 rejouer_mouvements(str(movement_path), reachy_mini, status=status)
             except Exception as e:
                 pass
@@ -111,6 +128,9 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                 status["robot_busy"] = False
                 status["stop_requested"] = False
 
+        # POINTS D'ENTRÉE API REST (FastAPI)
+
+        # Route 1 : Retourne les états globaux au Dashboard (barre de chargement, état d'occupation...)
         @self.settings_app.get("/status")
         async def get_status():
             is_ready = est_prete(status["selected_video"]) if status["selected_video"] else False
@@ -124,6 +144,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                 "live_active": live["active"]
             }
 
+        # Route 2 : Liste toutes les vidéos importées prêtes à l'analyse ou au rejeu
         @self.settings_app.get("/videos")
         async def lister_videos():
             if not DATA_DIR.exists():
@@ -134,6 +155,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                 if "_processed" not in f.name
             ]
 
+        # Route 3 : Sélectionne une vidéo spécifique dans l'interface
         @self.settings_app.post("/select_video")
         async def selectionner_video(filename: str):
             if not (DATA_DIR / filename).exists():
@@ -141,6 +163,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             status["selected_video"] = filename
             return {"selected": filename, "ready": est_prete(filename)}
 
+        # Route 4 : Lance l'analyse par IA (MediaPipe) en arrière-plan sans bloquer l'API
         @self.settings_app.post("/analyser")
         async def lancer_analyse(delai_danse_secondes: float = 2.0):
             if status["analysis_busy"]:
@@ -150,9 +173,11 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             if est_prete(status["selected_video"]):
                 return {"error": "Vidéo déjà analysée"}
                 
+            # Démarrage de l'analyse dans un thread séparé
             threading.Thread(target=pipeline_analyse, args=(status["selected_video"], delai_danse_secondes), daemon=True).start()
             return {"started": True, "delai_danse": delai_danse_secondes}
 
+        # Route 5 : Active les moteurs et lance le rejeu physique des expressions de la vidéo
         @self.settings_app.post("/run")
         async def run_pipeline():
             if status["robot_busy"]:
@@ -160,13 +185,16 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             if not est_prete(status["selected_video"]):
                 return {"error": "Vidéo non prête"}
 
+            # Phase d'initialisation du robot
             reachy_mini.wake_up()
             reachy_mini.enable_motors()
             await asyncio.sleep(0.5)
 
+            # Démarrage du rejeu synchrone dans un thread dédié
             threading.Thread(target=pipeline_replay, daemon=True).start()
             return {"started": True}
 
+        # Route 6 : Resynchronise la timeline vidéo du navigateur avec les moteurs du robot
         @self.settings_app.post("/sync_playback")
         async def sync_playback(state: str, current_time: float):
             status["playback_state"] = state
@@ -174,12 +202,14 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             status["playback_sync_flag"] = True
             return {"synced": True}
 
+        # Route 7 : Arrête immédiatement tous les mouvements du robot
         @self.settings_app.post("/stop")
         async def stop_pipeline():
             if status["robot_busy"]:
                 status["stop_requested"] = True
             return {"stopped": True}
 
+        # Route 8 : Gère le téléversement (upload) de vidéos MP4 et lance l'analyse automatique
         @self.settings_app.post("/upload_video")
         async def upload_video(file: UploadFile = File(...), delai_danse_secondes: float = 2.0):
             if not file.filename.lower().endswith(".mp4"):
@@ -198,12 +228,14 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
 
             status["selected_video"] = file.filename
 
+            # Si la vidéo est nouvelle, on lance l'analyse immédiatement
             if not est_prete(file.filename) and not status["analysis_busy"]:
                 threading.Thread(target=pipeline_analyse, args=(file.filename, delai_danse_secondes), daemon=True).start()
                 return {"uploaded": file.filename, "analyse_demarree": True, "delai_danse": delai_danse_secondes}
             
             return {"uploaded": file.filename, "analyse_demarree": False}
 
+        # Route 9 : Récupère le flux vidéo MP4 traité et annoté par notre IA pour l'affichage web
         @self.settings_app.get("/processed_video")
         async def get_processed_video():
             if not status["selected_video"]:
@@ -213,6 +245,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                 return FileResponse(str(video_path), media_type="video/mp4")
             return {"error": "Vidéo traitée non disponible"}
 
+        # Route 10 : Télécharge automatiquement une vidéo depuis YouTube via yt-dlp
         @self.settings_app.post("/download_youtube")
         async def download_youtube(data: dict):
             url = data.get('url')
@@ -235,6 +268,9 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             except Exception as e:
                 return {"success": False, "error": str(e)}
 
+        # MODULE WEBCAM EN DIRECT (Webcam Live Pipelines & Processing)
+
+        # Route 11 : Initialise l'IA de suivi de visage MediaPipe en mode dynamique et démarre le mode Live
         @self.settings_app.post("/live/start")
         async def live_start():
             if status["robot_busy"]:
@@ -242,10 +278,11 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             if live["active"]:
                 return {"error": "Mode live déjà actif"}
 
+            # Configuration avancée des options de suivi
             base_options = python.BaseOptions(model_asset_path=str(MODEL))
             options = vision.FaceLandmarkerOptions(
                 base_options=base_options,
-                running_mode=vision.RunningMode.VIDEO,
+                running_mode=vision.RunningMode.VIDEO, # Optimisé pour un flux d'images continu
                 output_face_blendshapes=True,
                 output_facial_transformation_matrixes=True,
                 num_faces=1,
@@ -255,6 +292,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             )
             live["detector"] = vision.FaceLandmarker.create_from_options(options)
 
+            # Réinitialisation de l'état temporel du suivi
             live.update({
                 "active": True,
                 "emotion": "Neutre",
@@ -278,34 +316,41 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             await asyncio.sleep(0.5)
             return {"started": True}
 
+        # Route 12 : Arrête le suivi par webcam et désactive les moteurs en toute sécurité
         @self.settings_app.post("/live/stop")
         async def live_stop():
             live["active"] = False
             live["detector"] = None
             status["robot_busy"] = False
             try:
+                # Retour au repos
                 reachy_mini.goto_sleep()
                 reachy_mini.disable_motors()
             except Exception:
                 pass
             return {"stopped": True}
 
+        # Route 13 : Reçoit chaque frame de la webcam capturée dans le navigateur,
+        # l'analyse par IA, calcule la pose 3D de la tête et anime le robot en direct.
         @self.settings_app.post("/live/frame")
         async def live_frame(file: UploadFile = File(...)):
             if not live["active"] or live["detector"] is None:
                 return {"error": "Mode live non actif"}
 
+            # 1. Décodage de l'image brute reçue en binaire
             contents = await file.read()
             nparr = np.frombuffer(contents, np.uint8)
             frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
             if frame is None:
                 return {"error": "Image invalide"}
             
+            # Réduction de la résolution de traitement pour minimiser la latence
             h, w = frame.shape[:2]
             if w > 960:
                 ratio = 960 / w
                 frame = cv2.resize(frame, (960, int(h * ratio)), interpolation=cv2.INTER_AREA)
 
+            # Amélioration du contraste en cas de faible éclairage
             lab = cv2.cvtColor(frame, cv2.COLOR_BGR2LAB)
             l_ch, a_ch, b_ch = cv2.split(lab)
             clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
@@ -314,6 +359,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
 
             image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
 
+            # 2. Détection IA de visage avec MediaPipe
             live["frame_timestamp_ms"] += 66
             resultat = live["detector"].detect_for_video(
                 mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb),
@@ -325,14 +371,17 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             cible_tete_p = cible_tete_y = cible_tete_r = 0.0
             cible_ag_p = cible_ag_r = cible_ad_p = cible_ad_r = 0.0
 
+            # Cas A : Un visage est détecté
             if resultat.face_landmarks:
                 live["dernier_temps_visage"] = temps_actuel
 
+                # Extraction de la matrice de transformation 3D et des coefficients musculaires (blendshapes)
                 matrice = resultat.facial_transformation_matrixes[0]
                 scores = {b.category_name: b.score for b in resultat.face_blendshapes[0]}
                 traits = extraire_donnees_visage(matrice, scores)
                 live["last_blendshapes"] = {k: round(v, 3) for k, v in traits.items()} 
 
+                # Détermination de l'émotion courante
                 emotion_detectee = trouver_emotion(traits)
                 if emotion_detectee == live["emotion_en_cours"]:
                     live["confirmations"] += 1
@@ -340,24 +389,32 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
                     live["emotion_en_cours"] = emotion_detectee
                     live["confirmations"] = 1
 
+                # Stabilisation de l'émotion (doit persister 3 frames pour être validée)
                 if live["confirmations"] >= 3 and live["emotion_en_cours"] != live["emotion_validee"]:
                     live["emotion_validee"] = live["emotion_en_cours"]
                     live["chronometre_emotion"] = temps_actuel
 
+                # Calcul des consigne pour les antennes selon l'émotion validée
                 temps_animation = temps_actuel - live["chronometre_emotion"]
                 cible_ag_p, cible_ag_r, cible_ad_p, cible_ad_r = calculer_cibles_antennes(
                     live["emotion_validee"], temps_animation
                 )
+                
+                # Consignes de pose 3D de la tête (angles de Pitch, Yaw et Roll)
                 cible_tete_p = traits["pitch"]
                 cible_tete_y = traits["yaw"]
                 cible_tete_r = traits["roll"]
                 live["emotion"] = live["emotion_validee"]
 
+            # Cas B : Aucun visage n'est détecté
             else:
                 delai = temps_actuel - live["dernier_temps_visage"]
+                # Après 2 secondes d'absence, le robot commence à "danser" avec ses antennes
                 if delai >= 2.0:
                     cible_ag_p, cible_ag_r, cible_ad_p, cible_ad_r = calculer_danse_antennes(delai - 2.0)
 
+            # 3. Lissage des mouvements (Interpolation linéaire / Lissage exponentiel)
+            # Permet d'éviter les saccades brusques du robot physique (préserve la mécanique)
             er["tete_p"] = adoucir_mouvement(er["tete_p"], cible_tete_p, 0.6)
             er["tete_y"] = adoucir_mouvement(er["tete_y"], cible_tete_y, 0.6)
             er["tete_r"] = adoucir_mouvement(er["tete_r"], cible_tete_r, 0.6)
@@ -366,6 +423,7 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
             er["ant_d_p"] = adoucir_mouvement(er["ant_d_p"], cible_ad_p, 0.5)
             er["ant_d_r"] = adoucir_mouvement(er["ant_d_r"], cible_ad_r, 0.5)
 
+            # 4. Envoi des angles à Reachy Mini
             pose_tete = create_head_pose(
                 pitch=er["tete_p"], yaw=er["tete_y"], roll=er["tete_r"], degrees=False
             )
@@ -379,13 +437,16 @@ class ReachyMiniHelloWorld(ReachyMiniApp):
 
             return {"emotion": live["emotion"]}
 
+        # Route 14 : Route utilitaire pour la calibration des muscles faciaux (Blendshapes)
         @self.settings_app.get("/live/debug")
         async def live_debug():
             """Retourne les dernières valeurs brutes des blendshapes pour calibration."""
             return live.get("last_blendshapes", {})
 
+        # Attente bloquante sur le thread principal pour maintenir les API FastAPI ouvertes
         stop_event.wait()
 
+# 4. SCRIPT D'EXÉCUTION 
 if __name__ == "__main__":
     app = ReachyMiniHelloWorld()
     try: