Spaces:

VeuReu
/

engine

Sleeping

engine / scripts /initial_identity_encoding.py

Upload 4 files

c854a70 verified about 1 year ago

14.3 kB

# identity_encoding.py (updated to use libs/*)
# Veureu — Identity Encoder (faces, voices, scenarios)
# -----------------------------------------------------------------------------
# This script replaces the original `identity_encoding.py` but **reuses**
# as much as possible the functions already present in `libs/`.
# It respects the project's path structure (identities/*, scenarios, chroma_db,
# results) and maintains the classic pipeline:
#   1) index_faces (ChromaDB)
#   2) identity_features.csv
#   3) index_voices (ChromaDB)
#   4) scenarios_descriptions.csv
#   5) index_scenarios (ChromaDB)
# -----------------------------------------------------------------------------
from __future__ import annotations
import argparse
import csv
import logging
import sys
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple

# ============================ LOGGING ========================================
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger("identity_encoding")

# ============================ DEPENDENCIES ===================================
# ChromaDB (persistente)
try:
    import chromadb
    from chromadb.config import Settings  # noqa: F401
except Exception as e:
    chromadb = None  # type: ignore
    log.error("No se pudo importar chromadb: %s", e)

from libs.vision_tools_salamandra import FaceAnalyzer
from collections import Counter

# Audio: reuse get_embedding from the existing pipeline
from libs.audio_tools_ana_2 import VoiceEmbedder
from libs.vision_tools_salamandra import FaceOfImageEmbedding

# Optional
try:
    import numpy as np
except Exception:
    np = None  # type: ignore

# ============================ UTILITIES =====================================
IMG_EXT = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
AUD_EXT = {".wav", ".mp3", ".flac", ".m4a", ".ogg"}


def list_files(root: Path, exts: Iterable[str]) -> List[Path]:
    root = Path(root)
    if not root.exists():
        return []
    return [p for p in root.rglob('*') if p.suffix.lower() in exts]


def ensure_chroma(db_dir: Path):
    if chromadb is None:
        raise RuntimeError("chromadb no instalado. pip install chromadb")
    db_dir.mkdir(parents=True, exist_ok=True)
    
    # Nueva forma de crear un cliente persistente
    client = chromadb.Client(Settings(
        chroma_db_impl="duckdb+parquet",
        persist_directory=str(db_dir)
    ))
    return client

# ============================ 1) INDEX FACES =================================
def build_faces_index(faces_dir: Path, client, collection_name: str = "index_faces",
                      deepface_model: str = 'Facenet512', drop: bool = True) -> int:
    # idempotency
    if collection_name in [c.name for c in client.list_collections()] and drop:
        client.delete_collection(name=collection_name)
    col = client.get_or_create_collection(name=collection_name)

    be = FaceOfImageEmbedding(deepface_model=deepface_model)
    count = 0
    registered_identities = set()  # 👈 para no repetir nombres

    for ident_dir in sorted(Path(faces_dir).iterdir() if Path(faces_dir).exists() else []):
        if not ident_dir.is_dir():
            continue
        ident = ident_dir.name
        for img_path in list_files(ident_dir, IMG_EXT):
            embeddings = be.encode_image(img_path)
            if embeddings is None:
                log.warning("No face embedding in %s", img_path)
                continue

            # Aplanar para que cada embedding sea una lista de floats
            for e in (embeddings if isinstance(embeddings[0], list) else [embeddings]):
                uid = str(uuid.uuid4())
                col.add(ids=[uid], embeddings=[e], metadatas=[{"identity": ident, "path": str(img_path)}])
                count += 1
                registered_identities.add(ident)  # 👈 guardamos el nombre

    # Mensajes finales
    print("Ha acabado de crear la base de datos.")
    print(f"Total de embeddings guardados: {count}")
    print("Identidades registradas:")
    for name in sorted(registered_identities):
        print(f" - {name}")

    log.info("index_faces => %d embeddings", count)
    return count

