| import cv2 |
| import os |
| import json |
| import logging |
| from pathlib import Path |
|
|
| from libs.vision_tools_salamandra_2 import FaceOfImageEmbedding_video_nuevo |
| from libs.audio_tools_ana_2 import extract_audio_ffmpeg, diarize_audio, embed_voice_segments |
| from libs.vision_tools_salamandra_2 import ImageEmbedding, keyframe_conditional_extraction_ana |
|
|
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger(__name__) |
|
|
| def faces_embedding_extraction(video_path: str, output_dir_caras: Path): |
| extract_every = 1.0 |
| embedder = FaceOfImageEmbedding_video_nuevo() |
| video = cv2.VideoCapture(video_path) |
| fps = int(video.get(cv2.CAP_PROP_FPS)) |
| frame_interval = int(fps * extract_every) |
| frame_count = 0 |
| saved_count = 0 |
|
|
| embeddings_caras = [] |
|
|
| while True: |
| ret, frame = video.read() |
| if not ret: |
| break |
|
|
| if frame_count % frame_interval == 0: |
| temp_path = output_dir_caras / "temp_frame.jpg" |
| cv2.imwrite(str(temp_path), frame) |
| resultados = embedder.encode_image(temp_path) |
|
|
| if resultados: |
| for i, r in enumerate(resultados): |
| embedding = r['embedding'] |
| cara = r['face_crop'] |
| save_path = output_dir_caras / f"frame_{saved_count:04d}.jpg" |
| cv2.imwrite(str(save_path), cv2.cvtColor(cara, cv2.COLOR_RGB2BGR)) |
| embeddings_caras.append({"embeddings":embedding, "path": str(save_path)}) |
| saved_count += 1 |
|
|
| os.remove(temp_path) |
|
|
| frame_count += 1 |
| video.release() |
|
|
| return embeddings_caras |
|
|
| def voices_embedding_extraction(video_path: str, output_dir_audio: Path): |
| sr = 16000 |
| fmt = "wav" |
|
|
| wav_path = extract_audio_ffmpeg( |
| video_path, |
| output_dir_audio / f"{Path(video_path).stem}.{fmt}", |
| sr=sr |
| ) |
| min_dur = 0.5 |
| max_dur = 10.0 |
|
|
| clip_paths, diar_segs = diarize_audio( |
| wav_path, |
| output_dir_audio, |
| "clips", |
| min_dur, |
| max_dur |
| ) |
|
|
| embeddings_voices = [] |
|
|
| embeddings = embed_voice_segments(clip_paths) |
|
|
| for i, emb in enumerate(embeddings): |
| embeddings_voices.append({"embeddings": emb, "path": str(clip_paths[i])}) |
|
|
| return embeddings_voices |
|
|
| def scenes_embedding_extraction(video_path: str, output_dir_scenes: Path): |
| keyframes_final =keyframe_conditional_extraction_ana( |
| video_path=video_path, |
| output_dir=output_dir_scenes, |
| threshold=30.0, |
| ) |
|
|
| image_embedder = ImageEmbedding() |
|
|
| embeddings_escenas = [] |
|
|
| for keyframe in keyframes_final: |
| frame_path = keyframe["path"] |
| embedding = image_embedder.encode_image(frame_path) |
| embeddings_escenas.append({"embeddings": embedding, "path": str(frame_path)}) |
|
|
| return embeddings_escenas |
|
|
| video_path = "/home/acasado/bsc/proyecto_bsc/base_datos_dif_catala_1/dif_catala_1.mp4" |
| video_concreto = "dif_catala_1_2" |
| output_dir_caras = Path(f"/home/acasado/bsc/proyecto_bsc/{video_concreto}/frames") |
| output_dir_caras.mkdir(parents=True, exist_ok=True) |
| output_dir_audio = Path(f"/home/acasado/bsc/proyecto_bsc/{video_concreto}/audio") |
| output_dir_audio.mkdir(parents=True, exist_ok=True) |
| output_dir_escenas = Path(f"/home/acasado/bsc/proyecto_bsc/{video_concreto}/escenas") |
| output_dir_escenas.mkdir(parents=True, exist_ok=True) |
|
|
| embeddings_caras = faces_embedding_extraction(video_path, output_dir_caras) |
| embeddings_voices = voices_embedding_extraction(video_path, output_dir_audio) |
| embeddings_escenas = scenes_embedding_extraction(video_path, output_dir_escenas) |
|
|
| embeddings_finales = { |
| "caras": embeddings_caras, |
| "voices": embeddings_voices, |
| "escenas": embeddings_escenas |
| } |
|
|
| analysis_path = f"/home/acasado/bsc/proyecto_bsc/{video_concreto}/analysis.json" |
|
|
| try: |
| with open(analysis_path, "w", encoding="utf-8") as f: |
| json.dump(embeddings_finales, f, indent=2, ensure_ascii=False) |
| logger.info("Analysis JSON saved: %s", analysis_path) |
| except Exception as e: |
| logger.warning(f"Failed to write analysis JSON: {e}") |
|
|
