Feature/voice authentication

.github/workflows/python-package.yml

@@ -43,4 +43,4 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: coverage-${{ matrix.python-version }}
-          path: .coverage.xml
+          path: coverage.xml

src/vocalid/audio_utils.py

@@ -1,62 +1,156 @@
+
+
+import logging
+
 import torch
 import torchaudio
 import sounddevice as sd
-import numpy as np
+import soundfile as sf
+
 from .config import SAMPLE_RATE
 
+logger = logging.getLogger(__name__)
+
 
 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.
 
-def record_audio(seconds=4, fs=SAMPLE_RATE):
-    print(f"Recording {seconds} seconds...")
+    Parameters
+    ----------
+    duration : float
+        Recording duration in seconds.
+    sample_rate : int
+        Recording sample rate.
+
+    Returns
+    -------
+    torch.Tensor
+        Recorded waveform with shape (1, T).
+    """
+
+    logger.info("Recording %.1f seconds...", duration)
 
-    # 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
+
+    logger.info("Recording complete")
+
+    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
+
 
-    print("Recording complete")
+def save_audio(
+    waveform,
+    path,
+    sample_rate=SAMPLE_RATE
+):
+    """
+    Save a waveform tensor as a WAV file.
 
-    audio_tensor = torch.tensor(audio.T, dtype=torch.float32)
+    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/cli.py

@@ -1,74 +1,96 @@
+
+
 import argparse
+import logging
 import os
 from glob import glob
+
+from .audio_utils import record_audio
 from .trainer import VoiceTrainer
 from .verifier import VoiceVerifier
-from .audio_utils import record_audio
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(message)s",
+)
+
+logger = logging.getLogger(__name__)
+
 
 def main():
     parser = argparse.ArgumentParser(description="Voice Verifier CLI")
     sub = parser.add_subparsers(dest="commands")
 
     # training command
     train = sub.add_parser("train", help="Train a voice authentication model")
-    train.add_argument("--positive", required=True, help = "Folder with your voice samples")
-    train.add_argument("--negative", required=True, help = "Folder with other voices")
-    train.add_argument("--output", default="voice_auth.pkl", help="path to save model")
+    train.add_argument("--positive", required=True, help="Folder with your voice samples")
+    train.add_argument("--negative", required=True, help="Folder with other voices")
+    train.add_argument("--output", default="voice_auth.pkl", help="Path to save model")
 
     # evaluate the model
-    evaluate = sub.add_parser("evaluate", help = "Evaluate the trained model")
-    evaluate.add_argument("--model", required=True, help= "Path to trained model")
-    evaluate.add_argument("--positive", required=True, help = "Folder with your voice samples")
-    evaluate.add_argument("--negative", required=True, help= "Folder with negative/random samples")
+    evaluate = sub.add_parser("evaluate", help="Evaluate the trained model")
+    evaluate.add_argument("--model", required=True, help="Path to trained model")
+    evaluate.add_argument("--positive", required=True, help="Folder with your voice samples")
+    evaluate.add_argument("--negative", required=True, help="Folder with negative/random samples")
 
     # verify file command
-    verify = sub.add_parser("verify", help = "Verify a voice file")
-    verify.add_argument("file", help="path to .wav voice file to verify")
-    verify.add_argument("--model", default="voice_auth.pkl", help="trained model path")
+    verify = sub.add_parser("verify", help="Verify a voice file")
+    verify.add_argument("file", help="Path to .wav voice file to verify")
+    verify.add_argument("--model", default="voice_auth.pkl", help="Trained model path")
 
     # live verification command
-    live = sub.add_parser("live", help = "Live microphone verification")
-    live.add_argument("--model", default="voice_auth.pkl", help= "trained model path")
-    live.add_argument("--seconds", type=int, default=4, help= "Recording duration in seconds")
+    live = sub.add_parser("live", help="Live microphone verification")
+    live.add_argument("--model", default="voice_auth.pkl", help="Trained model path")
+    live.add_argument("--seconds", type=int, default=4, help="Recording duration in seconds")
 
     args = parser.parse_args()
 
     if args.commands == "train":
         pos_files = glob(os.path.join(args.positive, "*.wav"))
         neg_files = glob(os.path.join(args.negative, "*.wav"))
+
         trainer = VoiceTrainer()
         trainer.train(pos_files, neg_files, args.output)
-        print(f"Model saved to {args.output}")
+
+        logger.info("Model saved to %s", args.output)
 
     elif args.commands == "evaluate":
         pos_files = glob(os.path.join(args.positive, "*.wav"))
         neg_files = glob(os.path.join(args.negative, "*.wav"))
 
         trainer = VoiceTrainer()
         trainer.load(args.model)
-        # X, y = trainer.prepare_features(pos_files, neg_files)
+
         results = trainer.evaluate(pos_files, neg_files)
 
-        print("\n===== Evaluation Results =====")
-        print("\nAccuracy", round(results["accuracy"], 4))
-        print("\nClassification Report:\n")
-        print(results["report"])
+        logger.info("")
+        logger.info("===== Evaluation Results =====")
+        logger.info("")
+        logger.info("Accuracy: %.4f", results["accuracy"])
+        logger.info("")
+        logger.info("Classification Report:")
+        logger.info("%s", results["report"])
 
     elif args.commands == "verify":
         verifier = VoiceVerifier(args.model)
+
         ok, score = verifier.verify_file(args.file)
-        print(f"Verified: {ok}, Score: {score:.2f}")
+
+        logger.info("Verified: %s, Score: %.2f", ok, score)
 
     elif args.commands == "live":
         verifier = VoiceVerifier(args.model)
+
         try:
             audio_tensor = record_audio(args.seconds)
+
         except RuntimeError as e:
-            print(str(e))
+            logger.error("%s", e)
             return
 
         ok, score = verifier.verify_array(audio_tensor)
-        print(f"Verified: {ok}, Score: {score:.2f}")
+
+        logger.info("Verified: %s, Score: %.2f", ok, score)
 
     else:
         parser.print_help()