# ===================== 2) IDENTITY FEATURES CSV ==============================

def aggregate_face_attributes(faces_dir: Path, out_csv: Path) -> int:
    """
    Procesa un directorio de caras por identidad y genera un CSV con edad y género.
    Usa FaceAnalyzer para extraer atributos.
    """
    # Inicializa el analizador
    from libs.vision_tools_salamandra import FaceAnalyzer
    analyzer = FaceAnalyzer()

    rows: List[Dict[str, Any]] = []

    faces_dir = Path(faces_dir)
    if not faces_dir.exists() or not faces_dir.is_dir():
        log.error("El directorio de caras no existe: %s", faces_dir)
        return 0

    def most_common(lst, default="unknown"):
        return Counter(lst).most_common(1)[0][0] if lst else default

    # Itera sobre cada identidad
    for ident_dir in sorted(faces_dir.iterdir()):
        if not ident_dir.is_dir():
            continue
        ident = ident_dir.name
        attrs: List[Dict[str, Any]] = []

        log.info("Procesando identidad: %s", ident)

        for img_path in sorted(list_files(ident_dir, IMG_EXT)):
            try:
                data = analyzer.analyze_image(str(img_path))
                if data:
                    attrs.append(data)
            except Exception as e:
                log.warning("Error procesando imagen %s: %s", img_path, e)

        genders = [a.get("gender", "unknown") for a in attrs]
        ages = [a.get("age", "unknown") for a in attrs]

        # Contexto opcional por identidad
        context_txt = (faces_dir.parent / "context" / f"{ident}.txt")
        identity_context = context_txt.read_text(encoding="utf-8").strip() if context_txt.exists() else ""

        rows.append({
            "identity": ident,
            "samples": len(attrs),
            "gender": most_common(genders),
            "age_bucket": most_common(ages),
            "identity_context": identity_context,
        })

        log.info("Procesados %d atributos para %s", len(attrs), ident)

    # Guardar CSV
    out_csv.parent.mkdir(parents=True, exist_ok=True)
    with out_csv.open("w", newline='', encoding="utf-8") as f:
        fieldnames = list(rows[0].keys()) if rows else ["identity", "identity_context"]
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    log.info("CSV generado correctamente: %s", out_csv)
    return len(rows)

# ============================ 3) INDEX VOICES =================================
from pydub import AudioSegment  # agregar al inicio de tu archivo junto a otros imports

def build_voices_index(voices_dir: Path, client, collection_name: str = "index_voices", drop: bool = True) -> int:
    if collection_name in [c.name for c in client.list_collections()] and drop:
        client.delete_collection(name=collection_name)
    col = client.get_or_create_collection(name=collection_name)

    ve = VoiceEmbedder()
    count = 0

    for ident_dir in sorted(Path(voices_dir).iterdir() if Path(voices_dir).exists() else []):
        if not ident_dir.is_dir():
            continue
        ident = ident_dir.name
        for wav_path in list_files(ident_dir, AUD_EXT):
            # Intentar embed directamente
            try:
                emb = ve.embed(wav_path)
            except Exception as e:
                log.warning("Error leyendo audio %s: %s. Intentando reconvertir...", wav_path, e)
                # Reconversión automática a WAV PCM
                try:
                    audio = AudioSegment.from_file(wav_path)
                    fixed_path = wav_path.with_name(wav_path.stem + "_fixed.wav")
                    audio.export(fixed_path, format="wav")
                    log.info("Archivo convertido a WAV compatible: %s", fixed_path)
                    emb = ve.embed(fixed_path)
                except Exception as e2:
                    log.error("No se pudo generar embedding tras reconversión para %s: %s", wav_path, e2)
                    continue  # saltar este archivo
            if emb is None:
                log.warning("No voice embedding en %s", wav_path)
                continue
            uid = str(uuid.uuid4())
            col.add(ids=[uid], embeddings=[emb], metadatas=[{"identity": ident, "path": str(wav_path)}])
            count += 1

    log.info("index_voices => %d embeddings", count)
    return count

