import struct import io import numpy as np # These are patched in tests; imported lazily at runtime to avoid docker-only dep try: from resemblyzer import VoiceEncoder, preprocess_wav except ImportError: VoiceEncoder = None preprocess_wav = None _encoder = None def get_encoder(): global _encoder if _encoder is None: if VoiceEncoder is None: raise ImportError("resemblyzer not installed") _encoder = VoiceEncoder() return _encoder def embed_audio(wav_bytes: bytes) -> np.ndarray: """Extract 256-dim d-vector from raw WAV bytes.""" wav = preprocess_wav(io.BytesIO(wav_bytes)) return get_encoder().embed_utterance(wav) def pack_embedding(embedding: np.ndarray) -> bytes: n = len(embedding) return struct.pack(f"{n}f", *embedding) def unpack_embedding(blob: bytes) -> np.ndarray: n = len(blob) // 4 return np.array(struct.unpack(f"{n}f", blob), dtype=np.float32) def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: norm_a = np.linalg.norm(a) norm_b = np.linalg.norm(b) if norm_a == 0 or norm_b == 0: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) def find_best_speaker_match( utterance_embedding: np.ndarray, stored_embeddings: list, known_threshold: float = 0.85, ambiguous_threshold: float = 0.60, ) -> tuple: """ Returns (match_status, speaker_id, confidence). Groups by speaker_id, takes MAX cosine similarity per speaker. """ if not stored_embeddings: return "unknown", None, None best_by_speaker: dict = {} for speaker_id, blob in stored_embeddings: ref = unpack_embedding(blob) sim = cosine_similarity(utterance_embedding, ref) if speaker_id not in best_by_speaker or sim > best_by_speaker[speaker_id]: best_by_speaker[speaker_id] = sim best_id = max(best_by_speaker, key=best_by_speaker.__getitem__) best_sim = best_by_speaker[best_id] if best_sim >= known_threshold: return "known", best_id, best_sim elif best_sim >= ambiguous_threshold: return "ambiguous", best_id, best_sim else: return "unknown", None, best_sim