Feature/voice authentication

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
+    )

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
+    )

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