# ============================ 4) SCENARIOS ==================================
@dataclass
class VisionClient:
    provider: str = "none"  # placeholder to plug in an LLM if desired

    def describe(self, image_path: str, prompt: str) -> str:
        return (f"Automatic description (placeholder) for {Path(image_path).name}. "
                f"{prompt}")


class TextEmbedder:
    """Text embeddings with Sentence-Transformers if available; fallback to TF-IDF."""
    def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
        self.kind = "tfidf"; self.model = None; self.vectorizer = None
        try:
            from sentence_transformers import SentenceTransformer
            self.model = SentenceTransformer(model_name)
            self.kind = "sbert"
        except Exception:
            from sklearn.feature_extraction.text import TfidfVectorizer
            self.vectorizer = TfidfVectorizer(max_features=768)

    def fit(self, texts: List[str]):
        if self.vectorizer is not None:
            self.vectorizer.fit(texts)

    def encode(self, texts: List[str]) -> List[List[float]]:
        if self.model is not None:
            arr = self.model.encode(texts, convert_to_numpy=True)
            return arr.astype(float).tolist()
        X = self.vectorizer.transform(texts) if self.vectorizer is not None else None
        return (X.toarray().astype(float).tolist() if X is not None else [[0.0]*128 for _ in texts])


def build_scenarios_descriptions(scenarios_dir: Path, out_csv: Path, vision: VisionClient,
                                 sample_per_scenario: int = 12) -> Tuple[int, List[Dict[str, Any]]]:
    rows: List[Dict[str, Any]] = []
    for scen_dir in sorted(Path(scenarios_dir).iterdir() if Path(scenarios_dir).exists() else []):
        if not scen_dir.is_dir():
            continue
        scen = scen_dir.name
        descs: List[str] = []
        imgs = list_files(scen_dir, IMG_EXT)[:sample_per_scenario]
        for img in imgs:
            d = vision.describe(str(img), prompt="Describe location, time period, lighting, and atmosphere without mentioning people or time of day.")
            if d:
                descs.append(d)
        if not descs:
            descs = [f"Scenario {scen} (no images)"]
        rows.append({"scenario": scen, "descriptions": " \n".join(descs)})

    out_csv.parent.mkdir(parents=True, exist_ok=True)
    with out_csv.open("w", newline='', encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["scenario", "descriptions"])
        w.writeheader(); w.writerows(rows)
    log.info("scenarios_descriptions => %s", out_csv)
    return len(rows), rows


def build_scenarios_index(client, rows: List[Dict[str, Any]], embedder: TextEmbedder,
                           collection_name: str = "index_scenarios", drop: bool = True) -> int:
    texts = [r["descriptions"] for r in rows]
    embedder.fit(texts)
    embs = embedder.encode(texts)

    if collection_name in [c.name for c in client.list_collections()] and drop:
        client.delete_collection(name=collection_name)
    col = client.get_or_create_collection(name=collection_name)

    for r, e in zip(rows, embs):
        col.add(ids=[r["scenario"]], embeddings=[e], metadatas=[{"scenario": r["scenario"]}])
    log.info("index_scenarios => %d descriptions", len(rows))
    return len(rows)

# ================================ CLI ========================================

