import cv2 import time import json import tempfile import shutil from pathlib import Path import mediapipe as mp import numpy as np from mediapipe.tasks import python from mediapipe.tasks.python import vision from reachy_mini.utils import create_head_pose from scripts.generation_mouvement import trouver_emotion, calculer_cibles_antennes, adoucir_mouvement, calculer_danse_antennes PROJECT_ROOT = Path(__file__).resolve().parent.parent DATA_DIR = PROJECT_ROOT / "data" #Prépare et charge le modèle d'Intelligence Artificielle MediaPipe en utilisant le chemin du fichier modèle def initialiser_mediapipe(chemin_modele): base_options = python.BaseOptions(model_asset_path=chemin_modele) options = vision.FaceLandmarkerOptions( base_options=base_options, output_face_blendshapes=True, output_facial_transformation_matrixes=True, num_faces=1 ) return vision.FaceLandmarker.create_from_options(options) #Convertit les données brutes de MediaPipe en valeurs exploitables pour le robot en utilisant la trigo et les angles d'inclinaison def extraire_donnees_visage(matrice, scores): return { "pitch": float(np.arcsin(-matrice[1, 2])), "yaw": float(np.arctan2(matrice[0, 2], matrice[2, 2])), "roll": float(np.arctan2(matrice[1, 0], matrice[1, 1])), "head_z_mm": float(matrice[2, 3]), "eyebrow_raise": float((scores.get('browInnerUp', 0) + scores.get('browOuterUpLeft', 0) + scores.get('browOuterUpRight', 0)) / 3), "left_eyebrow": float(scores.get('browOuterUpLeft', 0) - scores.get('browDownLeft', 0)), "right_eyebrow": float(scores.get('browOuterUpRight', 0) - scores.get('browDownRight', 0)), "mouth_open": float(scores.get('jawOpen', 0)), "mouth_smile": float((scores.get('mouthSmileLeft', 0) + scores.get('mouthSmileRight', 0)) / 2), "mouth_pucker": float(scores.get('mouthPucker', 0)), "eye_openness": float((2.0 - scores.get('eyeBlinkLeft', 0) - scores.get('eyeBlinkRight', 0)) / 2), "left_eye_openness": float(1.0 - scores.get('eyeBlinkLeft', 0)), "right_eye_openness": float(1.0 - scores.get('eyeBlinkRight', 0)), "gaze_horizontal": float(scores.get('eyeLookOutLeft', 0) - scores.get('eyeLookInLeft', 0)), "gaze_vertical": float(scores.get('eyeLookUpLeft', 0) - scores.get('eyeLookDownLeft', 0)), "cheek_squint": float((scores.get('cheekSquintLeft', 0) + scores.get('cheekSquintRight', 0)) / 2), "brow_furrow": float((scores.get('browDownLeft', 0) + scores.get('browDownRight', 0)) / 2), "brow_inner_up": float(scores.get('browInnerUp', 0)), "brow_outer_up": float((scores.get('browOuterUpLeft', 0) + scores.get('browOuterUpRight', 0)) / 2), "mouth_frown": float((scores.get('mouthFrownLeft', 0) + scores.get('mouthFrownRight', 0)) / 2), "eye_wide": float((scores.get('eyeWideLeft', 0) + scores.get('eyeWideRight', 0)) / 2), "mouth_press": float((scores.get('mouthPressLeft', 0) + scores.get('mouthPressRight', 0)) / 2), "nose_sneer": float((scores.get('noseSneerLeft', 0) + scores.get('noseSneerRight', 0)) / 2), "mouth_lower_down": float((scores.get('mouthLowerDownLeft', 0) + scores.get('mouthLowerDownRight', 0)) / 2) } #Sauvegarde les données faciales et les trajectoires calculées sur le disque et creer deux fichier JSON un pour les expressions #faciales et l'autre pour les commandes de mouvement du robot def sauvegarder_fichiers(video_file, fps, frame_idx, visages_detectes, historique_signaux, historique_mouvements): if len(historique_signaux) == 0: return DATA_DIR.mkdir(parents=True, exist_ok=True) fichier_signaux = DATA_DIR / f"{video_file.stem}_signals.json" fichier_mouvements = DATA_DIR / f"{video_file.stem}_movement.json" output_signaux = { "description": "Video mimicry from source", "source_file": video_file.name, "duration": round(frame_idx / fps, 2) if fps > 0 else 0.0, "face_frames_total": frame_idx, "face_frames_detected": visages_detectes, "mapping_preset": "full", "face_data": historique_signaux } with open(fichier_signaux, 'w', encoding='utf-8') as f: json.dump(output_signaux, f, indent=4) with open(fichier_mouvements, 'w', encoding='utf-8') as f: json.dump({"mouvements": historique_mouvements}, f, indent=4) #Gère l'extraction de la piste audio en utilisant 'moviepy' en format .wav pour la compatibilité #avec le robot et la synchronisation puis fusionne le son avec la vidéo analysée def _fusionner_audio_video(chemin_video_traitee, chemin_video_source, chemin_sortie, chemin_audio): try: from moviepy import VideoFileClip video_clip = VideoFileClip(chemin_video_traitee) source_clip = VideoFileClip(chemin_video_source) if source_clip.audio is not None: audio = source_clip.audio if audio.duration > video_clip.duration: audio = audio.subclipped(0, video_clip.duration) final = video_clip.with_audio(audio) final.write_videofile(chemin_sortie, codec='libx264', audio_codec='aac', logger=None) audio.write_audiofile(chemin_audio, codec='pcm_s16le', logger=None) final.close() else: video_clip.write_videofile(chemin_sortie, codec='libx264', logger=None) video_clip.close() source_clip.close() except Exception as e: shutil.copy(chemin_video_traitee, chemin_sortie) #Analyse la vidéo image par image en utilisant opencv pour lire le flux et mediapipe pour detecter les visages et en déduire une #émotion, calcul la position des moteurs et dessines les points de repère faciaux (landmarks) def analyser_video(video_path, model_path, status=None, delai_danse=2.0): video_file = Path(video_path) DATA_DIR.mkdir(parents=True, exist_ok=True) detecteur_visage = initialiser_mediapipe(model_path) cap = cv2.VideoCapture(str(video_path)) fps = cap.get(cv2.CAP_PROP_FPS) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) w_orig = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h_orig = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) if w_orig > 1280: ratio = 1280 / w_orig out_w, out_h = 1280, int(h_orig * ratio) else: out_w, out_h = w_orig, h_orig # On écrit la vidéo sans son dans un fichier temporaire. tmp_video = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) tmp_video_path = tmp_video.name tmp_video.close() fourcc = cv2.VideoWriter_fourcc(*'mp4v') writer = cv2.VideoWriter(tmp_video_path, fourcc, fps, (out_w, out_h)) output_video = DATA_DIR / f"{video_file.stem}_processed.mp4" output_audio = DATA_DIR / f"{video_file.stem}_audio.wav" if status: status["phase"] = "analyzing" status["analysis_progress"] = 0 etat_robot = { "tete_p": 0.0, "tete_y": 0.0, "tete_r": 0.0, "ant_g_p": 0.0, "ant_g_r": 0.0, "ant_d_p": 0.0, "ant_d_r": 0.0 } #Il faut que l'émotion soit détectée sur plusieurs images consécutives pour être "validée". emotion_validee = "Neutre" emotion_en_cours = "Neutre" confirmations_emotion = 0 chronometre_emotion = 0.0 dernier_temps_visage = 0.0 historique_signaux = [] historique_mouvements = [] frame_idx = 0 visages_detectes = 0 try: while cap.isOpened(): if status and status.get("stop_requested"): break ret, frame = cap.read() if not ret: break h, w = frame.shape[:2] if w > 1280: ratio = 1280 / w frame = cv2.resize(frame, (1280, int(h * ratio)), interpolation=cv2.INTER_AREA) h, w = frame.shape[:2] # MediaPipe exige des images en format RGB, OpenCV lit par défaut en BGR. image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) resultat = detecteur_visage.detect(mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb)) temps_actuel = round(frame_idx / fps, 2) if fps > 0 else 0.0 donnees_frame = {"timestamp": temps_actuel, "face_detected": False} cible_tete_p, cible_tete_y, cible_tete_r = 0.0, 0.0, 0.0 cible_ag_p, cible_ag_r, cible_ad_p, cible_ad_r = 0.0, 0.0, 0.0, 0.0 if resultat.face_landmarks: visages_detectes += 1 donnees_frame["face_detected"] = True dernier_temps_visage = temps_actuel matrice = resultat.facial_transformation_matrixes[0] scores = {b.category_name: b.score for b in resultat.face_blendshapes[0]} traits_visage = extraire_donnees_visage(matrice, scores) donnees_frame.update(traits_visage) emotion_detectee = trouver_emotion(donnees_frame) #Il faut 5 frames identiques pour valider un changement d'état if emotion_detectee == emotion_en_cours: confirmations_emotion += 1 else: emotion_en_cours = emotion_detectee confirmations_emotion = 1 if confirmations_emotion >= 5 and emotion_en_cours != emotion_validee: emotion_validee = emotion_en_cours chronometre_emotion = temps_actuel # Calcul des trajectoires d'antennes selon l'émotion et le temps passé dans cette émotion temps_animation = temps_actuel - chronometre_emotion cible_ag_p, cible_ag_r, cible_ad_p, cible_ad_r = calculer_cibles_antennes(emotion_validee, temps_animation) cible_tete_p = traits_visage["pitch"] cible_tete_y = traits_visage["yaw"] cible_tete_r = traits_visage["roll"] #Dessin des points verts (landmarks) for landmark in resultat.face_landmarks[0]: cv2.circle(frame, (int(landmark.x * w), int(landmark.y * h)), 1, (0, 255, 0), -1) cv2.putText(frame, f"Emotion: {emotion_validee}", (30, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) else: # Si il n'y a pas de visage détécté le robot fais une danse d'attente if temps_actuel - dernier_temps_visage >= delai_danse: temps_animation = (temps_actuel - dernier_temps_visage) - delai_danse cible_ag_p, cible_ag_r, cible_ad_p, cible_ad_r = calculer_danse_antennes(temps_animation) # Sert a lisser les mouvements du robot etat_robot["tete_p"] = adoucir_mouvement(etat_robot["tete_p"], cible_tete_p, vitesse=0.6) etat_robot["tete_y"] = adoucir_mouvement(etat_robot["tete_y"], cible_tete_y, vitesse=0.6) etat_robot["tete_r"] = adoucir_mouvement(etat_robot["tete_r"], cible_tete_r, vitesse=0.6) etat_robot["ant_g_p"] = adoucir_mouvement(etat_robot["ant_g_p"], cible_ag_p, vitesse=0.5) etat_robot["ant_g_r"] = adoucir_mouvement(etat_robot["ant_g_r"], cible_ag_r, vitesse=0.5) etat_robot["ant_d_p"] = adoucir_mouvement(etat_robot["ant_d_p"], cible_ad_p, vitesse=0.5) etat_robot["ant_d_r"] = adoucir_mouvement(etat_robot["ant_d_r"], cible_ad_r, vitesse=0.5) historique_mouvements.append({ "timestamp": temps_actuel, "head": {"pitch": etat_robot["tete_p"], "yaw": etat_robot["tete_y"], "roll": etat_robot["tete_r"]}, "body_yaw": 0.0, "antennas": { "left": {"pitch": etat_robot["ant_g_p"], "roll": etat_robot["ant_g_r"]}, "right": {"pitch": etat_robot["ant_d_p"], "roll": etat_robot["ant_d_r"]} } }) historique_signaux.append(donnees_frame) writer.write(frame) frame_idx += 1 if total_frames > 0 and status: status["analysis_progress"] = int((frame_idx / total_frames) * 100) except Exception as e: if status: status["phase"] = "error" finally: #s'exécute toujours même s'il y a un crash ou un bouton stop writer.release() cap.release() if status and status.get("stop_requested"): try: Path(tmp_video_path).unlink() except Exception: pass status["phase"] = "ready" status["analysis_progress"] = 0 return if status: status["phase"] = "encoding" _fusionner_audio_video(tmp_video_path, str(video_path), str(output_video), str(output_audio)) try: Path(tmp_video_path).unlink() except Exception: pass sauvegarder_fichiers(video_file, fps, frame_idx, visages_detectes, historique_signaux, historique_mouvements) if status: status["phase"] = "ready" status["analysis_progress"] = 100 #Contrôle le robot pour lui faire rejouer l'animation en direct en ouvrant le fichier de mouvements généré précédement #et en synchronisant les mouvements avec le temps écoulé réel et la lecture de la vidéo def rejouer_mouvements(chemin_mouvement, robot, status=None): with open(chemin_mouvement, 'r', encoding='utf-8') as f: data = json.load(f) mouvements = data.get("mouvements", []) if not mouvements: return debut = time.time() temps_ecoule = 0 idx = 0 try: while idx < len(mouvements): if status and status.get("stop_requested"): break if status and status.get("playback_sync_flag"): status["playback_sync_flag"] = False temps_ecoule = status.get("playback_time_sync", temps_ecoule) debut = time.time() - temps_ecoule idx = 0 while idx < len(mouvements) and mouvements[idx]["timestamp"] < temps_ecoule: idx += 1 continue if status and status.get("playback_state") == "paused": time.sleep(0.05) debut = time.time() - temps_ecoule continue temps_ecoule = time.time() - debut if idx >= len(mouvements): break mvt = mouvements[idx] attente = mvt["timestamp"] - temps_ecoule if attente > 0: time.sleep(min(attente, 0.05)) continue pose_tete = create_head_pose( pitch=mvt["head"]["pitch"], yaw=mvt["head"]["yaw"], roll=mvt["head"]["roll"], degrees=False ) robot.set_target( head=pose_tete, antennas=[ mvt["antennas"]["left"]["pitch"] + mvt["antennas"]["left"]["roll"], mvt["antennas"]["right"]["pitch"] + mvt["antennas"]["right"]["roll"] ] ) idx += 1 finally: if status: status["stop_requested"] = False try: robot.goto_sleep() except: pass try: robot.disable_motors() except: pass