scripts/play_json.py

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