diff --git a/dummy_model.pkl b/dummy_model.pkl new file mode 100644 index 0000000..feab21d Binary files /dev/null and b/dummy_model.pkl differ diff --git a/src/vocalid/audio_utils.py b/src/vocalid/audio_utils.py index 6227b1b..7e605a3 100644 --- a/src/vocalid/audio_utils.py +++ b/src/vocalid/audio_utils.py @@ -1,62 +1,150 @@ import torch import torchaudio import sounddevice as sd -import numpy as np +import soundfile as sf + from .config import SAMPLE_RATE def load_audio(path, target_sr=SAMPLE_RATE): - waveform, sr = torchaudio.load(path) + """ + Load a WAV file and return a mono waveform tensor. - # Convert to float32 - waveform = waveform.float() + Parameters + ---------- + path : str + Path to the WAV file. + target_sr : int + Desired sample rate. - # Resample if needed - if sr != target_sr: - waveform = torchaudio.functional.resample(waveform, sr, target_sr) + Returns + ------- + tuple(torch.Tensor, int) + Waveform with shape (1, T) and sample rate. + """ + + samples, sr = sf.read( + path, + dtype="float32", + always_2d=True + ) - # Mono - if waveform.ndim == 2 and waveform.shape[0] > 1: + waveform = torch.from_numpy(samples.T).float() + + # 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( + 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). + """ -def record_audio(seconds=4, fs=SAMPLE_RATE): - print(f"Recording {seconds} seconds...") + print(f"Recording {duration:.1f} seconds...") - # Check if sounddevice sees any input devices 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 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 + + +def save_audio( + waveform, + path, + sample_rate=SAMPLE_RATE +): + """ + Save a waveform tensor as a WAV file. + + Parameters + ---------- + waveform : torch.Tensor + Audio waveform of shape (1, T) or (T,). + path : str + Output WAV file path. + sample_rate : int + Sample rate. + """ - # 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)) + if waveform.ndim == 2: + waveform = waveform.squeeze(0) - return audio_tensor + 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 new file mode 100644 index 0000000..58e0b3f --- /dev/null +++ b/src/vocalid/auth_adapter.py @@ -0,0 +1,48 @@ +""" +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_array(audio) + + return AuthenticationResult( + success=verified, + confidence=score + ) \ No newline at end of file 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/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 new file mode 100644 index 0000000..88db58e --- /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 vocalid.enrollment import EnrollmentSession +from vocalid.trainer import VoiceTrainer +from vocalid.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..fd1b65b --- /dev/null +++ b/src/vocalid/enrollment.py @@ -0,0 +1,176 @@ +""" +enrollment.py + +Ties recorder + validator + dataset_manager +together into one guided session: + + record clip -> validate -> save to dataset + +Retries a rejected clip instead of silently skipping it. +""" + +import os +import tempfile + +from .recorder import record_one +from .validator import check_audio +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.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 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} " + f"(attempt {attempt}) " + f"- 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 + ) + + # Validate audio + is_valid, reason = check_audio( + tmp_path + ) + + if not is_valid: + + print( + f"Rejected: {reason}. Try again." + ) + + os.remove(tmp_path) + continue + + # 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}" + ) + + accepted = True + break + + if not accepted: + + 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 new file mode 100644 index 0000000..e4389ab --- /dev/null +++ b/src/vocalid/recorder.py @@ -0,0 +1,113 @@ +""" +recorder.py + +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 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" +): + + """ + 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 + ) + + + 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 \ 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/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/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/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..6ea7675 --- /dev/null +++ b/tests/test_auth_adapter.py @@ -0,0 +1,103 @@ +""" +test_auth_adapter.py + +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, +) + + +class _FakeVerifier: + def __init__(self, model_path): + self.model_path = model_path + + def verify_file(self, path): + return True, 0.87 + + def verify_array(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 +): + + 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() + + 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) \ No newline at end of file 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_enrollment.py b/tests/test_enrollment.py new file mode 100644 index 0000000..653ec85 --- /dev/null +++ b/tests/test_enrollment.py @@ -0,0 +1,151 @@ +""" +test_enrollment.py + +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 + + +# --------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------- + +@pytest.fixture(autouse=True) +def patch_recording(monkeypatch): + + monkeypatch.setattr( + enrollment_module, + "record_one", + lambda seconds, sample_rate: "fake_audio" + ) + + monkeypatch.setattr( + "vocalid.audio_utils.save_audio", + lambda audio, path, sr: None + ) + + +# --------------------------------------------------------- +# Tests +# --------------------------------------------------------- + +def test_accepts_valid_clip(monkeypatch, tmp_path): + + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: (True, "ok") + ) + + 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_validation_failure(monkeypatch, tmp_path): + + results = iter([ + (False, "too quiet"), + (True, "ok") + ]) + + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: next(results) + ) + + 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") + ) + + 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 == [] + + +def test_returns_path_for_every_sample(monkeypatch, tmp_path): + + monkeypatch.setattr( + enrollment_module, + "check_audio", + lambda path: (True, "ok") + ) + + 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 + + +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 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