Upload 4 files

character_detection.py

@@ -16,18 +16,22 @@ from sklearn.cluster import DBSCAN
 import numpy as np
 from typing import List, Dict, Any, Tuple
 
-# Imports de las herramientas de vision y audio
-# Nota: Estos imports asumen que los archivos están en originales/
-# y que tienen las dependencias necesarias instaladas
+# Imports de las herramientas de vision y audio desde los módulos de la raíz
 try:
-    import sys
-    sys.path.insert(0, str(Path(__file__).parent / "originales"))
-    from vision_tools_salamandra_2 import FaceOfImageEmbedding_video_nuevo, ImageEmbedding, keyframe_conditional_extraction_ana
-    from audio_tools_ana_2 import extract_audio_ffmpeg, diarize_audio, embed_voice_segments
-    TOOLS_AVAILABLE = True
+    # Vision tools del engine (ya incluye DeepFace y face_recognition)
+    from vision_tools import FaceOfImageEmbedding
+    VISION_TOOLS_AVAILABLE = True
 except Exception as e:
-    TOOLS_AVAILABLE = False
-    logging.warning(f"No se pudieron importar las herramientas de Ana: {e}")
+    VISION_TOOLS_AVAILABLE = False
+    logging.warning(f"Vision tools no disponibles: {e}")
+
+try:
+    # Audio tools del engine
+    from audio_tools import extract_audio_ffmpeg_simple, diarize_with_pyannote, get_speaker_embeddings
+    AUDIO_TOOLS_AVAILABLE = True
+except Exception as e:
+    AUDIO_TOOLS_AVAILABLE = False
+    logging.warning(f"Audio tools no disponibles: {e}")
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -58,19 +62,21 @@ class CharacterDetector:
     
     def extract_faces_embeddings(self) -> List[Dict[str, Any]]:
         """
-        Extrae caras del vídeo y calcula sus embeddings.
-        Basado en faces_embedding_extraction de Ana.
+        Extrae caras del vídeo y calcula sus embeddings usando FaceOfImageEmbedding.
         
         Returns:
             Lista de dicts con {"embeddings": [...], "path": "..."}
         """
-        if not TOOLS_AVAILABLE:
-            logger.warning("Herramientas no disponibles, retornando lista vacía")
+        if not VISION_TOOLS_AVAILABLE:
+            logger.warning("Vision tools no disponibles, retornando lista vacía")
             return []
         
         logger.info("Extrayendo caras del vídeo...")
+        
+        # Inicializar el embedder (usa face_recognition o DeepFace automáticamente)
+        embedder = FaceOfImageEmbedding(deepface_model='Facenet512')
+        
         extract_every = 1.0  # segundos
-        embedder = FaceOfImageEmbedding_video_nuevo()
         video = cv2.VideoCapture(self.video_path)
         fps = int(video.get(cv2.CAP_PROP_FPS))
         frame_interval = int(fps * extract_every)
@@ -87,20 +93,40 @@ class CharacterDetector:
             if frame_count % frame_interval == 0:
                 temp_path = self.faces_dir / "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 = self.faces_dir / 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),
-                            "frame": frame_count
-                        })
-                        saved_count += 1
+                try:
+                    # Extraer embeddings usando FaceOfImageEmbedding
+                    # Devuelve una lista de embeddings (uno por cada cara detectada)
+                    embeddings_list = embedder.encode_image(temp_path)
+                    
+                    if embeddings_list:
+                        # Si es una lista de listas (múltiples caras)
+                        if isinstance(embeddings_list[0], list):
+                            for i, embedding in enumerate(embeddings_list):
+                                save_path = self.faces_dir / f"frame_{saved_count:04d}.jpg"
+                                # Guardar el frame completo (la extracción de cara ya se hizo internamente)
+                                cv2.imwrite(str(save_path), frame)
+                                
+                                embeddings_caras.append({
+                                    "embeddings": embedding,
+                                    "path": str(save_path),
+                                    "frame": frame_count
+                                })
+                                saved_count += 1
+                        else:
+                            # Si es un solo embedding
+                            save_path = self.faces_dir / f"frame_{saved_count:04d}.jpg"
+                            cv2.imwrite(str(save_path), frame)
+                            
+                            embeddings_caras.append({
+                                "embeddings": embeddings_list,
+                                "path": str(save_path),
+                                "frame": frame_count
+                            })
+                            saved_count += 1
+                
+                except Exception as e:
+                    logger.debug(f"No se detectaron caras en frame {frame_count}: {e}")
                 
                 if temp_path.exists():
                     os.remove(temp_path)
@@ -114,80 +140,24 @@ class CharacterDetector:
     def extract_voices_embeddings(self) -> List[Dict[str, Any]]:
         """
         Extrae voces del vídeo y calcula sus embeddings.
-        Basado en voices_embedding_extraction de Ana.
+        Por ahora retorna lista vacía (funcionalidad opcional).
         
         Returns:
             Lista de dicts con {"embeddings": [...], "path": "..."}
         """
-        if not TOOLS_AVAILABLE:
-            logger.warning("Herramientas no disponibles, retornando lista vacía")
-            return []
-        
-        logger.info("Extrayendo voces del vídeo...")
-        sr = 16000
-        fmt = "wav"
-        
-        wav_path = extract_audio_ffmpeg(
-            self.video_path,
-            self.voices_dir / f"{Path(self.video_path).stem}.{fmt}",
-            sr=sr
-        )
-        
-        min_dur = 0.5
-        max_dur = 10.0
-        
-        clip_paths, diar_segs = diarize_audio(
-            wav_path,
-            self.voices_dir,
-            "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])
-            })
-        
-        logger.info(f"Voces extraídas: {len(embeddings_voices)}")
-        return embeddings_voices
+        logger.info("Extracción de voces deshabilitada temporalmente")
+        return []
     
     def extract_scenes_embeddings(self) -> List[Dict[str, Any]]:
         """
-        Extrae escenas clave del vídeo y calcula sus embeddings.
-        Basado en scenes_embedding_extraction de Ana.
+        Extrae escenas clave del vídeo.
+        Por ahora retorna lista vacía (funcionalidad opcional).
         
         Returns:
             Lista de dicts con {"embeddings": [...], "path": "..."}
         """
-        if not TOOLS_AVAILABLE:
-            logger.warning("Herramientas no disponibles, retornando lista vacía")
-            return []
-        
-        logger.info("Extrayendo escenas del vídeo...")
-        keyframes_final = keyframe_conditional_extraction_ana(
-            video_path=self.video_path,
-            output_dir=self.scenes_dir,
-            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)
-            })
-        
-        logger.info(f"Escenas extraídas: {len(embeddings_escenas)}")
-        return embeddings_escenas
+        logger.info("Extracción de escenas deshabilitada temporalmente")
+        return []
     
     def cluster_faces(self, embeddings_caras: List[Dict], epsilon: float, min_samples: int) -> np.ndarray:
         """