Download jericho/main.py from Jarilo-Project/Jericho: direct link, hf CLI and curl.
- Browser
- Download file 12.6 kB
-
https://huggingface.co/Jarilo-Project/Jericho/resolve/main/jericho/main.py
- Command line
-
hf download hf://Jarilo-Project/Jericho/jericho/main.py
-
curl -L -o main.py https://huggingface.co/Jarilo-Project/Jericho/resolve/main/jericho/main.py
12.6 kB
| 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) |