From 829dd7d0a54288a394c41f3fce2ca7b030d462e8 Mon Sep 17 00:00:00 2001 From: tayyaba-code Date: Fri, 17 Jul 2026 16:18:19 +0500 Subject: [PATCH 1/2] Add voice authentication utilities and comprehensive tests --- dummy_model.pkl | Bin 0 -> 2399 bytes src/vocalid/audio_utils.py | 27 +++++++- src/vocalid/auth_adapter.py | 38 +++++++++++ src/vocalid/dataset_manager.py | 48 ++++++++++++++ src/vocalid/duplicate_detector.py | 46 +++++++++++++ src/vocalid/enroll_and_authenticate.py | 44 +++++++++++++ src/vocalid/enrollment.py | 70 ++++++++++++++++++++ src/vocalid/recorder.py | 43 +++++++++++++ src/vocalid/validator.py | 48 ++++++++++++++ tests/conftest.py | 54 ++++++++++++++-- tests/test_auth_adapter.py | 69 ++++++++++++++++++++ tests/test_dataset_manager.py | 46 +++++++++++++ tests/test_duplicate_detector.py | 71 ++++++++++++++++++++ tests/test_enrollment.py | 86 +++++++++++++++++++++++++ tests/test_recorder.py | 66 +++++++++++++++++++ tests/test_validator.py | 40 ++++++++++++ 16 files changed, 786 insertions(+), 10 deletions(-) create mode 100644 dummy_model.pkl create mode 100644 src/vocalid/auth_adapter.py create mode 100644 src/vocalid/dataset_manager.py create mode 100644 src/vocalid/duplicate_detector.py create mode 100644 src/vocalid/enroll_and_authenticate.py create mode 100644 src/vocalid/enrollment.py create mode 100644 src/vocalid/recorder.py create mode 100644 src/vocalid/validator.py create mode 100644 tests/test_auth_adapter.py create mode 100644 tests/test_dataset_manager.py create mode 100644 tests/test_duplicate_detector.py create mode 100644 tests/test_enrollment.py create mode 100644 tests/test_recorder.py create mode 100644 tests/test_validator.py diff --git a/dummy_model.pkl b/dummy_model.pkl new file mode 100644 index 0000000000000000000000000000000000000000..2578c7191bf15cc7a2e6af72c775d3520b7cb819 GIT binary patch literal 2399 zcmaJ@c~lc=5|<-_C@%3F4U;v!-XoIP^uynVT4>oYbf?tho=nVGZZ6et`RjTOw3m*d9j2rf>hDI z)F1_4Dv(KdN)-do`2CeKaSWlLDV8`iJVHrR<`VvTo{%X^FS%q!nKAD|WlDnP_MLXc!@t5KP~K$?*g$w>^uS>G6vc7Nta+G|LMkHBW1K~0$Q-!NEVG9vuX2;92l z?=Zr@Q{n$&m@pVNJ@v|fA=SX27dF`Y*g@!8J94OCTq<0Zm zk~&(n^j`-7-k^_~I0P$OIx00mW#~9-ihXlJ4tOT`h@?awo=@y4WJm6X$ts8Vs7*&O zWi-~UJyMDPT;*1I{6r?Ms@R@6c54|}eCUpx^*#$bq&@v#Z^!}1tX9*Pf%)KR>pEk1 z!&WH1)E2O8dMW(i*;;EZDuLeSrpAs}S(ujOajvvW2j||bNV_;H8ya^hpI$W0#l81A*lOL(_T{N~Us-OwUY?7|*T_bWbrLkq@J?FPtw-zNnB?ot)!;w0rM%WN zA69#1NbjxG!$3O|R&485U}p~##akQjj|2YO|2jzt(Pt+`u9=^LfvNHAXA6%){ViRx zMYs+NAH|=}ys5{Vy{Zj2k}FX0cS-DXN(Zmz&FV-DuExrDIl=**8&PvPWS^|=C@h!Q zkLNx~gdJsG=}Qwz@$AWkvIE^(oVt}w^*E@(y~EOlbD;*8%(yVk#!U}dZu4$#8Giz% z49fB8Ni?7+-DpLpTLqY8oU^ix$j7I5^hwcC8E`B&bWdP&9$d?Eeqq_W1rsx4*PUxg z1pcm!Ft3!IP?ub?EzJobp}wW`&i%vKKe=YdU7uZ`BgE-J>7cbbz8nzcZ4DdV^J>(%-fx|Wz zoFt^%(Q8K*gIx#vCu8vx>C5n9jEssh$d`S*IAgYeNYF`JzO9#-JJ-eXsw#&Q` z+e>iZx(94i>na!=@c4Jf*VW*X{!$q@^bnjoT&$X-EyAF=btXmQ_o4lB=t;eF9Bi%F zMqO?h@L2vIoAjCrEC{-n^u}HY9>FG@w#rQGOgekd!~YbXET6T~X0sm3V_dCO>Tr13 zLU+*-f$Dq|M5#`D$fbgjuEVJ2VxWMP0WBrqQ{KcZaRvszOkU$U#de_qf6_HozT2w>$EsdVzDXX`E+J|4&(%oI-G6Fc zr53NO&D!&~ITbM6{C@Z2W&1$&^zw-l_IeckV0*~_SprrRv~HU6MV=AU@M%B zemHZawgw(>wKj%^3KVzjP-mqcg!~+KyT4}{nvZe$3z@$g>la)*5hT|_+oq_wFDEFO zdp0!!@i1J)b#cFmi(p8=nBOa#%iz^>=^<0cGUyiixi|QgfdA9x0fjDwnEK<-+d1b$ zVCAv)`;uSfP?(o*8FgVDI@})p!K1DmnoVj(Jlhci%Cv&`y_0k}!mvU2K~xA`(9JSNG^e>BV^ThvdVeVfWRR2#6N-S*e?lhi{ eTF#d`R=m$+6OWlC`bHv( (channels, frames), matches torchaudio's shape + return waveform, sr # Convert to float32 waveform = waveform.float() @@ -60,3 +63,21 @@ def record_audio(seconds=4, fs=SAMPLE_RATE): audio_tensor = torch.nn.functional.pad(audio_tensor, (0, pad_len)) return audio_tensor + +def save_audio(audio_tensor, path, sample_rate=SAMPLE_RATE): + """ + Save a mono waveform tensor to a WAV file. + + Parameters + ---------- + audio_tensor : torch.Tensor + Shape (1, T) or (T,) + path : str + Output WAV filename. + sample_rate : int + Sample rate. + """ + if audio_tensor.ndim == 1: + audio_tensor = audio_tensor.unsqueeze(0) + + torchaudio.save(path, audio_tensor.cpu(), sample_rate) \ No newline at end of file diff --git a/src/vocalid/auth_adapter.py b/src/vocalid/auth_adapter.py new file mode 100644 index 0000000..a0c5fc6 --- /dev/null +++ b/src/vocalid/auth_adapter.py @@ -0,0 +1,38 @@ +""" +auth_adapter.py +Adapts VocalID's verification result into an authentication decision. +Deliberately has no ML in it - it just calls VoiceVerifier and +translates (verified, score) into an AuthenticationResult that +application code (door lock, login screen, attendance app, etc.) +can branch on. +""" + +from dataclasses import dataclass +from .verifier import VoiceVerifier +from . import config + + +@dataclass +class AuthenticationResult: + success: bool + confidence: float + + def __str__(self): + status = "ACCESS GRANTED" if self.success else "ACCESS DENIED" + return f"{status} (confidence: {self.confidence:.2f})" + + +class VoiceAuthenticator: + def __init__(self, model_path: str): + self.verifier = VoiceVerifier(model_path) + + def authenticate_file(self, audio_path: str) -> AuthenticationResult: + verified, score = self.verifier.verify_file(audio_path) + return AuthenticationResult(success=verified, confidence=score) + + def authenticate_live(self, seconds: float = None) -> AuthenticationResult: + from .recorder import record_one + seconds = seconds or 4.0 + audio = record_one(seconds, config.SAMPLE_RATE) + verified, score = self.verifier.verify_live(audio) + return AuthenticationResult(success=verified, confidence=score) diff --git a/src/vocalid/dataset_manager.py b/src/vocalid/dataset_manager.py new file mode 100644 index 0000000..dc8660f --- /dev/null +++ b/src/vocalid/dataset_manager.py @@ -0,0 +1,48 @@ +""" +dataset_manager.py +Keeps the on-disk dataset in the layout trainer.py expects: + + dataset/ + my_voice/ (positive samples) + other_voices/ (negative samples) + +Just handles copying accepted files in with sequential, collision-free +names - no metadata, no database. +""" + +import os +import shutil + +POSITIVE_DIR = "my_voice" +NEGATIVE_DIR = "other_voices" + + +class DatasetManager: + def __init__(self, root: str = "dataset"): + self.root = root + self.positive_dir = os.path.join(root, POSITIVE_DIR) + self.negative_dir = os.path.join(root, NEGATIVE_DIR) + os.makedirs(self.positive_dir, exist_ok=True) + os.makedirs(self.negative_dir, exist_ok=True) + + def _target_dir(self, label: str) -> str: + if label not in ("positive", "negative"): + raise ValueError("label must be 'positive' or 'negative'") + return self.positive_dir if label == "positive" else self.negative_dir + + def list_samples(self, label: str) -> list[str]: + target_dir = self._target_dir(label) + return [ + os.path.join(target_dir, f) + for f in sorted(os.listdir(target_dir)) + if f.lower().endswith(".wav") + ] + + def add_sample(self, src_path: str, label: str) -> str: + """Copies src_path into the dataset with the next free index. Returns the new path.""" + target_dir = self._target_dir(label) + existing = self.list_samples(label) + next_index = len(existing) + 1 + dest_path = os.path.join(target_dir, f"sample{next_index:03d}.wav") + shutil.copy2(src_path, dest_path) + return dest_path diff --git a/src/vocalid/duplicate_detector.py b/src/vocalid/duplicate_detector.py new file mode 100644 index 0000000..4b823fd --- /dev/null +++ b/src/vocalid/duplicate_detector.py @@ -0,0 +1,46 @@ +""" +duplicate_detector.py +Flags a new clip as a duplicate if it's too similar (by embedding +cosine similarity) to a clip already in the dataset, or to another +clip recorded in the same enrollment batch. This is the same +embedding extractor the trainer/verifier already use, so no extra +model is loaded. +""" + +import numpy as np +from .embeddings import EmbeddingExtractor +from .audio_utils import load_audio + +DUPLICATE_THRESHOLD = 0.97 # cosine similarity above this = "same clip" + + +def _cosine(a, b) -> float: + a = np.asarray(a).flatten() + b = np.asarray(b).flatten() + denom = (np.linalg.norm(a) * np.linalg.norm(b)) or 1e-8 + return float(np.dot(a, b) / denom) + + +class DuplicateDetector: + def __init__(self, threshold: float = DUPLICATE_THRESHOLD): + self.extractor = EmbeddingExtractor() + self.threshold = threshold + + def _embed_file(self, path: str): + audio, sample_rate = load_audio(path) + return self.extractor.extract(audio) + + def find_duplicate(self, new_path: str, existing_paths: list[str]): + """ + Compares `new_path` against every clip in `existing_paths`. + Returns the path of the first near-duplicate found, or None. + """ + new_emb = self._embed_file(new_path) + for path in existing_paths: + sim = _cosine(new_emb, self._embed_file(path)) + if sim >= self.threshold: + return path + return None + + def is_duplicate(self, new_path: str, existing_paths: list[str]) -> bool: + return self.find_duplicate(new_path, existing_paths) is not None diff --git a/src/vocalid/enroll_and_authenticate.py b/src/vocalid/enroll_and_authenticate.py new file mode 100644 index 0000000..a501bbe --- /dev/null +++ b/src/vocalid/enroll_and_authenticate.py @@ -0,0 +1,44 @@ +""" +End-to-end demo: + 1. Enroll positive (your voice) and negative (other voices) samples + 2. Train a model on the collected dataset + 3. Authenticate live from the microphone + +Run: python examples/enroll_and_authenticate.py +""" + +import glob + +from voice_verifier.enrollment import EnrollmentSession +from voice_verifier.trainer import VoiceTrainer +from voice_verifier.auth_adapter import VoiceAuthenticator + +DATASET_ROOT = "dataset" +MODEL_PATH = "my_voice_model.pkl" + + +def main(): + session = EnrollmentSession(dataset_root=DATASET_ROOT) + + print("=== Enrolling YOUR voice (positive samples) ===") + session.enroll(label="positive", count=10, seconds=5.0) + + print("\n=== Enrolling OTHER voices (negative samples) ===") + print("Have a few different people speak for this part.") + session.enroll(label="negative", count=10, seconds=5.0) + + print("\n=== Training ===") + pos_files = glob.glob(f"{DATASET_ROOT}/my_voice/*.wav") + neg_files = glob.glob(f"{DATASET_ROOT}/other_voices/*.wav") + trainer = VoiceTrainer() + trainer.train(pos_files, neg_files, save_path=MODEL_PATH) + print(f"Model saved to {MODEL_PATH}") + + print("\n=== Authenticate ===") + authenticator = VoiceAuthenticator(MODEL_PATH) + result = authenticator.authenticate_live(seconds=4.0) + print(result) + + +if __name__ == "__main__": + main() diff --git a/src/vocalid/enrollment.py b/src/vocalid/enrollment.py new file mode 100644 index 0000000..ffb4872 --- /dev/null +++ b/src/vocalid/enrollment.py @@ -0,0 +1,70 @@ +""" +enrollment.py +Ties recorder + validator + duplicate_detector + dataset_manager +together into one guided session: + + record clip -> validate -> check duplicate -> save to dataset + +Retries a rejected clip instead of silently skipping it, so the user +always ends up with the number of samples they asked for. +""" + +import os +import tempfile + +from .recorder import record_one +from .validator import check_audio +from .duplicate_detector import DuplicateDetector +from .dataset_manager import DatasetManager +from . import config + + +class EnrollmentSession: + def __init__(self, dataset_root: str = "dataset", max_attempts_per_sample: int = 3): + self.dataset = DatasetManager(dataset_root) + self.dupe_detector = DuplicateDetector() + self.max_attempts_per_sample = max_attempts_per_sample + + def enroll(self, label: str, count: int = 10, seconds: float = 5.0) -> list[str]: + """ + Records and accepts `count` valid, non-duplicate clips for + `label` ("positive" or "negative"). Returns the saved paths. + """ + saved_paths = [] + + for i in range(count): + accepted = False + + for attempt in range(1, self.max_attempts_per_sample + 1): + print(f"\nSample {i + 1}/{count} (attempt {attempt}) - label: {label}") + audio = record_one(seconds, config.SAMPLE_RATE) + + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: + tmp_path = tmp.name + from .audio_utils import save_audio + save_audio(audio, tmp_path, config.SAMPLE_RATE) + + is_valid, reason = check_audio(tmp_path) + if not is_valid: + print(f"Rejected: {reason}. Try again.") + os.remove(tmp_path) + continue + + existing = self.dataset.list_samples(label) + saved_paths + dupe = self.dupe_detector.find_duplicate(tmp_path, existing) + if dupe: + print(f"Rejected: too similar to {os.path.basename(dupe)}. Try again.") + os.remove(tmp_path) + continue + + saved_path = self.dataset.add_sample(tmp_path, label) + os.remove(tmp_path) + saved_paths.append(saved_path) + print(f"Saved as {saved_path}") + accepted = True + break + + if not accepted: + print(f"Giving up on sample {i + 1} after {self.max_attempts_per_sample} attempts.") + + return saved_paths diff --git a/src/vocalid/recorder.py b/src/vocalid/recorder.py new file mode 100644 index 0000000..0dddd02 --- /dev/null +++ b/src/vocalid/recorder.py @@ -0,0 +1,43 @@ +""" +recorder.py +Records short audio clips one at a time for voice enrollment. + +Assumes voice_verifier.audio_utils already exposes a `record_audio` +function (mic -> tensor/np array) and a `save_audio` function, +matching what the library's live-verification path uses. +If your actual audio_utils.py names these differently, just +adjust the two imports below. +""" + +import os +from .audio_utils import record_audio, save_audio +from . import config + + +def record_one(seconds: float = 5.0, sample_rate: int = None): + """Record a single clip and return the raw audio tensor.""" + sample_rate = sample_rate or config.SAMPLE_RATE + print(f"Recording for {seconds:.1f}s... speak naturally.") + audio = record_audio(duration=seconds, sample_rate=sample_rate) + print("Done.") + return audio + + +def record_batch(out_dir: str, count: int = 10, seconds: float = 5.0, + sample_rate: int = None, prefix: str = "sample"): + """ + Record `count` clips back to back, saving each to `out_dir`. + Returns the list of saved file paths (unvalidated). + """ + os.makedirs(out_dir, exist_ok=True) + sample_rate = sample_rate or config.SAMPLE_RATE + paths = [] + + for i in range(count): + input(f"[{i + 1}/{count}] Press Enter, then start speaking...") + audio = record_one(seconds, sample_rate) + path = os.path.join(out_dir, f"{prefix}_{i + 1:03d}.wav") + save_audio(audio, path, sample_rate) + paths.append(path) + + return paths diff --git a/src/vocalid/validator.py b/src/vocalid/validator.py new file mode 100644 index 0000000..05452d8 --- /dev/null +++ b/src/vocalid/validator.py @@ -0,0 +1,48 @@ +""" +validator.py +Basic sanity checks on a recorded clip before it's allowed into the +dataset: right length, not silent, not clipped/overdriven. + +Uses audio_utils.load_audio(path) -> (tensor, sample_rate), which the +library already needs for file-based verification. +""" + +import numpy as np +from .audio_utils import load_audio + +MIN_SECONDS = 4.0 +MAX_SECONDS = 6.5 +MIN_RMS = 0.01 # below this, treat the clip as silence/noise floor +CLIP_THRESHOLD = 0.98 # fraction of full scale considered "clipping" +MAX_CLIPPED_RATIO = 0.01 + + +def _to_numpy(audio): + if hasattr(audio, "numpy"): + return audio.numpy().flatten() + return np.asarray(audio).flatten() + + +def check_audio(path: str) -> tuple[bool, str]: + """ + Returns (is_valid, reason). reason is 'ok' when valid, otherwise + a short human-readable explanation of why it was rejected. + """ + audio, sample_rate = load_audio(path) + samples = _to_numpy(audio) + + duration = len(samples) / float(sample_rate) + if duration < MIN_SECONDS: + return False, f"too short ({duration:.1f}s)" + if duration > MAX_SECONDS: + return False, f"too long ({duration:.1f}s)" + + rms = float(np.sqrt(np.mean(samples ** 2))) + if rms < MIN_RMS: + return False, "too quiet / silence detected" + + clipped_ratio = float(np.mean(np.abs(samples) >= CLIP_THRESHOLD)) + if clipped_ratio > MAX_CLIPPED_RATIO: + return False, "clipping / distortion detected" + + return True, "ok" diff --git a/tests/conftest.py b/tests/conftest.py index 9cfd57c..e133926 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,10 +1,50 @@ -import sys -import os +""" +conftest.py +Shared fixtures for the new modules' tests. Everything here writes +real, tiny .wav files with numpy + soundfile so tests run anywhere +(CI included) without a microphone or a real voice sample. +""" -ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +import numpy as np +import soundfile as sf +import pytest -if os.environ.get("GITHUB_ACTIONS") == "true": - os.environ["SKIP_SPEECHBRAIN"] = "1" +SAMPLE_RATE = 16000 -if ROOT not in sys.path: - sys.path.insert(0, ROOT) \ No newline at end of file + +def _tone(duration: float, sample_rate: int, freq: float = 220.0, amplitude: float = 0.3): + """A simple sine wave - stands in for a 'real' voice-shaped signal.""" + t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False) + return (amplitude * np.sin(2 * np.pi * freq * t)).astype(np.float32) + + +@pytest.fixture +def write_wav(tmp_path): + """ + Returns a function write_wav(name, **kwargs) -> path + that writes a synthetic clip and returns its path. + + kwargs: + duration seconds (default 5.0) + sample_rate (default 16000) + freq tone frequency, changes the clip's "identity" (default 220.0) + amplitude 0..1 (default 0.3) + silent if True, writes near-zero samples instead of a tone + clipped if True, writes a hard-clipped square-ish wave + """ + def _write(name: str, duration: float = 5.0, sample_rate: int = SAMPLE_RATE, + freq: float = 220.0, amplitude: float = 0.3, + silent: bool = False, clipped: bool = False): + if silent: + samples = np.zeros(int(sample_rate * duration), dtype=np.float32) + elif clipped: + samples = _tone(duration, sample_rate, freq, amplitude=1.0) + samples = np.clip(samples * 5.0, -1.0, 1.0) + else: + samples = _tone(duration, sample_rate, freq, amplitude) + + path = tmp_path / name + sf.write(str(path), samples, sample_rate) + return str(path) + + return _write diff --git a/tests/test_auth_adapter.py b/tests/test_auth_adapter.py new file mode 100644 index 0000000..d7de448 --- /dev/null +++ b/tests/test_auth_adapter.py @@ -0,0 +1,69 @@ +""" +test_auth_adapter.py + +VoiceVerifier is mocked out so this tests the adapter's translation +of (verified, score) -> AuthenticationResult, not the real +verification model. +""" + +import pytest +import vocalid.auth_adapter as auth_module +from vocalid.auth_adapter import VoiceAuthenticator, AuthenticationResult + + +class _FakeVerifier: + def __init__(self, model_path): + self.model_path = model_path + + def verify_file(self, path): + return True, 0.87 + + def verify_live(self, audio): + return False, 0.21 + + +@pytest.fixture +def authenticator(monkeypatch): + monkeypatch.setattr(auth_module, "VoiceVerifier", _FakeVerifier) + return VoiceAuthenticator("dummy_model.pkl") + + +def test_authenticate_file_returns_granted_result(authenticator): + result = authenticator.authenticate_file("some_clip.wav") + assert isinstance(result, AuthenticationResult) + assert result.success is True + assert result.confidence == 0.87 + + +def test_authenticate_live_returns_denied_result(authenticator, monkeypatch): + # authenticate_live does `from .recorder import record_one` internally, + # so the real patch point is vocalid.recorder.record_one. + monkeypatch.setattr("vocalid.recorder.record_one", + lambda seconds, sample_rate: "fake_audio") + + result = authenticator.authenticate_live(seconds=4.0) + + assert result.success is False + assert result.confidence == 0.21 + + +def test_authenticate_live_defaults_to_four_seconds(authenticator, monkeypatch): + captured = {} + + def fake_record_one(seconds, sample_rate): + captured["seconds"] = seconds + return "fake_audio" + + monkeypatch.setattr("vocalid.recorder.record_one", fake_record_one) + authenticator.authenticate_live(seconds=None) + + assert captured["seconds"] == 4.0 + + +def test_result_str_formatting(): + granted = AuthenticationResult(success=True, confidence=0.9123) + denied = AuthenticationResult(success=False, confidence=0.1) + + assert "ACCESS GRANTED" in str(granted) + assert "0.91" in str(granted) + assert "ACCESS DENIED" in str(denied) diff --git a/tests/test_dataset_manager.py b/tests/test_dataset_manager.py new file mode 100644 index 0000000..ebf4b54 --- /dev/null +++ b/tests/test_dataset_manager.py @@ -0,0 +1,46 @@ +""" +test_dataset_manager.py +""" + +import os +import pytest +from vocalid.dataset_manager import DatasetManager + + +def test_creates_folder_layout(tmp_path): + root = tmp_path / "dataset" + DatasetManager(str(root)) + assert (root / "my_voice").is_dir() + assert (root / "other_voices").is_dir() + + +def test_add_sample_copies_and_lists(tmp_path, write_wav): + dm = DatasetManager(str(tmp_path / "dataset")) + clip = write_wav("clip.wav") + + saved_path = dm.add_sample(clip, "positive") + + assert os.path.exists(saved_path) + assert saved_path in dm.list_samples("positive") + assert dm.list_samples("negative") == [] + + +def test_add_sample_sequential_naming(tmp_path, write_wav): + dm = DatasetManager(str(tmp_path / "dataset")) + clip_a = write_wav("a.wav") + clip_b = write_wav("b.wav") + + path1 = dm.add_sample(clip_a, "negative") + path2 = dm.add_sample(clip_b, "negative") + + assert os.path.basename(path1) == "sample001.wav" + assert os.path.basename(path2) == "sample002.wav" + assert len(dm.list_samples("negative")) == 2 + + +def test_rejects_bad_label(tmp_path, write_wav): + dm = DatasetManager(str(tmp_path / "dataset")) + clip = write_wav("clip.wav") + + with pytest.raises(ValueError): + dm.add_sample(clip, "unknown_label") diff --git a/tests/test_duplicate_detector.py b/tests/test_duplicate_detector.py new file mode 100644 index 0000000..05bf35b --- /dev/null +++ b/tests/test_duplicate_detector.py @@ -0,0 +1,71 @@ +""" +test_duplicate_detector.py + +The real EmbeddingExtractor loads a full ECAPA-TDNN model from +speechbrain, which is slow and needs a network connection on first +run. For unit tests we swap it for a fake extractor that just turns +the raw audio into its "embedding" - two clips with the same tone +frequency come out identical, different frequencies come out +different, which is enough to exercise the cosine-similarity logic +in duplicate_detector.py itself. Swap this fixture out once you're +ready to write a slower, real-model integration test. +""" + +import numpy as np +import soundfile as sf +import pytest + +import vocalid.duplicate_detector as dd_module +from vocalid.duplicate_detector import DuplicateDetector + + +class _FakeExtractor: + def extract(self, audio): + return np.asarray(audio).flatten() + + +def _fake_load_audio(path): + samples, sample_rate = sf.read(path, dtype="float32") + return samples, sample_rate + + +@pytest.fixture +def detector(monkeypatch): + monkeypatch.setattr(dd_module, "EmbeddingExtractor", _FakeExtractor) + monkeypatch.setattr(dd_module, "load_audio", _fake_load_audio) + return DuplicateDetector() + + +def test_identical_clip_is_flagged(detector, write_wav): + original = write_wav("original.wav", freq=220.0) + replay = write_wav("replay.wav", freq=220.0) # same tone == same "voice" + + match = detector.find_duplicate(replay, [original]) + assert match == original + assert detector.is_duplicate(replay, [original]) is True + + +def test_different_clip_is_not_flagged(detector, write_wav): + original = write_wav("original.wav", freq=220.0) + different = write_wav("different.wav", freq=880.0) # different tone == different "voice" + + match = detector.find_duplicate(different, [original]) + assert match is None + assert detector.is_duplicate(different, [original]) is False + + +def test_no_existing_clips_never_flags(detector, write_wav): + clip = write_wav("only.wav") + assert detector.find_duplicate(clip, []) is None + + +def test_custom_threshold_is_respected(write_wav, monkeypatch): + monkeypatch.setattr(dd_module, "EmbeddingExtractor", _FakeExtractor) + monkeypatch.setattr(dd_module, "load_audio", _fake_load_audio) + + strict_detector = DuplicateDetector(threshold=0.999999) + original = write_wav("original.wav", freq=220.0, amplitude=0.3) + slightly_different = write_wav("close.wav", freq=222.0, amplitude=0.3) + + # Close but not identical tones shouldn't clear an extremely strict threshold. + assert strict_detector.is_duplicate(slightly_different, [original]) is False diff --git a/tests/test_enrollment.py b/tests/test_enrollment.py new file mode 100644 index 0000000..f77c40a --- /dev/null +++ b/tests/test_enrollment.py @@ -0,0 +1,86 @@ +""" +test_enrollment.py + +record_one, audio_utils.save_audio, and DuplicateDetector are all +mocked so this exercises EnrollmentSession's retry/accept/give-up +logic without needing a mic, a real audio file, or the ECAPA model. +DatasetManager is used for real (against tmp_path) since it's just +file I/O and worth testing end to end here. +""" + +import pytest +import vocalid.enrollment as enrollment_module +from vocalid.enrollment import EnrollmentSession + + +class _FakeDupeDetector: + """queue of paths to report as duplicates, one per find_duplicate() call""" + def __init__(self, duplicate_paths=None): + self._duplicate_paths = list(duplicate_paths or []) + + def find_duplicate(self, new_path, existing_paths): + return self._duplicate_paths.pop(0) if self._duplicate_paths else None + + +@pytest.fixture(autouse=True) +def patch_recording(monkeypatch): + monkeypatch.setattr(enrollment_module, "record_one", lambda seconds, sample_rate: "fake_audio") + # save_audio is imported lazily inside enroll(), so the real patch + # point is the audio_utils module itself, not enrollment_module. + monkeypatch.setattr("vocalid.audio_utils.save_audio", lambda audio, path, sr: None) + + +def test_accepts_a_valid_unique_clip_on_first_try(monkeypatch, tmp_path): + monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) + monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + + session = EnrollmentSession(dataset_root=str(tmp_path / "dataset")) + saved = session.enroll(label="positive", count=1, seconds=5.0) + + assert len(saved) == 1 + + +def test_retries_after_a_rejected_clip(monkeypatch, tmp_path): + results = iter([(False, "too quiet"), (True, "ok")]) + monkeypatch.setattr(enrollment_module, "check_audio", lambda path: next(results)) + monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + + session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=3) + saved = session.enroll(label="positive", count=1, seconds=5.0) + + assert len(saved) == 1 # accepted on the second attempt + + +def test_retries_after_a_duplicate_clip(monkeypatch, tmp_path): + monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) + # first attempt flagged as a duplicate, second attempt is clean + monkeypatch.setattr( + enrollment_module, "DuplicateDetector", + lambda: _FakeDupeDetector(duplicate_paths=["dataset/my_voice/sample001.wav"]) + ) + + session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=3) + saved = session.enroll(label="positive", count=1, seconds=5.0) + + assert len(saved) == 1 + + +def test_gives_up_after_max_attempts(monkeypatch, tmp_path): + monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (False, "too quiet")) + monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + + session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=2) + saved = session.enroll(label="positive", count=1, seconds=5.0) + + assert saved == [] # never accepted, session moved on + + +def test_returns_a_path_for_every_accepted_sample(monkeypatch, tmp_path): + monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) + monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + + session = EnrollmentSession(dataset_root=str(tmp_path / "dataset")) + saved = session.enroll(label="negative", count=3, seconds=5.0) + + assert len(saved) == 3 + assert len(set(saved)) == 3 # all distinct paths, no overwrites diff --git a/tests/test_recorder.py b/tests/test_recorder.py new file mode 100644 index 0000000..28a528c --- /dev/null +++ b/tests/test_recorder.py @@ -0,0 +1,66 @@ +""" +test_recorder.py + +record_audio / save_audio (from audio_utils) are mocked out - these +tests check recorder.py's own logic (sample-rate defaulting, file +naming, loop count), not the real microphone or real file writing. +""" + +import pytest +import vocalid.recorder as recorder_module + + +@pytest.fixture +def fake_audio_utils(monkeypatch): + calls = {"record_audio": [], "save_audio": []} + + def fake_record_audio(duration, sample_rate): + calls["record_audio"].append((duration, sample_rate)) + return f"audio@{duration}s" + + def fake_save_audio(audio, path, sample_rate): + calls["save_audio"].append((audio, path, sample_rate)) + + monkeypatch.setattr(recorder_module, "record_audio", fake_record_audio) + monkeypatch.setattr(recorder_module, "save_audio", fake_save_audio) + return calls + + +def test_record_one_uses_given_seconds_and_rate(fake_audio_utils): + audio = recorder_module.record_one(seconds=6.0, sample_rate=22050) + assert audio == "audio@6.0s" + assert fake_audio_utils["record_audio"] == [(6.0, 22050)] + + +def test_record_one_falls_back_to_config_sample_rate(fake_audio_utils, monkeypatch): + monkeypatch.setattr(recorder_module.config, "SAMPLE_RATE", 16000) + recorder_module.record_one(seconds=5.0, sample_rate=None) + assert fake_audio_utils["record_audio"] == [(5.0, 16000)] + + +def test_record_batch_records_and_saves_expected_count(fake_audio_utils, monkeypatch, tmp_path): + monkeypatch.setattr("builtins.input", lambda prompt="": "") + + paths = recorder_module.record_batch(str(tmp_path), count=3, seconds=5.0, sample_rate=16000) + + assert len(paths) == 3 + assert len(fake_audio_utils["record_audio"]) == 3 + assert len(fake_audio_utils["save_audio"]) == 3 + + +def test_record_batch_names_files_sequentially(fake_audio_utils, monkeypatch, tmp_path): + monkeypatch.setattr("builtins.input", lambda prompt="": "") + + paths = recorder_module.record_batch(str(tmp_path), count=2, prefix="clip") + + assert paths[0].endswith("clip_001.wav") + assert paths[1].endswith("clip_002.wav") + + +def test_record_batch_creates_output_dir(fake_audio_utils, monkeypatch, tmp_path): + monkeypatch.setattr("builtins.input", lambda prompt="": "") + out_dir = tmp_path / "nested" / "clips" + + recorder_module.record_batch(str(out_dir), count=1) + + assert out_dir.is_dir() diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..24f95c9 --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,40 @@ +""" +test_validator.py +""" + +from vocalid.validator import check_audio + + +def test_accepts_good_clip(write_wav): + path = write_wav("good.wav", duration=5.0) + is_valid, reason = check_audio(path) + assert is_valid is True + assert reason == "ok" + + +def test_rejects_too_short(write_wav): + path = write_wav("short.wav", duration=1.5) + is_valid, reason = check_audio(path) + assert is_valid is False + assert "short" in reason + + +def test_rejects_too_long(write_wav): + path = write_wav("long.wav", duration=8.0) + is_valid, reason = check_audio(path) + assert is_valid is False + assert "long" in reason + + +def test_rejects_silence(write_wav): + path = write_wav("silent.wav", duration=5.0, silent=True) + is_valid, reason = check_audio(path) + assert is_valid is False + assert "quiet" in reason or "silence" in reason + + +def test_rejects_clipping(write_wav): + path = write_wav("clipped.wav", duration=5.0, clipped=True) + is_valid, reason = check_audio(path) + assert is_valid is False + assert "clip" in reason From 0922dd31b83ac33004e0d769bb1bc230efbaa005 Mon Sep 17 00:00:00 2001 From: tayyaba-code Date: Fri, 17 Jul 2026 21:02:57 +0500 Subject: [PATCH 2/2] Fix voice authentication pipeline and update tests --- dummy_model.pkl | Bin 2399 -> 2399 bytes src/vocalid/audio_utils.py | 159 ++++++++++++++------ src/vocalid/auth_adapter.py | 16 +- src/vocalid/duplicate_detector.py | 46 ------ src/vocalid/embeddings.py | 195 +++++++++++++++++++++---- src/vocalid/enroll_and_authenticate.py | 6 +- src/vocalid/enrollment.py | 162 ++++++++++++++++---- src/vocalid/recorder.py | 118 ++++++++++++--- src/vocalid/trainer.py | 2 + src/vocalid/verifier.py | 2 +- tests/test_auth_adapter.py | 70 ++++++--- tests/test_duplicate_detector.py | 71 --------- tests/test_enrollment.py | 161 ++++++++++++++------ 13 files changed, 688 insertions(+), 320 deletions(-) delete mode 100644 src/vocalid/duplicate_detector.py delete mode 100644 tests/test_duplicate_detector.py diff --git a/dummy_model.pkl b/dummy_model.pkl index 2578c7191bf15cc7a2e6af72c775d3520b7cb819..feab21ddec8de4a4ed032b2b2a8b4049927403bb 100644 GIT binary patch delta 1571 zcmWN|eLNEg9{}(hpQ1w7@fxthg7tw)CY=1jJ0LjvDu_DC;yFzx7_$%k_@2mtoVVTfCVH3f-qL~KTV_JElIVvx zF>@I>-T%bMM_&c!tZ8iyd>MMngF+31vhX%b{rDega*W8?JwH0h$29E)LP0_+EUPW- zZRU7*c#0k=_f#Tbq2)7qc@HGoGUWp`O1Nk4WTG=DK*w_f z=8hVi30AITOfX^7SX|)nDx(TsC4U!{`|tl^>_I!K<)JFLs7oX2&z8Wi8Na;@x@L*g z7g{#k{~(MBh(!(TLMWa~3Jy_R#dV9Xi!GHZEN7PuZM`kWA#LbD%v=+WmbC3XzPT99 zn&gKS{X)1=8ga%nwHE^WF3s`zc;!Z{p6vo4g>=yNi$xHp()B$OO5nsns|Tz_C8#Ss>`h7qSmd~NsPmN= zJnFM%Y^5#Ocu0E7I*5ZEHU(3j(OT#>Zk2xDs6y5|qF`%ZCX5quJ?kC_@ye~WYZ&2# zD6ph%dv2u>0@WoZbRR{9i;h2$Cw`Y9RYXtv`E)X7Bu=_Dr8nai_nVU+q6M(?izm~s zuQsE_@nI8&OBZt8To#1uk}=VUwlPDaz?_ewoHI>QNK!S@Z&Uf0Ny=;Ucd5oic8@bz zdJ$b+vK@<12^9(h#!NS@1Xo1yapPe!9Q1PZikCEjxsOld6uk(mpV#P|%~UuMY*R?! zlmc&W+e`Yq0L~V+ZJDlQ;tatPO?TGggmC|IHop#MWBcP24lQWf-G7DjixNifIIa|F z>oMvfD}bJV1s4Q--G&4S?(=3DbC<**k}@h{XruQ71rH-XFWfFRECN;K-l=yN!(c}E)bG`+EHq%m254559BfRp zcl7#8j3HcIs%KRN?5YuwrG*mMh2xoY%){;}8so#gJg|x7ncocKftvef#lgQ6Mz56_ zzUjFD5~`lB?UBov;S>1HDJ>r?-R(0co}7o8Zwa;cIqhgu(EM`EsRnS%ou@^eki(Ve zk6%Tab|6E?49TCas=|IF_O_!(1Q@=HdpNExwTLoBFwRv~Q%;nF&BM2>hI|$PNwCQKD1iJ^r18nd zTp)gV+T;bf05>#@($7Dxz`VPi>D4a;P{tlf+0N&~?vaf6uW=LZCk|h%v#5p-($yB{ z*r9HWvEPOew3r14Zd-8UWybhemcJcUd*&%xSqD(Nx~4|^t+0UF}8W2_}@vv z_r8}ni1`+?|>wEF+jbu9j|({ z+wd9RRzTfyj@<749^k(@d?EX}0Ir*R1x6}L@sE$Lc9WI0m|V?1Fs{>pQ~1DF`NRfv zc74|AmRt*N$M)DF^0Sqb$hAJT4Aht zU1}Sf44KX7(fsttYdm%7dBT$@$5D7jdr+k3zJJ4a&}`7GT4zAF(l;j@Gt)qr`!B1! z>hHqXne}VJzvDsgtM*meAC%}wu6X-)B^TBwWixklA#RbnY@``v8vk|DS^c0vfG1TMf=%g4Sn)*{ zweNEt4%75seM}R9U0#pTT#y_*NKSk6ThgKOS>LhayQ|?_j~=ZFwF;Ix+ggX`^DvX` z+EG0whK^4sGM{cJg4W!G>8D2YB3x*BviVaB4`cj|eWNF}NC_C}m)uf;_|RpaqbDoi z?YS21$C)xHV7G}LmDWR9LZwbIE5;l7k-=d>WhgJ&=Dhg47^e(2H}z~x#j&jKVCaGr zXEe@zRg3xPxa(BOCE0y^Xj2)RvrCDl-gL?LsWeE@i0y~y0X!T}sIj~lCqYgxrIl#O zhPJ(4>_a*!T9T7F{hb={Thm>m^^n6UPd;rlPzoz|7#gSaqywSIoSJGW!$0o$W&CYx z0wlF>jXHhcItKAl2{Q-pL-T7fXT>=&Dkf4N2?nG%u#}kgilf2{|AT!u{@ ziCuMgWN+6lD`zR>IlB*Bvu=Rx--vwlEE!U>mYo=JR)HbEV>Ri#9H-w%*-7zysF$2A z21trc2k5B}k!%c675KESLX_=k*+D;#A$p{2qbP++F=<{2qSAsmp%rlMCXA z=qzSwHOd8>Ua&5zF}%0`eMNr(7S~nrG~r&*Lw2C_y_7&m=ClAoh9;K_ZG)T@Ag_F( zX0mdy-yR;PycXjMe$3}4s0JTxqibZEl)m=1Nj-=NE zhc2F}fvvsUw%BD#kvo;+^jo70W@ia#Y+i`Naq&egpIl@RAJyvjOW;&rZ6brsfNz)+ zG?Ez~JQ|bC-cwX)ezojj1T6zS&D*%GA|W0nPVS+!mSfnS3#$gnLX4sZrX3FPj)i#6 z5P8{z7|*)$r3;EY@J{@hH$NeRwBoZNi+i|Wwe-sOpI6j)^oZ@I`rsU>wRSIEE#rgd zud};tEs|h-MDkR+R)&Kds-MI`4GX)cDjF!73ix{NAZ8TyW3WLXQVj!ph<_fZ3mS#t?NYf7NwUS;BbVFd=eH5pb| z-$t7cpyxe%0HoD~Wn<2Kn2h_MRaTt}%YsMQpKKz*m261tQwwl}-9GB-*N6{m_61s9 zl|oIjlVu|F9L(vReh!|yjJk4}n{haUj+ZvDDa=+TnzXnS)|p<#{xZwW#_g4m6+V5# z_3=%V6wP5!u?B9T7tzXrh0B)cZ`!hwarJBJO($X*oZC5U{I!D;y?sXQnx~ra*A>lL z>d7DBk?jL}BgS2r4e&Z(?of_n72K}#GaA@-+G#uKpbUQ$x2YEk=|ZrpT_Va2OF?^t za*O$s1}SdEjqYk8zBrRt@b~>HSZgw_n>>CS5~rUxG}uUy`Yox%Z-#-YvYsp3|4_iQ zfQcR4cj?fc^xl5GunyiG7Fx+#R2V&+&CKK71-XbY|aj$KemJ+ AaR2}S diff --git a/src/vocalid/audio_utils.py b/src/vocalid/audio_utils.py index 01fc38f..7e605a3 100644 --- a/src/vocalid/audio_utils.py +++ b/src/vocalid/audio_utils.py @@ -1,83 +1,150 @@ import torch import torchaudio import sounddevice as sd -import numpy as np +import soundfile as sf + from .config import SAMPLE_RATE -import soundfile as sf -import torch -def load_audio(path,target_sr=SAMPLE_RATE): - samples, sr = sf.read(path, dtype="float32", always_2d=True) # (frames, channels) - waveform = torch.from_numpy(samples.T) # -> (channels, frames), matches torchaudio's shape - return waveform, sr - # Convert to float32 - waveform = waveform.float() +def load_audio(path, target_sr=SAMPLE_RATE): + """ + Load a WAV file and return a mono waveform tensor. - # Resample if needed - if sr != target_sr: - waveform = torchaudio.functional.resample(waveform, sr, target_sr) + Parameters + ---------- + path : str + Path to the WAV file. + target_sr : int + Desired sample rate. + + Returns + ------- + tuple(torch.Tensor, int) + Waveform with shape (1, T) and sample rate. + """ + + samples, sr = sf.read( + path, + dtype="float32", + always_2d=True + ) + + waveform = torch.from_numpy(samples.T).float() - # Mono - if waveform.ndim == 2 and waveform.shape[0] > 1: + # Convert stereo to mono + if waveform.shape[0] > 1: waveform = waveform.mean(dim=0, keepdim=True) - # Ensure shape is (1, T) + # Resample if necessary + if sr != target_sr: + waveform = torchaudio.functional.resample( + waveform, + sr, + target_sr + ) + sr = target_sr + + # Ensure shape (1, T) if waveform.ndim == 1: waveform = waveform.unsqueeze(0) - # Pad short audio - min_len = target_sr # 1 second minimum - if waveform.shape[1] < min_len: - pad_len = min_len - waveform.shape[1] - waveform = torch.nn.functional.pad(waveform, (0, pad_len)) + # Pad to at least one second + if waveform.shape[1] < target_sr: + pad_len = target_sr - waveform.shape[1] + waveform = torch.nn.functional.pad( + waveform, + (0, pad_len) + ) - return waveform + return waveform,sr -def record_audio(seconds=4, fs=SAMPLE_RATE): - print(f"Recording {seconds} seconds...") - # Check if sounddevice sees any input devices +def record_audio( + duration=4, + sample_rate=SAMPLE_RATE +): + """ + Record audio from the microphone. + + Parameters + ---------- + duration : float + Recording duration in seconds. + sample_rate : int + Recording sample rate. + + Returns + ------- + torch.Tensor + Recorded waveform with shape (1, T). + """ + + print(f"Recording {duration:.1f} seconds...") + try: devices = sd.query_devices() - has_input = any(d["max_input_channels"] > 0 for d in devices) - if not has_input: - raise RuntimeError("No input microphone found. Live recording cannot run here.") - except Exception: - raise RuntimeError("No audio interface available. Live recording unsupported.") - # Try to record + if not any(device["max_input_channels"] > 0 for device in devices): + raise RuntimeError("No microphone found.") + + except Exception as e: + raise RuntimeError( + "Unable to access an input microphone." + ) from e + try: - audio = sd.rec(int(seconds * fs), samplerate=fs, channels=1, dtype='float32') + audio = sd.rec( + int(duration * sample_rate), + samplerate=sample_rate, + channels=1, + dtype="float32" + ) + sd.wait() + except Exception as e: - raise RuntimeError("Live recording failed. Likely running in Colab.") from e + raise RuntimeError( + "Audio recording failed." + ) from e print("Recording complete") - audio_tensor = torch.tensor(audio.T, dtype=torch.float32) + waveform = torch.from_numpy(audio.T).float() - # Ensure minimum length - if audio_tensor.shape[1] < fs: - pad_len = fs - audio_tensor.shape[1] - audio_tensor = torch.nn.functional.pad(audio_tensor, (0, pad_len)) + # Ensure minimum duration of one second + if waveform.shape[1] < sample_rate: + pad_len = sample_rate - waveform.shape[1] + waveform = torch.nn.functional.pad( + waveform, + (0, pad_len) + ) + + return waveform - return audio_tensor -def save_audio(audio_tensor, path, sample_rate=SAMPLE_RATE): +def save_audio( + waveform, + path, + sample_rate=SAMPLE_RATE +): """ - Save a mono waveform tensor to a WAV file. + Save a waveform tensor as a WAV file. Parameters ---------- - audio_tensor : torch.Tensor - Shape (1, T) or (T,) + waveform : torch.Tensor + Audio waveform of shape (1, T) or (T,). path : str - Output WAV filename. + Output WAV file path. sample_rate : int Sample rate. """ - if audio_tensor.ndim == 1: - audio_tensor = audio_tensor.unsqueeze(0) - torchaudio.save(path, audio_tensor.cpu(), sample_rate) \ No newline at end of file + if waveform.ndim == 2: + waveform = waveform.squeeze(0) + + sf.write( + path, + waveform.cpu().numpy(), + sample_rate + ) \ No newline at end of file diff --git a/src/vocalid/auth_adapter.py b/src/vocalid/auth_adapter.py index a0c5fc6..58e0b3f 100644 --- a/src/vocalid/auth_adapter.py +++ b/src/vocalid/auth_adapter.py @@ -32,7 +32,17 @@ def authenticate_file(self, audio_path: str) -> AuthenticationResult: def authenticate_live(self, seconds: float = None) -> AuthenticationResult: from .recorder import record_one + seconds = seconds or 4.0 - audio = record_one(seconds, config.SAMPLE_RATE) - verified, score = self.verifier.verify_live(audio) - return AuthenticationResult(success=verified, confidence=score) + + audio = record_one( + seconds, + config.SAMPLE_RATE + ) + + verified, score = self.verifier.verify_array(audio) + + return AuthenticationResult( + success=verified, + confidence=score + ) \ No newline at end of file diff --git a/src/vocalid/duplicate_detector.py b/src/vocalid/duplicate_detector.py deleted file mode 100644 index 4b823fd..0000000 --- a/src/vocalid/duplicate_detector.py +++ /dev/null @@ -1,46 +0,0 @@ -""" -duplicate_detector.py -Flags a new clip as a duplicate if it's too similar (by embedding -cosine similarity) to a clip already in the dataset, or to another -clip recorded in the same enrollment batch. This is the same -embedding extractor the trainer/verifier already use, so no extra -model is loaded. -""" - -import numpy as np -from .embeddings import EmbeddingExtractor -from .audio_utils import load_audio - -DUPLICATE_THRESHOLD = 0.97 # cosine similarity above this = "same clip" - - -def _cosine(a, b) -> float: - a = np.asarray(a).flatten() - b = np.asarray(b).flatten() - denom = (np.linalg.norm(a) * np.linalg.norm(b)) or 1e-8 - return float(np.dot(a, b) / denom) - - -class DuplicateDetector: - def __init__(self, threshold: float = DUPLICATE_THRESHOLD): - self.extractor = EmbeddingExtractor() - self.threshold = threshold - - def _embed_file(self, path: str): - audio, sample_rate = load_audio(path) - return self.extractor.extract(audio) - - def find_duplicate(self, new_path: str, existing_paths: list[str]): - """ - Compares `new_path` against every clip in `existing_paths`. - Returns the path of the first near-duplicate found, or None. - """ - new_emb = self._embed_file(new_path) - for path in existing_paths: - sim = _cosine(new_emb, self._embed_file(path)) - if sim >= self.threshold: - return path - return None - - def is_duplicate(self, new_path: str, existing_paths: list[str]) -> bool: - return self.find_duplicate(new_path, existing_paths) is not None diff --git a/src/vocalid/embeddings.py b/src/vocalid/embeddings.py index 55df8c5..1b02ae2 100644 --- a/src/vocalid/embeddings.py +++ b/src/vocalid/embeddings.py @@ -1,23 +1,109 @@ +# import torch +# from .audio_utils import load_audio +# import numpy as np + + +# class EmbeddingExtractor: +# def __init__(self, model_path="speechbrain/spkrec-ecapa-voxceleb"): +# # lazy torchaudio patch +# try: +# import torchaudio +# if not hasattr(torchaudio, "list_audio_backends"): +# torchaudio.list_audio_backends = lambda: ["sox_io"] +# except Exception: +# pass + +# try: +# from speechbrain.inference import EncoderClassifier +# except Exception as e: +# raise ImportError( +# "SpeechBrain could not load. Torchaudio is incompatible.\n" +# f"Original error: {e}" +# ) + +# self.model = EncoderClassifier.from_hparams( +# source=model_path, +# run_opts={"device": "cpu"}, +# savedir="pretrained_models/ecapa", +# ) + +# def _prepare_waveform(self, wav): +# # Convert numpy → tensor +# if not isinstance(wav, torch.Tensor): +# wav = torch.tensor(wav, dtype=torch.float32) + +# # Ensure shape (1, T) +# if wav.ndim == 1: +# wav = wav.unsqueeze(0) + +# # If multichannel, average +# if wav.shape[0] > 1: +# wav = wav.mean(dim=0, keepdim=True) + +# # Pad waves shorter than 1 sec (ECAPA expects enough frames) +# min_len = 16000 # 1 s at 16kHz +# if wav.shape[1] < min_len: +# pad_len = min_len - wav.shape[1] +# wav = torch.nn.functional.pad(wav, (0, pad_len)) + +# return wav + +# def _normalize(self, emb): +# emb = np.asarray(emb).squeeze() +# norm = np.linalg.norm(emb) +# if norm == 0: +# return emb +# return emb / norm + +# def embed_file(self, path): +# try: +# waveform, _ = load_audio(path) +# waveform = self._prepare_waveform(waveform) + +# with torch.no_grad(): +# emb = self.model.encode_batch(waveform)[0].cpu().numpy() + +# return self._normalize(emb) + +# except Exception as e: +# print(f"\nEmbedding failed for: {path}") +# raise +# def emb_waveform(self, waveform): +# waveform = self._prepare_waveform(waveform) + +# with torch.no_grad(): +# emb = self.model.encode_batch(waveform)[0].cpu().numpy() + +# return self._normalize(emb) + +# def extract(self, waveform): +# return self.emb_waveform(waveform) + + import torch -from .audio_utils import load_audio import numpy as np +from .audio_utils import load_audio + class EmbeddingExtractor: def __init__(self, model_path="speechbrain/spkrec-ecapa-voxceleb"): - # lazy torchaudio patch + # Lazy torchaudio compatibility patch try: import torchaudio + if not hasattr(torchaudio, "list_audio_backends"): torchaudio.list_audio_backends = lambda: ["sox_io"] + except Exception: pass try: from speechbrain.inference import EncoderClassifier + except Exception as e: raise ImportError( - "SpeechBrain could not load. Torchaudio is incompatible.\n" + "SpeechBrain could not load.\n" f"Original error: {e}" ) @@ -27,53 +113,98 @@ def __init__(self, model_path="speechbrain/spkrec-ecapa-voxceleb"): savedir="pretrained_models/ecapa", ) - def _prepare_waveform(self, wav): - # Convert numpy → tensor - if not isinstance(wav, torch.Tensor): - wav = torch.tensor(wav, dtype=torch.float32) + def _prepare_waveform(self, waveform): + """ + Convert waveform into the format expected by SpeechBrain. + + Expected shape: + (1, T) + """ + + if not isinstance(waveform, torch.Tensor): + waveform = torch.tensor( + waveform, + dtype=torch.float32 + ) - # Ensure shape (1, T) - if wav.ndim == 1: - wav = wav.unsqueeze(0) + # Convert (T,) -> (1, T) + if waveform.ndim == 1: + waveform = waveform.unsqueeze(0) - # If multichannel, average - if wav.shape[0] > 1: - wav = wav.mean(dim=0, keepdim=True) + # Stereo -> Mono + if waveform.shape[0] > 1: + waveform = waveform.mean( + dim=0, + keepdim=True + ) - # Pad waves shorter than 1 sec (ECAPA expects enough frames) - min_len = 16000 # 1 s at 16kHz - if wav.shape[1] < min_len: - pad_len = min_len - wav.shape[1] - wav = torch.nn.functional.pad(wav, (0, pad_len)) + # Pad to at least 1 second + min_len = 16000 - return wav + if waveform.shape[1] < min_len: + waveform = torch.nn.functional.pad( + waveform, + (0, min_len - waveform.shape[1]) + ) + + return waveform + + def _normalize(self, embedding): + embedding = np.asarray(embedding).squeeze() + + norm = np.linalg.norm(embedding) - def _normalize(self, emb): - emb = np.asarray(emb).squeeze() - norm = np.linalg.norm(emb) if norm == 0: - return emb - return emb / norm + return embedding + + return embedding / norm def embed_file(self, path): + """ + Load a WAV file and return a normalized speaker embedding. + """ + try: - waveform = load_audio(path) + waveform, _ = load_audio(path) + waveform = self._prepare_waveform(waveform) - except Exception: - return None - with torch.no_grad(): - emb = self.model.encode_batch(waveform)[0].cpu().numpy() + with torch.no_grad(): + embedding = ( + self.model + .encode_batch(waveform) + .squeeze() + .cpu() + .numpy() + ) + + return self._normalize(embedding) - return self._normalize(emb) + except Exception as e: + print(f"\nEmbedding failed for file: {path}") + print(f"Reason: {e}") + raise def emb_waveform(self, waveform): + """ + Create an embedding directly from a waveform tensor. + """ + waveform = self._prepare_waveform(waveform) with torch.no_grad(): - emb = self.model.encode_batch(waveform)[0].cpu().numpy() + embedding = ( + self.model + .encode_batch(waveform) + .squeeze() + .cpu() + .numpy() + ) - return self._normalize(emb) + return self._normalize(embedding) def extract(self, waveform): + """ + Compatibility wrapper. + """ return self.emb_waveform(waveform) \ No newline at end of file diff --git a/src/vocalid/enroll_and_authenticate.py b/src/vocalid/enroll_and_authenticate.py index a501bbe..88db58e 100644 --- a/src/vocalid/enroll_and_authenticate.py +++ b/src/vocalid/enroll_and_authenticate.py @@ -9,9 +9,9 @@ import glob -from voice_verifier.enrollment import EnrollmentSession -from voice_verifier.trainer import VoiceTrainer -from voice_verifier.auth_adapter import VoiceAuthenticator +from vocalid.enrollment import EnrollmentSession +from vocalid.trainer import VoiceTrainer +from vocalid.auth_adapter import VoiceAuthenticator DATASET_ROOT = "dataset" MODEL_PATH = "my_voice_model.pkl" diff --git a/src/vocalid/enrollment.py b/src/vocalid/enrollment.py index ffb4872..fd1b65b 100644 --- a/src/vocalid/enrollment.py +++ b/src/vocalid/enrollment.py @@ -1,12 +1,12 @@ """ enrollment.py -Ties recorder + validator + duplicate_detector + dataset_manager + +Ties recorder + validator + dataset_manager together into one guided session: - record clip -> validate -> check duplicate -> save to dataset + record clip -> validate -> save to dataset -Retries a rejected clip instead of silently skipping it, so the user -always ends up with the number of samples they asked for. +Retries a rejected clip instead of silently skipping it. """ import os @@ -14,57 +14,163 @@ from .recorder import record_one from .validator import check_audio -from .duplicate_detector import DuplicateDetector from .dataset_manager import DatasetManager from . import config class EnrollmentSession: - def __init__(self, dataset_root: str = "dataset", max_attempts_per_sample: int = 3): + + def __init__( + self, + dataset_root: str = "dataset", + max_attempts_per_sample: int = 3 + ): + self.dataset = DatasetManager(dataset_root) - self.dupe_detector = DuplicateDetector() self.max_attempts_per_sample = max_attempts_per_sample - def enroll(self, label: str, count: int = 10, seconds: float = 5.0) -> list[str]: + def enroll( + self, + label: str, + count: int = 10, + seconds: float = 5.0 + ) -> list[str]: + """ - Records and accepts `count` valid, non-duplicate clips for - `label` ("positive" or "negative"). Returns the saved paths. + Records and accepts `count` valid clips for + `label` ("positive" or "negative"). + + Returns saved file paths. + + Label options: + + positive: + Target/enrolled speaker voice samples + + negative: + Other speaker voice samples + + Flow: + + record clip + | + validate audio quality + | + save accepted clip """ + + # Validate label before doing anything + if label not in ("positive", "negative"): + raise ValueError( + "label must be 'positive' or 'negative'" + ) + + print(""" +================================================== + VocalID Enrollment Instructions +================================================== + +Label options: + + positive -> Target/enrolled speaker voice samples + + negative -> Other speaker voice samples + + +Enrollment flow: + + record clip + | + validate audio quality + | + save accepted clip + + +Recording guidelines: + + - Duration: 4-6.5 seconds + - Speak naturally + - Avoid silence + - Avoid background noise + +================================================== +""") + saved_paths = [] for i in range(count): + accepted = False - for attempt in range(1, self.max_attempts_per_sample + 1): - print(f"\nSample {i + 1}/{count} (attempt {attempt}) - label: {label}") - audio = record_one(seconds, config.SAMPLE_RATE) + for attempt in range( + 1, + self.max_attempts_per_sample + 1 + ): + + print( + f"\nSample {i + 1}/{count} " + f"(attempt {attempt}) " + f"- label: {label}" + ) + + audio = record_one( + seconds, + config.SAMPLE_RATE + ) + + with tempfile.NamedTemporaryFile( + suffix=".wav", + delete=False + ) as tmp: - with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: tmp_path = tmp.name + from .audio_utils import save_audio - save_audio(audio, tmp_path, config.SAMPLE_RATE) - is_valid, reason = check_audio(tmp_path) + save_audio( + audio, + tmp_path, + config.SAMPLE_RATE + ) + + # Validate audio + is_valid, reason = check_audio( + tmp_path + ) + if not is_valid: - print(f"Rejected: {reason}. Try again.") - os.remove(tmp_path) - continue - existing = self.dataset.list_samples(label) + saved_paths - dupe = self.dupe_detector.find_duplicate(tmp_path, existing) - if dupe: - print(f"Rejected: too similar to {os.path.basename(dupe)}. Try again.") + print( + f"Rejected: {reason}. Try again." + ) + os.remove(tmp_path) continue - saved_path = self.dataset.add_sample(tmp_path, label) + # Save accepted sample + saved_path = self.dataset.add_sample( + tmp_path, + label + ) + os.remove(tmp_path) - saved_paths.append(saved_path) - print(f"Saved as {saved_path}") + + saved_paths.append( + saved_path + ) + + print( + f"Saved as {saved_path}" + ) + accepted = True break if not accepted: - print(f"Giving up on sample {i + 1} after {self.max_attempts_per_sample} attempts.") - return saved_paths + print( + f"Giving up on sample {i + 1} " + f"after {self.max_attempts_per_sample} attempts." + ) + + return saved_paths \ No newline at end of file diff --git a/src/vocalid/recorder.py b/src/vocalid/recorder.py index 0dddd02..e4389ab 100644 --- a/src/vocalid/recorder.py +++ b/src/vocalid/recorder.py @@ -1,43 +1,113 @@ """ recorder.py -Records short audio clips one at a time for voice enrollment. -Assumes voice_verifier.audio_utils already exposes a `record_audio` -function (mic -> tensor/np array) and a `save_audio` function, -matching what the library's live-verification path uses. -If your actual audio_utils.py names these differently, just -adjust the two imports below. +Records audio clips for enrollment. """ + import os + from .audio_utils import record_audio, save_audio from . import config -def record_one(seconds: float = 5.0, sample_rate: int = None): - """Record a single clip and return the raw audio tensor.""" - sample_rate = sample_rate or config.SAMPLE_RATE - print(f"Recording for {seconds:.1f}s... speak naturally.") - audio = record_audio(duration=seconds, sample_rate=sample_rate) - print("Done.") + +def record_one( + seconds: float = 5.0, + sample_rate: int = None +): + + """ + Record one audio clip. + """ + + sample_rate = ( + sample_rate + or config.SAMPLE_RATE + ) + + + print( + f"Recording for {seconds:.1f}s... speak naturally." + ) + + + audio = record_audio( + duration=seconds, + sample_rate=sample_rate + ) + + + print( + "Done." + ) + + return audio -def record_batch(out_dir: str, count: int = 10, seconds: float = 5.0, - sample_rate: int = None, prefix: str = "sample"): + + +def record_batch( + out_dir: str, + count: int = 10, + seconds: float = 5.0, + sample_rate: int = None, + prefix: str = "sample" +): + """ - Record `count` clips back to back, saving each to `out_dir`. - Returns the list of saved file paths (unvalidated). + Record multiple clips and save them. + + Returns: + list of saved paths """ - os.makedirs(out_dir, exist_ok=True) - sample_rate = sample_rate or config.SAMPLE_RATE + + + os.makedirs( + out_dir, + exist_ok=True + ) + + + sample_rate = ( + sample_rate + or config.SAMPLE_RATE + ) + + paths = [] + for i in range(count): - input(f"[{i + 1}/{count}] Press Enter, then start speaking...") - audio = record_one(seconds, sample_rate) - path = os.path.join(out_dir, f"{prefix}_{i + 1:03d}.wav") - save_audio(audio, path, sample_rate) - paths.append(path) - return paths + input( + f"[{i+1}/{count}] Press Enter, then start speaking..." + ) + + + audio = record_one( + seconds, + sample_rate + ) + + + path = os.path.join( + out_dir, + f"{prefix}_{i+1:03d}.wav" + ) + + + save_audio( + audio, + path, + sample_rate + ) + + + paths.append( + path + ) + + + return paths \ No newline at end of file diff --git a/src/vocalid/trainer.py b/src/vocalid/trainer.py index 7dfc6fa..4abe883 100644 --- a/src/vocalid/trainer.py +++ b/src/vocalid/trainer.py @@ -31,7 +31,9 @@ def train(self, positive_paths, negative_paths, save_path="voice_auth.pkl"): y.append(1) for p in negative_paths: + print(f"Loading {p}") emb = self.extractor.embed_file(p) + print(type(emb), None if emb is None else emb.shape) if emb is None: continue X.append(normalize(emb)) diff --git a/src/vocalid/verifier.py b/src/vocalid/verifier.py index 12748c7..4745dc0 100644 --- a/src/vocalid/verifier.py +++ b/src/vocalid/verifier.py @@ -10,7 +10,7 @@ def __init__(self, model_path): self.extractor = EmbeddingExtractor() def verify_file(self, file_path, threshold=THRESHOLD): - wav = load_audio(file_path) # waveform tensor + wav,_ = load_audio(file_path) # waveform tensor emb = self.extractor.emb_waveform(wav) # embedding as numpy score = self.model.predict_proba([emb])[0][1] verified = score >= threshold diff --git a/tests/test_auth_adapter.py b/tests/test_auth_adapter.py index d7de448..6ea7675 100644 --- a/tests/test_auth_adapter.py +++ b/tests/test_auth_adapter.py @@ -1,14 +1,18 @@ """ test_auth_adapter.py -VoiceVerifier is mocked out so this tests the adapter's translation -of (verified, score) -> AuthenticationResult, not the real -verification model. +VoiceVerifier is mocked so these tests only verify that +VoiceAuthenticator correctly converts verification results +into AuthenticationResult objects. """ import pytest + import vocalid.auth_adapter as auth_module -from vocalid.auth_adapter import VoiceAuthenticator, AuthenticationResult +from vocalid.auth_adapter import ( + VoiceAuthenticator, + AuthenticationResult, +) class _FakeVerifier: @@ -18,52 +22,82 @@ def __init__(self, model_path): def verify_file(self, path): return True, 0.87 - def verify_live(self, audio): + def verify_array(self, audio): return False, 0.21 @pytest.fixture def authenticator(monkeypatch): - monkeypatch.setattr(auth_module, "VoiceVerifier", _FakeVerifier) + monkeypatch.setattr( + auth_module, + "VoiceVerifier", + _FakeVerifier + ) + return VoiceAuthenticator("dummy_model.pkl") def test_authenticate_file_returns_granted_result(authenticator): + result = authenticator.authenticate_file("some_clip.wav") + assert isinstance(result, AuthenticationResult) assert result.success is True assert result.confidence == 0.87 -def test_authenticate_live_returns_denied_result(authenticator, monkeypatch): - # authenticate_live does `from .recorder import record_one` internally, - # so the real patch point is vocalid.recorder.record_one. - monkeypatch.setattr("vocalid.recorder.record_one", - lambda seconds, sample_rate: "fake_audio") +def test_authenticate_live_returns_denied_result( + authenticator, + monkeypatch +): + + monkeypatch.setattr( + "vocalid.recorder.record_one", + lambda seconds, sample_rate: "fake_audio" + ) - result = authenticator.authenticate_live(seconds=4.0) + result = authenticator.authenticate_live( + seconds=4.0 + ) assert result.success is False assert result.confidence == 0.21 -def test_authenticate_live_defaults_to_four_seconds(authenticator, monkeypatch): +def test_authenticate_live_defaults_to_four_seconds( + authenticator, + monkeypatch +): + captured = {} def fake_record_one(seconds, sample_rate): captured["seconds"] = seconds return "fake_audio" - monkeypatch.setattr("vocalid.recorder.record_one", fake_record_one) - authenticator.authenticate_live(seconds=None) + monkeypatch.setattr( + "vocalid.recorder.record_one", + fake_record_one + ) + + authenticator.authenticate_live() assert captured["seconds"] == 4.0 def test_result_str_formatting(): - granted = AuthenticationResult(success=True, confidence=0.9123) - denied = AuthenticationResult(success=False, confidence=0.1) + + granted = AuthenticationResult( + success=True, + confidence=0.9123 + ) + + denied = AuthenticationResult( + success=False, + confidence=0.1 + ) assert "ACCESS GRANTED" in str(granted) assert "0.91" in str(granted) - assert "ACCESS DENIED" in str(denied) + + assert "ACCESS DENIED" in str(denied) \ No newline at end of file diff --git a/tests/test_duplicate_detector.py b/tests/test_duplicate_detector.py deleted file mode 100644 index 05bf35b..0000000 --- a/tests/test_duplicate_detector.py +++ /dev/null @@ -1,71 +0,0 @@ -""" -test_duplicate_detector.py - -The real EmbeddingExtractor loads a full ECAPA-TDNN model from -speechbrain, which is slow and needs a network connection on first -run. For unit tests we swap it for a fake extractor that just turns -the raw audio into its "embedding" - two clips with the same tone -frequency come out identical, different frequencies come out -different, which is enough to exercise the cosine-similarity logic -in duplicate_detector.py itself. Swap this fixture out once you're -ready to write a slower, real-model integration test. -""" - -import numpy as np -import soundfile as sf -import pytest - -import vocalid.duplicate_detector as dd_module -from vocalid.duplicate_detector import DuplicateDetector - - -class _FakeExtractor: - def extract(self, audio): - return np.asarray(audio).flatten() - - -def _fake_load_audio(path): - samples, sample_rate = sf.read(path, dtype="float32") - return samples, sample_rate - - -@pytest.fixture -def detector(monkeypatch): - monkeypatch.setattr(dd_module, "EmbeddingExtractor", _FakeExtractor) - monkeypatch.setattr(dd_module, "load_audio", _fake_load_audio) - return DuplicateDetector() - - -def test_identical_clip_is_flagged(detector, write_wav): - original = write_wav("original.wav", freq=220.0) - replay = write_wav("replay.wav", freq=220.0) # same tone == same "voice" - - match = detector.find_duplicate(replay, [original]) - assert match == original - assert detector.is_duplicate(replay, [original]) is True - - -def test_different_clip_is_not_flagged(detector, write_wav): - original = write_wav("original.wav", freq=220.0) - different = write_wav("different.wav", freq=880.0) # different tone == different "voice" - - match = detector.find_duplicate(different, [original]) - assert match is None - assert detector.is_duplicate(different, [original]) is False - - -def test_no_existing_clips_never_flags(detector, write_wav): - clip = write_wav("only.wav") - assert detector.find_duplicate(clip, []) is None - - -def test_custom_threshold_is_respected(write_wav, monkeypatch): - monkeypatch.setattr(dd_module, "EmbeddingExtractor", _FakeExtractor) - monkeypatch.setattr(dd_module, "load_audio", _fake_load_audio) - - strict_detector = DuplicateDetector(threshold=0.999999) - original = write_wav("original.wav", freq=220.0, amplitude=0.3) - slightly_different = write_wav("close.wav", freq=222.0, amplitude=0.3) - - # Close but not identical tones shouldn't clear an extremely strict threshold. - assert strict_detector.is_duplicate(slightly_different, [original]) is False diff --git a/tests/test_enrollment.py b/tests/test_enrollment.py index f77c40a..653ec85 100644 --- a/tests/test_enrollment.py +++ b/tests/test_enrollment.py @@ -1,86 +1,151 @@ """ test_enrollment.py -record_one, audio_utils.save_audio, and DuplicateDetector are all -mocked so this exercises EnrollmentSession's retry/accept/give-up -logic without needing a mic, a real audio file, or the ECAPA model. -DatasetManager is used for real (against tmp_path) since it's just -file I/O and worth testing end to end here. +Unit tests for EnrollmentSession. + +External dependencies are mocked: + +- recorder.record_one +- validator.check_audio +- audio_utils.save_audio + +DatasetManager is used normally against pytest's temporary directory. """ import pytest + import vocalid.enrollment as enrollment_module from vocalid.enrollment import EnrollmentSession -class _FakeDupeDetector: - """queue of paths to report as duplicates, one per find_duplicate() call""" - def __init__(self, duplicate_paths=None): - self._duplicate_paths = list(duplicate_paths or []) - - def find_duplicate(self, new_path, existing_paths): - return self._duplicate_paths.pop(0) if self._duplicate_paths else None - +# --------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------- @pytest.fixture(autouse=True) def patch_recording(monkeypatch): - monkeypatch.setattr(enrollment_module, "record_one", lambda seconds, sample_rate: "fake_audio") - # save_audio is imported lazily inside enroll(), so the real patch - # point is the audio_utils module itself, not enrollment_module. - monkeypatch.setattr("vocalid.audio_utils.save_audio", lambda audio, path, sr: None) + + monkeypatch.setattr( + enrollment_module, + "record_one", + lambda seconds, sample_rate: "fake_audio" + ) + + monkeypatch.setattr( + "vocalid.audio_utils.save_audio", + lambda audio, path, sr: None + ) -def test_accepts_a_valid_unique_clip_on_first_try(monkeypatch, tmp_path): - monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) - monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) +# --------------------------------------------------------- +# Tests +# --------------------------------------------------------- - session = EnrollmentSession(dataset_root=str(tmp_path / "dataset")) - saved = session.enroll(label="positive", count=1, seconds=5.0) +def test_accepts_valid_clip(monkeypatch, tmp_path): - assert len(saved) == 1 + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: (True, "ok") + ) + session = EnrollmentSession( + dataset_root=str(tmp_path / "dataset") + ) -def test_retries_after_a_rejected_clip(monkeypatch, tmp_path): - results = iter([(False, "too quiet"), (True, "ok")]) - monkeypatch.setattr(enrollment_module, "check_audio", lambda path: next(results)) - monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + saved = session.enroll( + label="positive", + count=1, + seconds=5.0 + ) - session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=3) - saved = session.enroll(label="positive", count=1, seconds=5.0) + assert len(saved) == 1 - assert len(saved) == 1 # accepted on the second attempt +def test_retries_after_validation_failure(monkeypatch, tmp_path): + + results = iter([ + (False, "too quiet"), + (True, "ok") + ]) -def test_retries_after_a_duplicate_clip(monkeypatch, tmp_path): - monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) - # first attempt flagged as a duplicate, second attempt is clean monkeypatch.setattr( - enrollment_module, "DuplicateDetector", - lambda: _FakeDupeDetector(duplicate_paths=["dataset/my_voice/sample001.wav"]) + enrollment_module, + "check_audio", + lambda path: next(results) + ) + + session = EnrollmentSession( + dataset_root=str(tmp_path / "dataset"), + max_attempts_per_sample=3 ) - session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=3) - saved = session.enroll(label="positive", count=1, seconds=5.0) + saved = session.enroll( + label="positive", + count=1, + seconds=5.0 + ) assert len(saved) == 1 def test_gives_up_after_max_attempts(monkeypatch, tmp_path): - monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (False, "too quiet")) - monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) - session = EnrollmentSession(dataset_root=str(tmp_path / "dataset"), max_attempts_per_sample=2) - saved = session.enroll(label="positive", count=1, seconds=5.0) + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: (False, "too quiet") + ) + + session = EnrollmentSession( + dataset_root=str(tmp_path / "dataset"), + max_attempts_per_sample=2 + ) + + saved = session.enroll( + label="positive", + count=1, + seconds=5.0 + ) + + assert saved == [] + - assert saved == [] # never accepted, session moved on +def test_returns_path_for_every_sample(monkeypatch, tmp_path): + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: (True, "ok") + ) -def test_returns_a_path_for_every_accepted_sample(monkeypatch, tmp_path): - monkeypatch.setattr(enrollment_module, "check_audio", lambda path: (True, "ok")) - monkeypatch.setattr(enrollment_module, "DuplicateDetector", lambda: _FakeDupeDetector()) + session = EnrollmentSession( + dataset_root=str(tmp_path / "dataset") + ) - session = EnrollmentSession(dataset_root=str(tmp_path / "dataset")) - saved = session.enroll(label="negative", count=3, seconds=5.0) + saved = session.enroll( + label="negative", + count=3, + seconds=5.0 + ) assert len(saved) == 3 - assert len(set(saved)) == 3 # all distinct paths, no overwrites + assert len(set(saved)) == 3 + + +def test_invalid_label_is_rejected(tmp_path): + + session = EnrollmentSession( + dataset_root=str(tmp_path / "dataset") + ) + + with pytest.raises( + ValueError, + match="label must be 'positive' or 'negative'" + ): + + session.enroll( + label="kai", + count=1, + seconds=5.0 + ) \ No newline at end of file