pluto
4bb609049a
Implements pack/unpack_embedding, cosine_similarity, find_best_speaker_match (max-per-speaker grouping, known/ambiguous/unknown thresholds), and embed_audio with lazy resemblyzer import so tests run without the docker-only dependency. 10 tests passing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
77 lines
2.1 KiB
Python
77 lines
2.1 KiB
Python
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
|