def main():
    ap = argparse.ArgumentParser(description="Veureu — Build identity/scenario indices and CSVs")
    ap.add_argument('--faces_dir', default='identities/faces', help='Root directory of face images per identity')
    ap.add_argument('--voices_dir', default='identities/voices', help='Root directory of voice clips per identity')
    ap.add_argument('--scenarios_dir', default='scenarios', help='Root directory of scenario folders with images')
    ap.add_argument('--db_dir', default='chroma_db', help='ChromaDB persistence directory')
    ap.add_argument('--out_dir', default='results', help='Output directory for CSVs')
    ap.add_argument('--drop_collections', action='store_true', help='Delete collections if they exist before rebuilding')
    ap.add_argument('--deepface_model', default='Facenet512', help='DeepFace model to use as fallback')
    ap.add_argument('--scenario_samples', type=int, default=12, help='Number of images per scenario to describe')

    args = ap.parse_args()

    faces_dir = Path(args.faces_dir)
    voices_dir = Path(args.voices_dir)
    print(voices_dir)
    scenarios_dir = Path(args.scenarios_dir)
    out_dir = Path(args.out_dir); out_dir.mkdir(parents=True, exist_ok=True)

    client = ensure_chroma(Path(args.db_dir))

    # 1) Faces index
    build_faces_index(faces_dir, client, collection_name="index_faces", deepface_model=args.deepface_model, drop=args.drop_collections)

    # 2) Identity features CSV
    #id_csv = out_dir / 'identity_features.csv'
    #aggregate_face_attributes(faces_dir, id_csv)

    # 3) Voices index
    build_voices_index(voices_dir, client, collection_name="index_voices", drop=args.drop_collections)

    # 4) Scenarios descriptions
    #vision = VisionClient()
    #scen_csv = out_dir / 'scenarios_descriptions.csv'
    #_, scen_rows = build_scenarios_descriptions(scenarios_dir, scen_csv, vision, sample_per_scenario=args.scenario_samples)

    # 5) Scenarios index
    #embedder = TextEmbedder()
    #build_scenarios_index(client, scen_rows, embedder, collection_name="index_scenarios", drop=args.drop_collections)

    log.info("✅ Identity encoding completed.")


