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>
107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
import numpy as np
|
|
import struct
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
def test_pack_unpack_roundtrip():
|
|
from app.pipeline.recognition import pack_embedding, unpack_embedding
|
|
original = np.array([0.1, 0.2, 0.3, -0.5], dtype=np.float32)
|
|
packed = pack_embedding(original)
|
|
recovered = unpack_embedding(packed)
|
|
np.testing.assert_allclose(recovered, original, rtol=1e-6)
|
|
|
|
|
|
def test_cosine_similarity_identical():
|
|
from app.pipeline.recognition import cosine_similarity
|
|
v = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
assert cosine_similarity(v, v) == pytest.approx(1.0)
|
|
|
|
|
|
def test_cosine_similarity_orthogonal():
|
|
from app.pipeline.recognition import cosine_similarity
|
|
a = np.array([1.0, 0.0], dtype=np.float32)
|
|
b = np.array([0.0, 1.0], dtype=np.float32)
|
|
assert cosine_similarity(a, b) == pytest.approx(0.0)
|
|
|
|
|
|
def test_cosine_similarity_zero_vector():
|
|
from app.pipeline.recognition import cosine_similarity
|
|
a = np.array([0.0, 0.0], dtype=np.float32)
|
|
b = np.array([1.0, 0.0], dtype=np.float32)
|
|
assert cosine_similarity(a, b) == 0.0
|
|
|
|
|
|
def test_find_best_match_known():
|
|
from app.pipeline.recognition import find_best_speaker_match, pack_embedding
|
|
ref = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
query = np.array([0.98, 0.2, 0.0], dtype=np.float32)
|
|
query /= np.linalg.norm(query)
|
|
ref /= np.linalg.norm(ref)
|
|
status, sid, conf = find_best_speaker_match(
|
|
query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60
|
|
)
|
|
assert status == "known"
|
|
assert sid == 1
|
|
assert conf >= 0.85
|
|
|
|
|
|
def test_find_best_match_ambiguous():
|
|
from app.pipeline.recognition import find_best_speaker_match, pack_embedding
|
|
ref = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
# cos similarity ~0.707 (45 degrees)
|
|
query = np.array([1.0, 1.0, 0.0], dtype=np.float32)
|
|
query /= np.linalg.norm(query)
|
|
ref /= np.linalg.norm(ref)
|
|
status, sid, conf = find_best_speaker_match(
|
|
query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60
|
|
)
|
|
assert status == "ambiguous"
|
|
assert sid == 1
|
|
|
|
|
|
def test_find_best_match_unknown():
|
|
from app.pipeline.recognition import find_best_speaker_match, pack_embedding
|
|
ref = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
query = np.array([0.0, 1.0, 0.0], dtype=np.float32) # orthogonal = 0 similarity
|
|
status, sid, conf = find_best_speaker_match(
|
|
query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60
|
|
)
|
|
assert status == "unknown"
|
|
assert sid is None
|
|
|
|
|
|
def test_find_best_match_empty_embeddings():
|
|
from app.pipeline.recognition import find_best_speaker_match
|
|
query = np.array([1.0, 0.0], dtype=np.float32)
|
|
status, sid, conf = find_best_speaker_match(query, [])
|
|
assert status == "unknown"
|
|
assert sid is None
|
|
|
|
|
|
def test_find_best_match_takes_max_per_speaker():
|
|
from app.pipeline.recognition import find_best_speaker_match, pack_embedding
|
|
# Speaker 1 has two embeddings — one poor, one good
|
|
ref_bad = np.array([0.0, 1.0, 0.0], dtype=np.float32)
|
|
ref_good = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
query = np.array([1.0, 0.0, 0.0], dtype=np.float32)
|
|
stored = [(1, pack_embedding(ref_bad)), (1, pack_embedding(ref_good))]
|
|
status, sid, conf = find_best_speaker_match(
|
|
query, stored, known_threshold=0.85, ambiguous_threshold=0.60
|
|
)
|
|
assert status == "known"
|
|
assert sid == 1
|
|
|
|
|
|
def test_embed_audio_calls_resemblyzer(mocker):
|
|
from app.pipeline.recognition import embed_audio
|
|
mock_wav = np.zeros(16000, dtype=np.float32)
|
|
mock_emb = np.ones(256, dtype=np.float32)
|
|
mocker.patch("app.pipeline.recognition.preprocess_wav", return_value=mock_wav)
|
|
mock_encoder = MagicMock()
|
|
mock_encoder.embed_utterance.return_value = mock_emb
|
|
mocker.patch("app.pipeline.recognition.get_encoder", return_value=mock_encoder)
|
|
result = embed_audio(b"\x00" * 100)
|
|
assert result.shape == (256,)
|
|
mock_encoder.embed_utterance.assert_called_once_with(mock_wav)
|