|
|
@@ -0,0 +1,867 @@
|
|
|
+"""Tests for seed-based assisted speaker enrollment.
|
|
|
+
|
|
|
+Hardware- and network-free: WAVs are synthesized with stdlib ``wave`` (a DC
|
|
|
+level encodes the "speaker id"), the embedder is a fake, reviews are scripted,
|
|
|
+and playback is a recording stub. The real sherpa-onnx package, questionary,
|
|
|
+and audio players are never required.
|
|
|
+"""
|
|
|
+
|
|
|
+import argparse
|
|
|
+import io
|
|
|
+import shutil
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+import wave
|
|
|
+from contextlib import redirect_stderr, redirect_stdout
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+import numpy as np
|
|
|
+import pytest
|
|
|
+
|
|
|
+import stt_daemon.cli as cli_module
|
|
|
+import stt_daemon.enroll_assist as ea
|
|
|
+from stt_daemon.config import load_config
|
|
|
+from stt_daemon.enroll import enroll_from_files
|
|
|
+from stt_daemon.enroll_assist import (
|
|
|
+ Candidate,
|
|
|
+ EnrollmentAborted,
|
|
|
+ QuestionaryReview,
|
|
|
+ play_wav,
|
|
|
+ rank_candidates,
|
|
|
+ review_candidates,
|
|
|
+ seed_enroll,
|
|
|
+)
|
|
|
+from stt_daemon.speakers import SpeakerStore
|
|
|
+
|
|
|
+A = 0.30
|
|
|
+B = -0.30
|
|
|
+
|
|
|
+
|
|
|
+def write_const_wav(path: Path, seconds: float, value: float,
|
|
|
+ sample_rate: int = 16000, nchannels: int = 1) -> Path:
|
|
|
+ """Write a constant-level int16 PCM WAV; the level encodes the speaker."""
|
|
|
+ count = int(seconds * sample_rate)
|
|
|
+ samples = np.full(count, value, dtype=np.float32)
|
|
|
+ frames = (samples * 32767.0).astype(np.int16)
|
|
|
+ if nchannels == 2:
|
|
|
+ frames = np.stack([frames, frames], axis=1).reshape(-1)
|
|
|
+ with wave.open(str(path), "wb") as handle:
|
|
|
+ handle.setnchannels(nchannels)
|
|
|
+ handle.setsampwidth(2)
|
|
|
+ handle.setframerate(sample_rate)
|
|
|
+ handle.writeframes(frames.tobytes())
|
|
|
+ return path
|
|
|
+
|
|
|
+
|
|
|
+class ConstEmbedder:
|
|
|
+ """Deterministic embedding from the clip DC level (the fake speaker id)."""
|
|
|
+
|
|
|
+ def __init__(self, dim: int = 2) -> None:
|
|
|
+ self.dim = dim
|
|
|
+ self.calls = 0
|
|
|
+
|
|
|
+ def embed(self, audio: np.ndarray) -> np.ndarray:
|
|
|
+ self.calls += 1
|
|
|
+ mean = float(np.mean(np.asarray(audio, dtype=np.float32)))
|
|
|
+ if self.dim == 2:
|
|
|
+ if mean > 0.15:
|
|
|
+ vector = [1.0, 0.0]
|
|
|
+ elif mean < -0.15:
|
|
|
+ vector = [0.0, 1.0]
|
|
|
+ else:
|
|
|
+ vector = [0.7071, 0.7071]
|
|
|
+ else:
|
|
|
+ vector = [1.0, 0.0, 0.0] if mean > 0.15 else [0.0, 1.0, 0.0]
|
|
|
+ return np.array(vector, dtype=np.float32)
|
|
|
+
|
|
|
+
|
|
|
+class VariedEmbedder:
|
|
|
+ """Same-speaker clips with distinct-but-similar embeddings.
|
|
|
+
|
|
|
+ Rotates the A-speaker vector 20 degrees per call, so same-speaker clips
|
|
|
+ score high (neighbors at cosine 0.94: above ``min_score``, below the
|
|
|
+ ``dup_threshold`` duplicate floor) instead of collapsing to exact
|
|
|
+ duplicates the way ConstEmbedder does. B-speaker clips map opposite the
|
|
|
+ seed so they are always filtered by ``min_score``. Tracks ``calls`` like
|
|
|
+ ConstEmbedder.
|
|
|
+ """
|
|
|
+
|
|
|
+ def __init__(self, dim: int = 2) -> None:
|
|
|
+ self.dim = dim
|
|
|
+ self.calls = 0
|
|
|
+
|
|
|
+ def embed(self, audio: np.ndarray) -> np.ndarray:
|
|
|
+ self.calls += 1
|
|
|
+ mean = float(np.mean(np.asarray(audio, dtype=np.float32)))
|
|
|
+ if mean < -0.15:
|
|
|
+ return np.array([-1.0, 0.0], dtype=np.float32)
|
|
|
+ angle = np.deg2rad(20.0 * self.calls)
|
|
|
+ vector = np.array([np.cos(angle), np.sin(angle)], dtype=np.float32)
|
|
|
+ return vector / np.linalg.norm(vector)
|
|
|
+
|
|
|
+
|
|
|
+class ScriptReview:
|
|
|
+ """Scripted ClipReview; records every choose() call."""
|
|
|
+
|
|
|
+ def __init__(self, actions) -> None:
|
|
|
+ self.actions = list(actions)
|
|
|
+ self.seen: list = []
|
|
|
+
|
|
|
+ def choose(self, candidate, accepted, target) -> str:
|
|
|
+ self.seen.append((candidate, accepted, target))
|
|
|
+ if not self.actions:
|
|
|
+ return "skip"
|
|
|
+ return self.actions.pop(0)
|
|
|
+
|
|
|
+
|
|
|
+def write_cfg(tmp_path: Path, *, min_enroll_s: float = 2.0,
|
|
|
+ max_prototypes: int = 8, model_ok: bool = True):
|
|
|
+ """Write a temp config; return (cfg, audio_dir, store_dir, model)."""
|
|
|
+ model = tmp_path / "model.onnx"
|
|
|
+ if model_ok:
|
|
|
+ model.write_bytes(b"fake-model")
|
|
|
+ audio_dir = tmp_path / "audio"
|
|
|
+ audio_dir.mkdir(exist_ok=True)
|
|
|
+ store_dir = tmp_path / "speakers"
|
|
|
+ cfg_path = tmp_path / "config.toml"
|
|
|
+ cfg_path.write_text(
|
|
|
+ "[speaker]\n"
|
|
|
+ f'model_path = "{model}"\n'
|
|
|
+ f"min_enroll_s = {min_enroll_s}\n"
|
|
|
+ f"max_prototypes = {max_prototypes}\n"
|
|
|
+ f'store_dir = "{store_dir}"\n'
|
|
|
+ "[output]\n"
|
|
|
+ f'audio_dir = "{audio_dir}"\n'
|
|
|
+ "[events]\n"
|
|
|
+ f'socket = "{tmp_path / "bus.sock"}"\n',
|
|
|
+ encoding="utf-8",
|
|
|
+ )
|
|
|
+ return load_config(str(cfg_path)), audio_dir, store_dir
|
|
|
+
|
|
|
+
|
|
|
+def make_store(cfg) -> SpeakerStore:
|
|
|
+ """Build a SpeakerStore honoring cfg (mirrors the CLI handler)."""
|
|
|
+ speaker_cfg = cfg.speaker
|
|
|
+ return SpeakerStore(
|
|
|
+ Path(speaker_cfg.store_dir),
|
|
|
+ threshold_id=speaker_cfg.threshold_id,
|
|
|
+ threshold_new=speaker_cfg.threshold_new,
|
|
|
+ min_enroll_s=speaker_cfg.min_enroll_s,
|
|
|
+ rms_floor=speaker_cfg.rms_floor,
|
|
|
+ margin=speaker_cfg.margin,
|
|
|
+ mint_cooldown_s=speaker_cfg.mint_cooldown_s,
|
|
|
+ confirm_s=speaker_cfg.confirm_s,
|
|
|
+ max_prototypes=speaker_cfg.max_prototypes,
|
|
|
+ provisional_ttl_h=speaker_cfg.provisional_ttl_h,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def record_enroll(monkeypatch):
|
|
|
+ """Wrap ea.enroll_from_files to record enrolled paths; return the log."""
|
|
|
+ called: list = []
|
|
|
+ real = ea.enroll_from_files
|
|
|
+
|
|
|
+ def fake(store, embedder, name, paths, min_s):
|
|
|
+ called.append(list(paths))
|
|
|
+ return real(store, embedder, name, paths, min_s)
|
|
|
+
|
|
|
+ monkeypatch.setattr(ea, "enroll_from_files", fake)
|
|
|
+ return called
|
|
|
+
|
|
|
+
|
|
|
+def store_files(store_dir: Path) -> bool:
|
|
|
+ return (store_dir / "speakers.json").exists()
|
|
|
+
|
|
|
+
|
|
|
+def fresh_speakers(store_dir: Path) -> dict:
|
|
|
+ store = SpeakerStore(store_dir)
|
|
|
+ store.load()
|
|
|
+ return dict(store.speakers)
|
|
|
+
|
|
|
+
|
|
|
+# -- rank_candidates ------------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_orders_by_score_and_counts_skipped(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ good_a = write_const_wav(tmp_path / "a.wav", 2.5, A)
|
|
|
+ good_b = write_const_wav(tmp_path / "b.wav", 2.5, B)
|
|
|
+ (tmp_path / "notes.wav").write_text("not audio", encoding="utf-8")
|
|
|
+ write_const_wav(tmp_path / "stereo.wav", 2.5, A, nchannels=2)
|
|
|
+ write_const_wav(tmp_path / "lowrate.wav", 2.5, A, sample_rate=8000)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+
|
|
|
+ ranked, skipped = rank_candidates(
|
|
|
+ ConstEmbedder(), seed_audio,
|
|
|
+ [good_a, good_b, tmp_path / "notes.wav", tmp_path / "stereo.wav",
|
|
|
+ tmp_path / "lowrate.wav", tmp_path / "missing.wav"],
|
|
|
+ min_enroll_s=2.0, min_score=0.55)
|
|
|
+
|
|
|
+ assert [c.path for c in ranked] == [good_a]
|
|
|
+ assert skipped == 4
|
|
|
+ assert ranked[0].score == pytest.approx(1.0)
|
|
|
+ assert ranked[0].duration_s == pytest.approx(2.5)
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_excludes_seed_and_short_clips(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ short = write_const_wav(tmp_path / "short.wav", 0.5, A)
|
|
|
+ long_a = write_const_wav(tmp_path / "long.wav", 2.5, A)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(
|
|
|
+ ConstEmbedder(), seed_audio, [seed, short, long_a],
|
|
|
+ min_enroll_s=2.0, min_score=0.55, exclude=(seed.resolve(),))
|
|
|
+
|
|
|
+ assert [c.path for c in ranked] == [long_a]
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_min_score_floor(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ mid = write_const_wav(tmp_path / "mid.wav", 2.5, 0.0)
|
|
|
+ other = write_const_wav(tmp_path / "other.wav", 2.5, B)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(
|
|
|
+ ConstEmbedder(), seed_audio, [other, mid],
|
|
|
+ min_enroll_s=2.0, min_score=0.55)
|
|
|
+
|
|
|
+ assert [c.path for c in ranked] == [mid]
|
|
|
+ assert ranked[0].score == pytest.approx(0.7071, abs=1e-3)
|
|
|
+
|
|
|
+
|
|
|
+# -- review_candidates ----------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def test_review_accept_skip_play_quit(tmp_path):
|
|
|
+ a = Candidate(Path("a.wav"), 0.9, 2.5)
|
|
|
+ b = Candidate(Path("b.wav"), 0.8, 2.5)
|
|
|
+ c = Candidate(Path("c.wav"), 0.7, 2.5)
|
|
|
+ played: list = []
|
|
|
+ review = ScriptReview(["accept", "skip", "play", "accept"])
|
|
|
+
|
|
|
+ accepted = review_candidates(
|
|
|
+ [a, b, c], review, 5, play=lambda p: played.append(p))
|
|
|
+
|
|
|
+ assert accepted == [a, c]
|
|
|
+ assert played == [c.path]
|
|
|
+ assert [s[0] for s in review.seen] == [a, b, c, c]
|
|
|
+
|
|
|
+
|
|
|
+def test_review_play_does_not_consume_candidate():
|
|
|
+ a = Candidate(Path("a.wav"), 0.9, 2.5)
|
|
|
+ review = ScriptReview(["play", "play", "accept"])
|
|
|
+
|
|
|
+ accepted = review_candidates([a], review, 1, play=lambda p: None)
|
|
|
+
|
|
|
+ assert accepted == [a]
|
|
|
+ assert len(review.seen) == 3
|
|
|
+
|
|
|
+
|
|
|
+def test_review_quit_returns_accepted_so_far():
|
|
|
+ a = Candidate(Path("a.wav"), 0.9, 2.5)
|
|
|
+ b = Candidate(Path("b.wav"), 0.8, 2.5)
|
|
|
+ review = ScriptReview(["accept", "quit"])
|
|
|
+
|
|
|
+ assert review_candidates([a, b], review, 5, play=lambda p: None) == [a]
|
|
|
+
|
|
|
+
|
|
|
+def test_review_count_stops_accepting():
|
|
|
+ cands = [Candidate(Path(f"{i}.wav"), 0.9, 2.5) for i in range(4)]
|
|
|
+
|
|
|
+ accepted = review_candidates(
|
|
|
+ cands, ScriptReview(["accept"] * 4), 2, play=lambda p: None)
|
|
|
+
|
|
|
+ assert len(accepted) == 2
|
|
|
+
|
|
|
+
|
|
|
+def test_review_rejects_unknown_action():
|
|
|
+ with pytest.raises(ValueError, match="unknown review action"):
|
|
|
+ review_candidates(
|
|
|
+ [Candidate(Path("a.wav"), 0.9, 2.5)], ScriptReview(["dance"]),
|
|
|
+ 1, play=lambda p: None)
|
|
|
+
|
|
|
+
|
|
|
+# -- play_wav -------------------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def test_play_wav_no_players_returns_none(monkeypatch, tmp_path):
|
|
|
+ monkeypatch.setattr(shutil, "which", lambda _name: None)
|
|
|
+
|
|
|
+ assert play_wav(tmp_path / "clip.wav") is None
|
|
|
+
|
|
|
+
|
|
|
+def test_play_wav_uses_first_available(monkeypatch, tmp_path):
|
|
|
+ monkeypatch.setattr(shutil, "which",
|
|
|
+ lambda name: "/usr/bin/paplay" if name == "paplay" else None)
|
|
|
+ ran: list = []
|
|
|
+ monkeypatch.setattr(subprocess, "run",
|
|
|
+ lambda *a, **k: ran.append(a[0]) or
|
|
|
+ subprocess.CompletedProcess(a[0], 0))
|
|
|
+
|
|
|
+ assert play_wav(tmp_path / "clip.wav") == "paplay"
|
|
|
+ assert ran[0][0] == "paplay"
|
|
|
+
|
|
|
+
|
|
|
+# -- seed_enroll ----------------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def seed_and_clips(audio_dir: Path, *, seed_value=A, seed_s=2.5,
|
|
|
+ clips=((("a1.wav", A), ("a2.wav", A), ("b1.wav", B)))):
|
|
|
+ seed = write_const_wav(audio_dir / "seed.wav", seed_s, seed_value)
|
|
|
+ paths = [write_const_wav(audio_dir / name, 2.5, value)
|
|
|
+ for name, value in clips]
|
|
|
+ return seed, paths
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_auto_enrolls_seed_plus_top_count(monkeypatch, tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ enrolled_log = record_enroll(monkeypatch)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=2, auto=True)
|
|
|
+
|
|
|
+ assert [c.path.name for c in accepted] == ["a1.wav", "a2.wav"]
|
|
|
+ assert [p.name for p in enrolled_log[0]] == ["seed.wav", "a1.wav", "a2.wav"]
|
|
|
+ assert store_files(store_dir)
|
|
|
+ assert fresh_speakers(store_dir)["spk_0001"]["name"] == "Efren"
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_short_seed_is_query_only(monkeypatch, tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir, seed_s=0.5)
|
|
|
+ enrolled_log = record_enroll(monkeypatch)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=2, auto=True)
|
|
|
+
|
|
|
+ assert len(accepted) == 2
|
|
|
+ assert [p.name for p in enrolled_log[0]] == ["a1.wav", "a2.wav"]
|
|
|
+ assert "query only" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_seed_excluded_from_candidates(tmp_path):
|
|
|
+ cfg, audio_dir, _ = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=10, auto=True)
|
|
|
+
|
|
|
+ assert seed not in [c.path for c in accepted]
|
|
|
+ assert len(accepted) == 2
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_fewer_candidates_than_count(monkeypatch, tmp_path):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir, clips=(("a1.wav", A),))
|
|
|
+ enrolled_log = record_enroll(monkeypatch)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=5, auto=True)
|
|
|
+
|
|
|
+ assert len(accepted) == 1
|
|
|
+ assert len(enrolled_log[0]) == 2
|
|
|
+ assert store_files(store_dir)
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_all_skipped_writes_nothing(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, review=ScriptReview(["skip"] * 3))
|
|
|
+
|
|
|
+ assert accepted == []
|
|
|
+ assert not store_files(store_dir)
|
|
|
+ assert "nothing enrolled" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_empty_dir_writes_nothing(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=3, auto=True)
|
|
|
+
|
|
|
+ assert accepted == []
|
|
|
+ assert not store_files(store_dir)
|
|
|
+ assert "nothing enrolled" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_quit_aborts_without_writing(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+
|
|
|
+ with pytest.raises(EnrollmentAborted):
|
|
|
+ seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren", seed,
|
|
|
+ review=ScriptReview(["accept", "quit"]))
|
|
|
+
|
|
|
+ assert not store_files(store_dir)
|
|
|
+ out = capsys.readouterr().out
|
|
|
+ assert "nothing enrolled" in out
|
|
|
+ assert "speaker-enroll" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_keyboard_interrupt_aborts_without_writing(tmp_path):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+
|
|
|
+ class ExplodingReview:
|
|
|
+ def choose(self, candidate, accepted, target):
|
|
|
+ raise KeyboardInterrupt
|
|
|
+
|
|
|
+ with pytest.raises(EnrollmentAborted):
|
|
|
+ seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren", seed,
|
|
|
+ review=ExplodingReview())
|
|
|
+
|
|
|
+ assert not store_files(store_dir)
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_dim_mismatch_errors_before_writing(tmp_path):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ old = make_store(cfg)
|
|
|
+ enroll_from_files(old, ConstEmbedder(dim=3), "Efren", [seed], 2.0)
|
|
|
+ store = make_store(cfg)
|
|
|
+ store.load()
|
|
|
+
|
|
|
+ with pytest.raises(ValueError, match="different model"):
|
|
|
+ seed_enroll(cfg, ConstEmbedder(dim=2), store, "Efren", seed, auto=True)
|
|
|
+
|
|
|
+ assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_max_prototypes_warns_and_evicts(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path, max_prototypes=2)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+
|
|
|
+ seed_enroll(cfg, VariedEmbedder(), make_store(cfg), "Efren", seed,
|
|
|
+ count=2, auto=True)
|
|
|
+
|
|
|
+ assert "evict" in capsys.readouterr().out
|
|
|
+ assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 2
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_max_scan_caps_embedding(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, _ = write_cfg(tmp_path)
|
|
|
+ seed = write_const_wav(audio_dir / "seed.wav", 2.5, A)
|
|
|
+ for i in range(5):
|
|
|
+ write_const_wav(audio_dir / f"clip{i}.wav", 2.5, A)
|
|
|
+ embedder = VariedEmbedder()
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, embedder, make_store(cfg), "Efren", seed,
|
|
|
+ count=5, max_scan=2, auto=True)
|
|
|
+
|
|
|
+ assert len(accepted) == 2
|
|
|
+ # seed embed (dim check) + seed embed (rank) + 2 candidates + enroll of 3
|
|
|
+ assert embedder.calls == 7
|
|
|
+ assert "max-scan 2" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+# -- CLI handler ----------------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def run_seed_handler(monkeypatch, tmp_path, *, name="Efren", seed=None,
|
|
|
+ count=5, min_score=0.55, audio_dir=None, max_scan=0,
|
|
|
+ yes=False, no_play=False, force=False, model_ok=True,
|
|
|
+ tty=True, review=None, min_enroll_s=2.0,
|
|
|
+ max_prototypes=8, dim=2, pre_enroll_dim=None,
|
|
|
+ embedder_cls: type = ConstEmbedder):
|
|
|
+ """Drive ``_handle_speaker_enroll_seed`` with fakes; return (rc, out, err)."""
|
|
|
+ cfg, default_audio, store_dir = write_cfg(
|
|
|
+ tmp_path, min_enroll_s=min_enroll_s, max_prototypes=max_prototypes,
|
|
|
+ model_ok=model_ok)
|
|
|
+ if audio_dir is None:
|
|
|
+ audio_dir = default_audio
|
|
|
+ if seed is None:
|
|
|
+ seed = write_const_wav(Path(audio_dir) / "seed.wav", 2.5, A)
|
|
|
+ if pre_enroll_dim is not None:
|
|
|
+ old = SpeakerStore(store_dir)
|
|
|
+ enroll_from_files(old, ConstEmbedder(dim=pre_enroll_dim), name,
|
|
|
+ [seed], min_enroll_s)
|
|
|
+ fake = embedder_cls(dim=dim)
|
|
|
+
|
|
|
+ class FakeSherpa:
|
|
|
+ def __init__(self, model_path, provider="cpu"):
|
|
|
+ self._embedder = fake
|
|
|
+
|
|
|
+ def embed(self, audio):
|
|
|
+ return fake.embed(audio)
|
|
|
+
|
|
|
+ monkeypatch.setattr(cli_module, "SherpaEmbedder", FakeSherpa)
|
|
|
+ if review is not None:
|
|
|
+ monkeypatch.setattr(cli_module, "QuestionaryReview", lambda **k: review)
|
|
|
+ if tty:
|
|
|
+ monkeypatch.setattr(sys, "stdin", _TtyStdin())
|
|
|
+ else:
|
|
|
+ monkeypatch.setattr(sys, "stdin", io.StringIO())
|
|
|
+ args = argparse.Namespace(
|
|
|
+ config=str(tmp_path / "config.toml"), name=name, seed=str(seed),
|
|
|
+ count=count, min_score=min_score, audio_dir=str(audio_dir),
|
|
|
+ max_scan=max_scan, yes=yes, no_play=no_play, force=force,
|
|
|
+ )
|
|
|
+ out, err = io.StringIO(), io.StringIO()
|
|
|
+ with redirect_stdout(out), redirect_stderr(err):
|
|
|
+ rc = cli_module._handle_speaker_enroll_seed(args)
|
|
|
+ return rc, out.getvalue(), err.getvalue()
|
|
|
+
|
|
|
+
|
|
|
+class _TtyStdin(io.StringIO):
|
|
|
+ def isatty(self):
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+def _audio_with_clips(tmp_path: Path, sub="audio") -> Path:
|
|
|
+ audio_dir = tmp_path / sub
|
|
|
+ audio_dir.mkdir(exist_ok=True)
|
|
|
+ write_const_wav(audio_dir / "seed.wav", 2.5, A)
|
|
|
+ write_const_wav(audio_dir / "a1.wav", 2.5, A)
|
|
|
+ write_const_wav(audio_dir / "a2.wav", 2.5, A)
|
|
|
+ write_const_wav(audio_dir / "b1.wav", 2.5, B)
|
|
|
+ return audio_dir
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_auto_success(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+
|
|
|
+ rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ count=2, yes=True,
|
|
|
+ embedder_cls=VariedEmbedder)
|
|
|
+
|
|
|
+ assert rc == 0
|
|
|
+ assert "enrolled 'Efren' as spk_0001 from seed" in out
|
|
|
+ assert "3 clip(s) (seed + 2 reviewed" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_short_seed_query_only(monkeypatch, tmp_path):
|
|
|
+ audio_dir = tmp_path / "audio"
|
|
|
+ audio_dir.mkdir(exist_ok=True)
|
|
|
+ seed = write_const_wav(audio_dir / "seed.wav", 0.5, A)
|
|
|
+ write_const_wav(audio_dir / "a1.wav", 2.5, A)
|
|
|
+
|
|
|
+ rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ seed=seed, count=1, yes=True)
|
|
|
+
|
|
|
+ assert rc == 0
|
|
|
+ assert "query-only" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_non_tty_without_yes_errors(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+
|
|
|
+ rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ tty=False, yes=False)
|
|
|
+
|
|
|
+ assert rc == 1
|
|
|
+ assert "--yes" in err
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_non_tty_with_yes_auto_accepts(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+
|
|
|
+ rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ tty=False, yes=True, count=1,
|
|
|
+ embedder_cls=VariedEmbedder)
|
|
|
+
|
|
|
+ assert rc == 0
|
|
|
+ assert "from seed" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_daemon_guard_and_force(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+ (tmp_path / "bus.sock").write_text("", encoding="utf-8")
|
|
|
+
|
|
|
+ rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ yes=True)
|
|
|
+ assert rc == 1
|
|
|
+ assert "daemon" in err.lower()
|
|
|
+
|
|
|
+ rc, _, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ yes=True, force=True,
|
|
|
+ embedder_cls=VariedEmbedder)
|
|
|
+ assert rc == 0
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_missing_model(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+
|
|
|
+ rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ yes=True, model_ok=False)
|
|
|
+
|
|
|
+ assert rc == 1
|
|
|
+ assert "model" in err.lower()
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_dim_mismatch_no_write(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+ seed = audio_dir / "seed.wav"
|
|
|
+
|
|
|
+ rc, _, err = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ seed=seed, yes=True, dim=2, pre_enroll_dim=3)
|
|
|
+
|
|
|
+ assert rc == 1
|
|
|
+ assert "different model" in err
|
|
|
+ assert fresh_speakers(tmp_path / "speakers")["spk_0001"]["prototypes"] == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_quit_no_write(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+
|
|
|
+ rc, out, _ = run_seed_handler(
|
|
|
+ monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ review=ScriptReview(["accept", "quit"]))
|
|
|
+
|
|
|
+ assert rc == 1
|
|
|
+ assert not (tmp_path / "speakers" / "speakers.json").exists()
|
|
|
+ assert "speaker-enroll" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_no_play_disables_playback(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+ seen: dict = {}
|
|
|
+
|
|
|
+ class SpyReviewFactory:
|
|
|
+ def __init__(self, **kwargs):
|
|
|
+ seen.update(kwargs)
|
|
|
+
|
|
|
+ def choose(self, candidate, accepted, target):
|
|
|
+ return "accept" if accepted < 1 else "quit"
|
|
|
+
|
|
|
+ monkeypatch.setattr(cli_module, "QuestionaryReview", SpyReviewFactory)
|
|
|
+ monkeypatch.setattr(ea, "play_wav", lambda p: (_ for _ in ()).throw(
|
|
|
+ AssertionError("playback attempted")))
|
|
|
+
|
|
|
+ rc, _, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ count=2, no_play=True)
|
|
|
+
|
|
|
+ assert seen.get("allow_play") is False
|
|
|
+ assert rc == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_no_candidates_no_write(monkeypatch, tmp_path):
|
|
|
+ audio_dir = tmp_path / "audio"
|
|
|
+ audio_dir.mkdir(exist_ok=True)
|
|
|
+ seed = write_const_wav(audio_dir / "seed.wav", 2.5, A)
|
|
|
+
|
|
|
+ rc, out, _ = run_seed_handler(monkeypatch, tmp_path, audio_dir=audio_dir,
|
|
|
+ seed=seed, yes=True)
|
|
|
+
|
|
|
+ assert rc == 1
|
|
|
+ assert "nothing enrolled" in out
|
|
|
+ assert not (tmp_path / "speakers" / "speakers.json").exists()
|
|
|
+
|
|
|
+
|
|
|
+def test_questionary_review_defaults():
|
|
|
+ import dataclasses
|
|
|
+
|
|
|
+ review = QuestionaryReview()
|
|
|
+
|
|
|
+ assert review.allow_play is True
|
|
|
+ candidate = Candidate(Path("x.wav"), 0.5, 1.0)
|
|
|
+ with pytest.raises(dataclasses.FrozenInstanceError):
|
|
|
+ candidate.score = 0.1 # type: ignore[misc]
|
|
|
+
|
|
|
+
|
|
|
+# -- duplicate protection -------------------------------------------------------
|
|
|
+
|
|
|
+
|
|
|
+def _store_with(root: Path, profiles) -> SpeakerStore:
|
|
|
+ store = SpeakerStore(root)
|
|
|
+ for name, vecs in profiles:
|
|
|
+ store.enroll(name, [np.array(v, dtype=np.float32) for v in vecs])
|
|
|
+ return store
|
|
|
+
|
|
|
+
|
|
|
+def test_candidate_equality_ignores_embedding_and_flags():
|
|
|
+ first = Candidate(Path("a.wav"), 0.9, 2.5,
|
|
|
+ embedding=np.array([1.0, 0.0], dtype=np.float32),
|
|
|
+ flags=("duplicate",))
|
|
|
+ second = Candidate(Path("a.wav"), 0.9, 2.5,
|
|
|
+ embedding=np.array([0.0, 1.0], dtype=np.float32),
|
|
|
+ flags=())
|
|
|
+
|
|
|
+ assert first == second
|
|
|
+ assert hash(first) == hash(second)
|
|
|
+ assert first != Candidate(Path("b.wav"), 0.9, 2.5)
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_sets_unit_embeddings_without_store(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
|
|
|
+ min_enroll_s=2.0, min_score=0.55)
|
|
|
+
|
|
|
+ assert len(ranked) == 1
|
|
|
+ embedding = ranked[0].embedding
|
|
|
+ assert embedding is not None
|
|
|
+ assert embedding.dtype == np.float32
|
|
|
+ assert float(np.linalg.norm(embedding)) == pytest.approx(1.0)
|
|
|
+ assert ranked[0].flags == ()
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_flags_duplicate_against_target(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+ store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]])])
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
|
|
|
+ min_enroll_s=2.0, min_score=0.55,
|
|
|
+ store=store, speaker_name="Efren")
|
|
|
+
|
|
|
+ assert ranked[0].flags == ("duplicate",)
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_flags_best_other_speaker_by_display_name(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ clip = write_const_wav(tmp_path / "b.wav", 2.5, B)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+ store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]]),
|
|
|
+ ("Rival", [[0.0, 1.0]]),
|
|
|
+ ("Mid", [[0.7071, 0.7071]])])
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
|
|
|
+ min_enroll_s=2.0, min_score=-1.0,
|
|
|
+ store=store, speaker_name="Efren")
|
|
|
+
|
|
|
+ assert ranked[0].flags == ("other-speaker:Rival",)
|
|
|
+
|
|
|
+
|
|
|
+def test_rank_no_flags_without_store_or_name(tmp_path):
|
|
|
+ seed = write_const_wav(tmp_path / "seed.wav", 2.5, A)
|
|
|
+ clip = write_const_wav(tmp_path / "a.wav", 2.5, A)
|
|
|
+ from stt_daemon.enroll import read_wav
|
|
|
+ seed_audio, _ = read_wav(seed)
|
|
|
+ store = _store_with(tmp_path / "store", [("Efren", [[1.0, 0.0]])])
|
|
|
+
|
|
|
+ ranked, _ = rank_candidates(ConstEmbedder(), seed_audio, [clip],
|
|
|
+ min_enroll_s=2.0, min_score=0.55,
|
|
|
+ store=store)
|
|
|
+
|
|
|
+ assert ranked[0].flags == ()
|
|
|
+
|
|
|
+
|
|
|
+def test_review_refuses_duplicate_accept_without_consuming_budget(capsys):
|
|
|
+ dup = Candidate(Path("dup.wav"), 0.99, 2.5, flags=("duplicate",))
|
|
|
+ good = Candidate(Path("good.wav"), 0.9, 2.5)
|
|
|
+ review = ScriptReview(["accept", "accept"])
|
|
|
+
|
|
|
+ accepted = review_candidates([dup, good], review, 1, play=lambda p: None)
|
|
|
+
|
|
|
+ assert accepted == [good]
|
|
|
+ out = capsys.readouterr().out
|
|
|
+ assert "dup.wav" in out
|
|
|
+ assert "already covered" in out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_auto_dedupes_incoming_set(monkeypatch, tmp_path, capsys):
|
|
|
+ cfg, audio_dir, _ = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ enrolled_log = record_enroll(monkeypatch)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, count=2, auto=True)
|
|
|
+
|
|
|
+ assert accepted == []
|
|
|
+ assert [p.name for p in enrolled_log[0]] == ["seed.wav"]
|
|
|
+ assert "duplicate" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_review_path_dedupes_incoming_set(monkeypatch, tmp_path, capsys):
|
|
|
+ cfg, audio_dir, _ = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ enrolled_log = record_enroll(monkeypatch)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, ConstEmbedder(), make_store(cfg), "Efren",
|
|
|
+ seed, review=ScriptReview(["accept", "accept"]))
|
|
|
+
|
|
|
+ assert accepted == []
|
|
|
+ assert [p.name for p in enrolled_log[0]] == ["seed.wav"]
|
|
|
+ assert "duplicate" in capsys.readouterr().out
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_review_vetoes_store_duplicate(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ store = make_store(cfg)
|
|
|
+ enroll_from_files(store, ConstEmbedder(), "Efren", [seed], 2.0)
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, ConstEmbedder(), store, "Efren", seed,
|
|
|
+ review=ScriptReview(["accept"] * 5))
|
|
|
+
|
|
|
+ assert accepted == []
|
|
|
+ out = capsys.readouterr().out
|
|
|
+ assert "already covered" in out
|
|
|
+ assert "nothing enrolled" in out
|
|
|
+ assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 1
|
|
|
+
|
|
|
+
|
|
|
+def test_seed_enroll_removes_store_duplicates_and_saves(tmp_path, capsys):
|
|
|
+ cfg, audio_dir, store_dir = write_cfg(tmp_path)
|
|
|
+ seed, _ = seed_and_clips(audio_dir)
|
|
|
+ store = make_store(cfg)
|
|
|
+ enroll_from_files(store, ConstEmbedder(), "Efren",
|
|
|
+ [seed, audio_dir / "a1.wav"], 2.0)
|
|
|
+ saves: list = []
|
|
|
+ real_save = store.save
|
|
|
+
|
|
|
+ def counting_save():
|
|
|
+ saves.append(1)
|
|
|
+ return real_save()
|
|
|
+
|
|
|
+ store.save = counting_save # type: ignore[method-assign]
|
|
|
+
|
|
|
+ accepted = seed_enroll(cfg, VariedEmbedder(), store, "Efren", seed,
|
|
|
+ count=2, auto=True)
|
|
|
+
|
|
|
+ assert [c.path.name for c in accepted] == ["a1.wav", "a2.wav"]
|
|
|
+ out = capsys.readouterr().out
|
|
|
+ assert "removed 1 duplicate" in out
|
|
|
+ assert len(saves) == 2 # enroll_from_files save + post-dedupe save
|
|
|
+ assert (store_dir / "speakers.json").exists()
|
|
|
+ assert fresh_speakers(store_dir)["spk_0001"]["prototypes"] == 4
|
|
|
+
|
|
|
+
|
|
|
+def test_cli_forwards_dedupe_and_other_thresholds(monkeypatch, tmp_path):
|
|
|
+ audio_dir = _audio_with_clips(tmp_path)
|
|
|
+ seed = audio_dir / "seed.wav"
|
|
|
+ write_cfg(tmp_path)
|
|
|
+ seen: dict = {}
|
|
|
+
|
|
|
+ def fake_seed_enroll(cfg, embedder, store, name, seed_path, **kwargs):
|
|
|
+ seen.update(kwargs)
|
|
|
+ return [Candidate(audio_dir / "a1.wav", 0.99, 2.5)]
|
|
|
+
|
|
|
+ monkeypatch.setattr(cli_module, "seed_enroll", fake_seed_enroll)
|
|
|
+ fake = ConstEmbedder()
|
|
|
+
|
|
|
+ class FakeSherpa:
|
|
|
+ def __init__(self, model_path, provider="cpu"):
|
|
|
+ pass
|
|
|
+
|
|
|
+ def embed(self, audio):
|
|
|
+ return fake.embed(audio)
|
|
|
+
|
|
|
+ monkeypatch.setattr(cli_module, "SherpaEmbedder", FakeSherpa)
|
|
|
+ args = argparse.Namespace(
|
|
|
+ config=str(tmp_path / "config.toml"), name="Efren", seed=str(seed),
|
|
|
+ count=2, min_score=0.55, audio_dir=str(audio_dir),
|
|
|
+ max_scan=0, yes=True, no_play=True, force=True,
|
|
|
+ dedupe_threshold=0.97, other_threshold=0.44,
|
|
|
+ )
|
|
|
+ out, err = io.StringIO(), io.StringIO()
|
|
|
+ with redirect_stdout(out), redirect_stderr(err):
|
|
|
+ rc = cli_module._handle_speaker_enroll_seed(args)
|
|
|
+
|
|
|
+ assert rc == 0
|
|
|
+ assert seen.get("dup_threshold") == pytest.approx(0.97)
|
|
|
+ assert seen.get("other_threshold") == pytest.approx(0.44)
|