if __name__ == '__main__' and '--video' not in sys.argv:
    main()

	# identity_encoding.py (updated to use libs/*)
	# Veureu — Identity Encoder (faces, voices, scenarios)
	# -----------------------------------------------------------------------------
	# This script replaces the original `identity_encoding.py` but reuses
	# as much as possible the functions already present in `libs/`.
	# It respects the project's path structure (identities/*, scenarios, chroma_db,
	# results) and maintains the classic pipeline:
	# 1) index_faces (ChromaDB)
	# 2) identity_features.csv
	# 3) index_voices (ChromaDB)
	# 4) scenarios_descriptions.csv
	# 5) index_scenarios (ChromaDB)
	# -----------------------------------------------------------------------------
	from __future__ import annotations
	import argparse
	import csv
	import logging
	import sys
	import uuid
	from dataclasses import dataclass
	from pathlib import Path
	from typing import Any, Dict, Iterable, List, Optional, Tuple

	# ============================ LOGGING ========================================
	logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
	log = logging.getLogger("identity_encoding")

	# ============================ DEPENDENCIES ===================================
	# ChromaDB (persistente)
	try:
	import chromadb
	from chromadb.config import Settings # noqa: F401
	except Exception as e:
	chromadb = None # type: ignore
	log.error("No se pudo importar chromadb: %s", e)

	from libs.vision_tools_salamandra import FaceAnalyzer
	from collections import Counter

	# Audio: reuse get_embedding from the existing pipeline
	from libs.audio_tools_ana_2 import VoiceEmbedder
	from libs.vision_tools_salamandra import FaceOfImageEmbedding

	# Optional
	try:
	import numpy as np
	except Exception:
	np = None # type: ignore

	# ============================ UTILITIES =====================================
	IMG_EXT = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
	AUD_EXT = {".wav", ".mp3", ".flac", ".m4a", ".ogg"}


	def list_files(root: Path, exts: Iterable[str]) -> List[Path]:
	root = Path(root)
	if not root.exists():
	return []
	return [p for p in root.rglob('*') if p.suffix.lower() in exts]


	def ensure_chroma(db_dir: Path):
	if chromadb is None:
	raise RuntimeError("chromadb no instalado. pip install chromadb")
	db_dir.mkdir(parents=True, exist_ok=True)

	# Nueva forma de crear un cliente persistente
	client = chromadb.Client(Settings(
	chroma_db_impl="duckdb+parquet",
	persist_directory=str(db_dir)
	))
	return client

	# ============================ 1) INDEX FACES =================================
	def build_faces_index(faces_dir: Path, client, collection_name: str = "index_faces",
	deepface_model: str = 'Facenet512', drop: bool = True) -> int:
	# idempotency
	if collection_name in [c.name for c in client.list_collections()] and drop:
	client.delete_collection(name=collection_name)
	col = client.get_or_create_collection(name=collection_name)

	be = FaceOfImageEmbedding(deepface_model=deepface_model)
	count = 0
	registered_identities = set() # 👈 para no repetir nombres

	for ident_dir in sorted(Path(faces_dir).iterdir() if Path(faces_dir).exists() else []):
	if not ident_dir.is_dir():
	continue
	ident = ident_dir.name
	for img_path in list_files(ident_dir, IMG_EXT):
	embeddings = be.encode_image(img_path)
	if embeddings is None:
	log.warning("No face embedding in %s", img_path)
	continue

	# Aplanar para que cada embedding sea una lista de floats
	for e in (embeddings if isinstance(embeddings[0], list) else [embeddings]):
	uid = str(uuid.uuid4())
	col.add(ids=[uid], embeddings=[e], metadatas=[{"identity": ident, "path": str(img_path)}])
	count += 1
	registered_identities.add(ident) # 👈 guardamos el nombre

	# Mensajes finales
	print("Ha acabado de crear la base de datos.")
	print(f"Total de embeddings guardados: {count}")
	print("Identidades registradas:")
	for name in sorted(registered_identities):
	print(f" - {name}")

	log.info("index_faces => %d embeddings", count)
	return count

	# ===================== 2) IDENTITY FEATURES CSV ==============================

	def aggregate_face_attributes(faces_dir: Path, out_csv: Path) -> int:
	"""
	Procesa un directorio de caras por identidad y genera un CSV con edad y género.
	Usa FaceAnalyzer para extraer atributos.
	"""
	# Inicializa el analizador
	from libs.vision_tools_salamandra import FaceAnalyzer
	analyzer = FaceAnalyzer()

	rows: List[Dict[str, Any]] = []

	faces_dir = Path(faces_dir)
	if not faces_dir.exists() or not faces_dir.is_dir():
	log.error("El directorio de caras no existe: %s", faces_dir)
	return 0

	def most_common(lst, default="unknown"):
	return Counter(lst).most_common(1)[0][0] if lst else default

	# Itera sobre cada identidad
	for ident_dir in sorted(faces_dir.iterdir()):
	if not ident_dir.is_dir():
	continue
	ident = ident_dir.name
	attrs: List[Dict[str, Any]] = []

	log.info("Procesando identidad: %s", ident)

	for img_path in sorted(list_files(ident_dir, IMG_EXT)):
	try:
	data = analyzer.analyze_image(str(img_path))
	if data:
	attrs.append(data)
	except Exception as e:
	log.warning("Error procesando imagen %s: %s", img_path, e)

	genders = [a.get("gender", "unknown") for a in attrs]
	ages = [a.get("age", "unknown") for a in attrs]

	# Contexto opcional por identidad
	context_txt = (faces_dir.parent / "context" / f"{ident}.txt")
	identity_context = context_txt.read_text(encoding="utf-8").strip() if context_txt.exists() else ""

	rows.append({
	"identity": ident,
	"samples": len(attrs),
	"gender": most_common(genders),
	"age_bucket": most_common(ages),
	"identity_context": identity_context,
	})

	log.info("Procesados %d atributos para %s", len(attrs), ident)

	# Guardar CSV
	out_csv.parent.mkdir(parents=True, exist_ok=True)
	with out_csv.open("w", newline='', encoding="utf-8") as f:
	fieldnames = list(rows[0].keys()) if rows else ["identity", "identity_context"]
	writer = csv.DictWriter(f, fieldnames=fieldnames)
	writer.writeheader()
	writer.writerows(rows)

	log.info("CSV generado correctamente: %s", out_csv)
	return len(rows)

	# ============================ 3) INDEX VOICES =================================
	from pydub import AudioSegment # agregar al inicio de tu archivo junto a otros imports

	def build_voices_index(voices_dir: Path, client, collection_name: str = "index_voices", drop: bool = True) -> int:
	if collection_name in [c.name for c in client.list_collections()] and drop:
	client.delete_collection(name=collection_name)
	col = client.get_or_create_collection(name=collection_name)

	ve = VoiceEmbedder()
	count = 0

	for ident_dir in sorted(Path(voices_dir).iterdir() if Path(voices_dir).exists() else []):
	if not ident_dir.is_dir():
	continue
	ident = ident_dir.name
	for wav_path in list_files(ident_dir, AUD_EXT):
	# Intentar embed directamente
	try:
	emb = ve.embed(wav_path)
	except Exception as e:
	log.warning("Error leyendo audio %s: %s. Intentando reconvertir...", wav_path, e)
	# Reconversión automática a WAV PCM
	try:
	audio = AudioSegment.from_file(wav_path)
	fixed_path = wav_path.with_name(wav_path.stem + "_fixed.wav")
	audio.export(fixed_path, format="wav")
	log.info("Archivo convertido a WAV compatible: %s", fixed_path)
	emb = ve.embed(fixed_path)
	except Exception as e2:
	log.error("No se pudo generar embedding tras reconversión para %s: %s", wav_path, e2)
	continue # saltar este archivo
	if emb is None:
	log.warning("No voice embedding en %s", wav_path)
	continue
	uid = str(uuid.uuid4())
	col.add(ids=[uid], embeddings=[emb], metadatas=[{"identity": ident, "path": str(wav_path)}])
	count += 1

	log.info("index_voices => %d embeddings", count)
	return count

	# ============================ 4) SCENARIOS ==================================
	@dataclass
	class VisionClient:
	provider: str = "none" # placeholder to plug in an LLM if desired

	def describe(self, image_path: str, prompt: str) -> str:
	return (f"Automatic description (placeholder) for {Path(image_path).name}. "
	f"{prompt}")


	class TextEmbedder:
	"""Text embeddings with Sentence-Transformers if available; fallback to TF-IDF."""
	def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
	self.kind = "tfidf"; self.model = None; self.vectorizer = None
	try:
	from sentence_transformers import SentenceTransformer
	self.model = SentenceTransformer(model_name)
	self.kind = "sbert"
	except Exception:
	from sklearn.feature_extraction.text import TfidfVectorizer
	self.vectorizer = TfidfVectorizer(max_features=768)

	def fit(self, texts: List[str]):
	if self.vectorizer is not None:
	self.vectorizer.fit(texts)

	def encode(self, texts: List[str]) -> List[List[float]]:
	if self.model is not None:
	arr = self.model.encode(texts, convert_to_numpy=True)
	return arr.astype(float).tolist()
	X = self.vectorizer.transform(texts) if self.vectorizer is not None else None
	return (X.toarray().astype(float).tolist() if X is not None else [[0.0]*128 for _ in texts])


	def build_scenarios_descriptions(scenarios_dir: Path, out_csv: Path, vision: VisionClient,
	sample_per_scenario: int = 12) -> Tuple[int, List[Dict[str, Any]]]:
	rows: List[Dict[str, Any]] = []
	for scen_dir in sorted(Path(scenarios_dir).iterdir() if Path(scenarios_dir).exists() else []):
	if not scen_dir.is_dir():
	continue
	scen = scen_dir.name
	descs: List[str] = []
	imgs = list_files(scen_dir, IMG_EXT)[:sample_per_scenario]
	for img in imgs:
	d = vision.describe(str(img), prompt="Describe location, time period, lighting, and atmosphere without mentioning people or time of day.")
	if d:
	descs.append(d)
	if not descs:
	descs = [f"Scenario {scen} (no images)"]
	rows.append({"scenario": scen, "descriptions": " \n".join(descs)})

	out_csv.parent.mkdir(parents=True, exist_ok=True)
	with out_csv.open("w", newline='', encoding="utf-8") as f:
	w = csv.DictWriter(f, fieldnames=["scenario", "descriptions"])
	w.writeheader(); w.writerows(rows)
	log.info("scenarios_descriptions => %s", out_csv)
	return len(rows), rows


	def build_scenarios_index(client, rows: List[Dict[str, Any]], embedder: TextEmbedder,
	collection_name: str = "index_scenarios", drop: bool = True) -> int:
	texts = [r["descriptions"] for r in rows]
	embedder.fit(texts)
	embs = embedder.encode(texts)

	if collection_name in [c.name for c in client.list_collections()] and drop:
	client.delete_collection(name=collection_name)
	col = client.get_or_create_collection(name=collection_name)

	for r, e in zip(rows, embs):
	col.add(ids=[r["scenario"]], embeddings=[e], metadatas=[{"scenario": r["scenario"]}])
	log.info("index_scenarios => %d descriptions", len(rows))
	return len(rows)

	# ================================ CLI ========================================

	def main():
	ap = argparse.ArgumentParser(description="Veureu — Build identity/scenario indices and CSVs")
	ap.add_argument('--faces_dir', default='identities/faces', help='Root directory of face images per identity')
	ap.add_argument('--voices_dir', default='identities/voices', help='Root directory of voice clips per identity')
	ap.add_argument('--scenarios_dir', default='scenarios', help='Root directory of scenario folders with images')
	ap.add_argument('--db_dir', default='chroma_db', help='ChromaDB persistence directory')
	ap.add_argument('--out_dir', default='results', help='Output directory for CSVs')
	ap.add_argument('--drop_collections', action='store_true', help='Delete collections if they exist before rebuilding')
	ap.add_argument('--deepface_model', default='Facenet512', help='DeepFace model to use as fallback')
	ap.add_argument('--scenario_samples', type=int, default=12, help='Number of images per scenario to describe')

	args = ap.parse_args()

	faces_dir = Path(args.faces_dir)
	voices_dir = Path(args.voices_dir)
	print(voices_dir)
	scenarios_dir = Path(args.scenarios_dir)
	out_dir = Path(args.out_dir); out_dir.mkdir(parents=True, exist_ok=True)

	client = ensure_chroma(Path(args.db_dir))

	# 1) Faces index
	build_faces_index(faces_dir, client, collection_name="index_faces", deepface_model=args.deepface_model, drop=args.drop_collections)

	# 2) Identity features CSV
	#id_csv = out_dir / 'identity_features.csv'
	#aggregate_face_attributes(faces_dir, id_csv)

	# 3) Voices index
	build_voices_index(voices_dir, client, collection_name="index_voices", drop=args.drop_collections)

	# 4) Scenarios descriptions
	#vision = VisionClient()
	#scen_csv = out_dir / 'scenarios_descriptions.csv'
	#_, scen_rows = build_scenarios_descriptions(scenarios_dir, scen_csv, vision, sample_per_scenario=args.scenario_samples)

	# 5) Scenarios index
	#embedder = TextEmbedder()
	#build_scenarios_index(client, scen_rows, embedder, collection_name="index_scenarios", drop=args.drop_collections)

	log.info("✅ Identity encoding completed.")


	if __name__ == '__main__' and '--video' not in sys.argv:
	main()