import argparse import csv import math import os import subprocess import sys import urllib.request import cv2 import mediapipe as mp import numpy as np from mediapipe.tasks import python as mp_python from mediapipe.tasks.python import vision as mp_vision from mediapipe.tasks.python.vision import FaceLandmarkerOptions, FaceLandmarker # MediaPipe MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/1/face_landmarker.task" MODEL_PATH = "face_landmarker.task" def ensure_model(): if not os.path.exists(MODEL_PATH): print(f"Téléchargement du modèle MediaPipe -> {MODEL_PATH} ...") urllib.request.urlretrieve(MODEL_URL, MODEL_PATH) print(" Modele telecharge.") # Indices visage LEFT_EYE = [362, 385, 387, 263, 373, 380] # point qui contoure l oeil RIGHT_EYE = [33, 160, 158, 133, 153, 144] MOUTH_OUTER = [61, 291, 0, 17, 37, 267] MOUTH_L, MOUTH_R = 61, 291 LEFT_PUP, RIGHT_PUP = 468, 473 EAR_THRESHOLD = 0.20 # en dessous de ce seuil l'oeil considéré fermé #EAR = eye aspect ratio HEAD_POSE_IDX = [1, 152, 226, 446, 57, 287] MODEL_3D = np.array([# genere un model 3D d'une tete d'humain [ 0.0, 0.0, 0.0 ], [ 0.0, -63.6, -12.5], [-43.3, 32.7, -26.0], [ 43.3, 32.7, -26.0], [-28.9, -28.9, -24.1], [ 28.9, -28.9, -24.1], ], dtype=np.float64)#float 64 oblige a etre tres orecis # Calculs def _pts(lms, indices, w, h): # convertit les coordonnées en pixel return np.array([[lms[i].x * w, lms[i].y * h] for i in indices]) def ear(lms, indices, w, h): # ca calule le degres d'ouverture de l'oeil p = _pts(lms, indices, w, h)# recupere les pixels de l oeil A = np.linalg.norm(p[1]-p[5]); B = np.linalg.norm(p[2]-p[4]) # distance verticale des paupiere return (A+B) / (2*np.linalg.norm(p[0]-p[3])+1e-6) def mar(lms, indices, w, h):# calcul le degres d'ouverture de la bouche p = _pts(lms, indices, w, h) A = np.linalg.norm(p[1]-p[5]); B = np.linalg.norm(p[2]-p[4]) return (A+B) / (2*np.linalg.norm(p[0]-p[3])+1e-6) def smile_index(lms, w, h):# calcul a quel point la personne sourit ml = np.array([lms[MOUTH_L].x*w, lms[MOUTH_L].y*h])# coordonne coin de bouche droit et gauche mr = np.array([lms[MOUTH_R].x*w, lms[MOUTH_R].y*h]) pl = np.array([lms[LEFT_PUP].x*w, lms[LEFT_PUP].y*h])#coordonne pupilles pr = np.array([lms[RIGHT_PUP].x*w, lms[RIGHT_PUP].y*h]) return np.linalg.norm(mr-ml) / (np.linalg.norm(pr-pl)+1e-6)# divise largeur bouche par largeur pupille car elle se dilate en souriant def head_pose(lms, w, h): # fonction qui estime l'orientation de la tete img_pts = np.array([[lms[i].x*w, lms[i].y*h] for i in HEAD_POSE_IDX], dtype=np.float64)# extrait les coordonnées des points de references cam = np.array([[w,0,w/2],[0,w,h/2],[0,0,1]], dtype=np.float64) dist = np.zeros((4,1)) ok, rvec, _ = cv2.solvePnP(MODEL_3D, img_pts, cam, dist, flags=cv2.SOLVEPNP_ITERATIVE) if not ok: return 0.0, 0.0, 0.0 R, _ = cv2.Rodrigues(rvec) sy = math.sqrt(R[0,0]**2 + R[1,0]**2) if sy > 1e-6: roll = math.degrees(math.atan2( R[2,1], R[2,2])) pitch = math.degrees(math.atan2(-R[2,0], sy)) yaw = math.degrees(math.atan2( R[1,0], R[0,0])) else: roll = math.degrees(math.atan2(-R[1,2], R[1,1])) pitch = math.degrees(math.atan2(-R[2,0], sy)) yaw = 0.0 return yaw, pitch, roll def face_bbox(lms, w, h): # focntion pour calculer une boite qui enveloppe le visage xs = [lm.x for lm in lms]; ys = [lm.y for lm in lms]# ca creer une liste de tout les X et Y du visage xmn,xmx,ymn,ymx = min(xs),max(xs),min(ys),max(ys) return (xmn+xmx)/2,(ymn+ymx)/2,xmx-xmn,ymx-ymn,xmn,ymn,xmx,ymx # retourne tout les infos sur la boites # Annotation def annotate_frame(frame, lms, sig, w, h): # fonction qui va dessiner sur l video x1,y1 = int(sig["bbox_xmin"]*w), int(sig["bbox_ymin"]*h) x2,y2 = int(sig["bbox_xmax"]*w), int(sig["bbox_ymax"]*h) cv2.rectangle(frame,(x1,y1),(x2,y2),(0,255,0),2) # fait un carre vert autour du visage for idx in LEFT_EYE+RIGHT_EYE+MOUTH_OUTER: # dessine des points sur les differents endroit du visage cv2.circle(frame,(int(lms[idx].x*w),int(lms[idx].y*h)),2,(255,0,255),-1) for idx in [LEFT_PUP, RIGHT_PUP]: cv2.circle(frame,(int(lms[idx].x*w),int(lms[idx].y*h)),3,(0,200,255),-1) lines = [ # affiche les lignes texte qui contient les differents f"Yaw:{sig['yaw']:+.1f} Pitch:{sig['pitch']:+.1f} Roll:{sig['roll']:+.1f}", f"EAR L:{sig['ear_left']:.3f} R:{sig['ear_right']:.3f}", f"MAR:{sig['mar']:.3f} Smile:{sig['smile']:.3f}", f"T:{sig['timestamp_s']:.2f}s", ] for i, line in enumerate(lines):# ecrit les ligne de texte au dessus de la boite cv2.putText(frame, line, (x1, max(y1-10-i*18, 14)), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (0,255,0), 1, cv2.LINE_AA) nose = (int(lms[1].x*w), int(lms[1].y*h)) tip = (int(nose[0]+60*math.sin(math.radians(sig["yaw"]))), int(nose[1]-60*math.sin(math.radians(sig["pitch"])))) cv2.arrowedLine(frame, nose, tip, (0,100,255), 2, tipLength=0.3)# dessine feche partant du nez montrant orientation tete return frame # Fusion audio via FFmpeg def merge_audio(original_video, silent_video, output_path):# fonction qui reassemble le son de la video avec la video try: subprocess.run( ["ffmpeg", "-version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True ) except (subprocess.CalledProcessError, FileNotFoundError): print("\n [AVERTISSEMENT] ffmpeg introuvable dans le PATH.") print(" -> L'audio n'a pas ete recollé.") print(" Installez ffmpeg puis lancez manuellement :") print(f' ffmpeg -i "{silent_video}" -i "{original_video}" ' f'-c copy -map 0:v:0 -map 1:a:0 -shortest "{output_path}"') return False cmd = [ "ffmpeg", "-y", "-i", silent_video, # vidéo annotée (sans audio) "-i", original_video, # vidéo originale (source audio) "-c", "copy", # pas de ré-encodage "-map", "0:v:0", # flux vidéo de la vidéo annotée "-map", "1:a:0", # flux audio de l'originale "-shortest", # aligne sur la durée la plus courte output_path ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: print(f" -> Video finale avec audio : {output_path}") return True else: print(f" [ERREUR FFmpeg] {result.stderr[-300:]}") return False # Pipeline principal def process_video(video_path, output_dir): os.makedirs(output_dir, exist_ok=True) ensure_model() # telecharge le modele si il n'est pas deja present cap = cv2.VideoCapture(video_path) # ouvre le fichier video pour le lire image par image if not cap.isOpened(): sys.exit(f"[ERREUR] Impossible d'ouvrir : {video_path}") fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 # nombre d'image par seconde W = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) # dimension de la video H = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # nombre total d'image dans la video print(f"Video : {W}x{H} {fps:.1f} fps {total} frames") base_opts = mp_python.BaseOptions(model_asset_path=MODEL_PATH) # charge MediaPipe et configure les options du modele options = FaceLandmarkerOptions( base_options=base_opts, running_mode=mp_vision.RunningMode.VIDEO, num_faces=1, # cherche qu'un seul visage min_face_detection_confidence=0.5, min_face_presence_confidence=0.5, min_tracking_confidence=0.5, output_face_blendshapes=False, output_facial_transformation_matrixes=False, ) landmarker = FaceLandmarker.create_from_options(options) # cree le detecteur de visage # Fichier intermédiaire sans audio silent_path = os.path.join(output_dir, "_annotated_silent.mp4") final_path = os.path.join(output_dir, "output_annotated.mp4") writer = cv2.VideoWriter( silent_path, cv2.VideoWriter_fourcc(*"mp4v"), fps, (W, H) # ouvre un fichier de sortie pour ecrire les frames annotées sans audio ) records = [] # liste pour stocker les données extraites de chaque frame frame_idx = 0 # index de la frame actuelle while True: ret, frame = cap.read() # lit la frame suivante de la video if not ret: break ts_ms = int(frame_idx * 1000 / fps) # calcule le timestamp en millisecondes pour la frame actuelle ts_s = frame_idx / fps # calcule le timestamp en secondes pour la frame actuelle rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb) # convertit la frame en format RGB et crée une image MediaPipe à partir de celle-ci result = landmarker.detect_for_video(mp_image, ts_ms) # applique le modèle de détection de visage à l'image et obtient les résultats pour la frame actuelle if result.face_landmarks: # si un visage est détecté, extraire les informations et les stocker dans un dictionnaire lms = result.face_landmarks[0] cx,cy,bw,bh,xmn,ymn,xmx,ymx = face_bbox(lms, W, H) yaw,pitch,roll = head_pose(lms, W, H) el = ear(lms, LEFT_EYE, W, H) er = ear(lms, RIGHT_EYE, W, H) m = mar(lms, MOUTH_OUTER, W, H) sm = smile_index(lms, W, H) rec = { "frame":frame_idx, "timestamp_s":round(ts_s,4), "face_detected":True, "face_cx":round(cx,4), "face_cy":round(cy,4), "face_w":round(bw,4), "face_h":round(bh,4), "bbox_xmin":round(xmn,4), "bbox_ymin":round(ymn,4), "bbox_xmax":round(xmx,4), "bbox_ymax":round(ymx,4), "yaw":round(yaw,2), "pitch":round(pitch,2), "roll":round(roll,2), "ear_left":round(el,4), "ear_right":round(er,4), "eye_left_open": bool(el >= EAR_THRESHOLD), "eye_right_open": bool(er >= EAR_THRESHOLD), "mar":round(m,4), "smile":round(sm,4), } records.append(rec) frame = annotate_frame(frame, lms, rec, W, H) else: records.append({"frame":frame_idx,"timestamp_s":round(ts_s,4),"face_detected":False}) cv2.putText(frame,"Visage non detecte",(20,40), cv2.FONT_HERSHEY_SIMPLEX,0.8,(0,0,255),2) writer.write(frame) frame_idx += 1 if frame_idx % 100 == 0: pct = 100*frame_idx//total if total else 0 print(f" Frame {frame_idx}/{total} ({pct}%)") cap.release(); writer.release(); landmarker.close() # libère les ressources utilisées pour la lecture de la vidéo, l'écriture de la vidéo annotée et le modèle de détection de visage print(f" -> Video sans audio : {silent_path}") # ── Recoller l'audio avec FFmpeg ────────────── audio_ok = merge_audio(video_path, silent_path, final_path) if audio_ok: os.remove(silent_path) # supprime le fichier intermédiaire else: # FFmpeg absent : renommer le silencieux en sortie finale os.replace(silent_path, final_path) # ── CSV ────────────────────────────────────── csv_path = os.path.join(output_dir, "signals.csv") fields = ["frame","timestamp_s","face_detected","face_cx","face_cy","face_w","face_h", "bbox_xmin","bbox_ymin","bbox_xmax","bbox_ymax", "yaw","pitch","roll", "ear_left","ear_right","eye_left_open","eye_right_open", "mar","smile"] with open(csv_path,"w",newline="",encoding="utf-8") as f: wc = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") wc.writeheader(); wc.writerows(records) print(f" -> CSV : {csv_path}") print("\n Traitement termine.") print(f" Fichiers dans : {os.path.abspath(output_dir)}/") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--video", required=True) parser.add_argument("--output_dir", default="output") args = parser.parse_args() process_video(args.video, args.output_dir)