Initial commit: real-time multi-model baby cry classifier

.gitignore

@@ -0,0 +1,28 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+*.egg-info/
+dist/
+build/
+
+# Virtual environment
+.venv/
+
+# uv
+.python-version
+
+# OS
+.DS_Store
+Thumbs.db
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# Logs
+*.jsonl
+*.log

README.md

@@ -0,0 +1,65 @@
+# 🍼 TotTalk Cry Eval
+
+Real-time multi-model baby cry classification CLI tool. Captures mic audio (or reads a file), runs four open-source models simultaneously on 1-second windows, and displays a live comparison table in the terminal.
+
+## Models
+
+| # | Name | Type | Source | Speed |
+|---|------|------|--------|-------|
+| 1 | **foduucom-SVC** | sklearn SVC, 194-dim MFCC features | [HuggingFace](https://huggingface.co/foduucom/baby-cry-classification) | < 1 ms |
+| 2 | **DistilHuBERT** | DistilHuBERT fine-tune (5 classes) | [HuggingFace](https://huggingface.co/AmeerHesham/distilhubert-finetuned-baby_cry) | ~35 ms |
+| 3 | **Kibalama-9c** | Wav2Vec2 fine-tune (9 classes incl. discomfort, tired, cold/hot) | [HuggingFace](https://huggingface.co/Kibalama/baby_cry_classification_model) | ~90 ms |
+| 4 | **YAMNet-detector** | TF Hub YAMNet (binary cry gate) | [TF Hub](https://tfhub.dev/google/yamnet/1) | < 10 ms |
+
+## Quick start
+
+```bash
+# Install dependencies (using uv)
+cd cry-eval
+uv sync
+
+# Run with mic input
+uv run python main.py
+
+# Run with an audio file
+uv run python main.py --file path/to/cry.wav
+
+# Select specific models
+uv run python main.py --models svc,hubert,kibalama
+
+# Disable YAMNet gating
+uv run python main.py --no-yamnet-gate
+
+# Save predictions to JSONL
+uv run python main.py --save-log results.jsonl
+```
+
+## Requirements
+
+- Python ≥ 3.11
+- A working microphone (for live mode)
+- ~1 GB RAM for transformer models
+
+Model weights are auto-downloaded on first run into HuggingFace/TF Hub caches.
+
+## Project structure
+
+```
+cry-eval/
+├── pyproject.toml
+├── README.md
+├── main.py               # CLI entrypoint
+├── models/
+│   ├── base.py           # abstract CryClassifier + CryPrediction
+│   ├── foduucom_svc.py   # sklearn SVC
+│   ├── wiam_wav2vec2.py  # DistilHuBERT fine-tune
+│   ├── kibalama.py       # Wav2Vec2 9-class fine-tune
+│   ├── yamnet.py         # YAMNet binary detector
+│   └── ensemble.py       # orchestrates all models
+├── audio/
+│   ├── capture.py        # MicCapture + FileCapture
+│   └── preprocess.py     # MFCC, mel, resample, RMS
+├── display/
+│   └── table.py          # Rich live table renderer
+└── weights/              # auto-downloaded (gitignored)
+```

audio/capture.py

@@ -0,0 +1,209 @@
+"""Microphone capture with overlapping sliding windows using sounddevice."""
+
+from __future__ import annotations
+
+import queue
+import threading
+
+import numpy as np
+import sounddevice as sd
+
+from audio.preprocess import SAMPLE_RATE, WINDOW_SECONDS
+
+
+class MicCapture:
+    """Captures audio from the default mic and emits overlapping windows.
+
+    Uses a ring-buffer with a 1-second hop (50 % overlap on 2 s windows) so
+    short or quiet sounds that straddle a boundary are still captured in at
+    least one complete window.
+
+    Parameters
+    ----------
+    sample_rate : int
+        Target sample rate (default 16 000).
+    window_seconds : float
+        Window length in seconds (default 2.0).
+    hop_seconds : float
+        Hop between consecutive emitted windows (default 1.0 → 50 % overlap).
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = SAMPLE_RATE,
+        window_seconds: float = WINDOW_SECONDS,
+        hop_seconds: float = 0.5,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.window_seconds = window_seconds
+        self.hop_seconds = hop_seconds
+
+        self._window_samples = int(sample_rate * window_seconds)
+        self._hop_samples = int(sample_rate * hop_seconds)
+
+        # Ring buffer — pre-allocated numpy array
+        self._buf = np.zeros(self._window_samples, dtype=np.float32)
+        self._write_pos = 0          # how many samples written since last emit
+        self._buf_filled = False     # True once we have at least one full window
+
+        # Thread-safe queue so the main loop can pull complete windows
+        self.window_queue: queue.Queue[np.ndarray] = queue.Queue(maxsize=16)
+
+        self._stream: sd.InputStream | None = None
+        self._running = threading.Event()
+
+    # ── sounddevice callback ──────────────────────────────────────────────
+    def _audio_callback(
+        self,
+        indata: np.ndarray,
+        frames: int,
+        time_info: object,
+        status: sd.CallbackFlags,
+    ) -> None:
+        mono = indata[:, 0].copy()
+        n = len(mono)
+
+        # Shift buffer left and append new samples at the end
+        if n >= self._window_samples:
+            # Chunk larger than window — just keep the last window_samples
+            self._buf[:] = mono[-self._window_samples:]
+            self._write_pos = self._window_samples
+            self._buf_filled = True
+        else:
+            self._buf[:-n] = self._buf[n:]
+            self._buf[-n:] = mono
+            self._write_pos += n
+
+        # After initial fill, emit a window every hop_samples
+        if not self._buf_filled:
+            if self._write_pos >= self._window_samples:
+                self._buf_filled = True
+                self._write_pos = 0
+                self._emit()
+        else:
+            if self._write_pos >= self._hop_samples:
+                self._write_pos -= self._hop_samples
+                self._emit()
+
+    def _emit(self) -> None:
+        window = self._buf.copy()
+        try:
+            self.window_queue.put_nowait(window)
+        except queue.Full:
+            # Drop the oldest unprocessed window to keep latency low
+            try:
+                self.window_queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self.window_queue.put_nowait(window)
+            except queue.Full:
+                pass
+
+    # ── public API ────────────────────────────────────────────────────────
+    def start(self) -> None:
+        """Open the mic stream and begin capturing."""
+        self._running.set()
+        self._stream = sd.InputStream(
+            samplerate=self.sample_rate,
+            channels=1,
+            dtype="float32",
+            blocksize=int(self.sample_rate * 0.1),  # 100 ms blocks
+            callback=self._audio_callback,
+        )
+        self._stream.start()
+
+    def stop(self) -> None:
+        """Stop capturing and close the stream."""
+        self._running.clear()
+        if self._stream is not None:
+            self._stream.stop()
+            self._stream.close()
+            self._stream = None
+
+    @property
+    def is_running(self) -> bool:
+        return self._running.is_set()
+
+
+class FileCapture:
+    """Reads an audio file and emits sliding 2-second windows into a queue.
+
+    Parameters
+    ----------
+    path : str
+        Path to a WAV/FLAC/MP3 file.
+    sample_rate : int
+        Target sample rate.
+    window_seconds : float
+        Window size in seconds.
+    hop_seconds : float
+        Hop between consecutive windows (default 1.0 s for 50 % overlap).
+    loop : bool
+        Whether to loop the file indefinitely.
+    """
+
+    def __init__(
+        self,
+        path: str,
+        sample_rate: int = SAMPLE_RATE,
+        window_seconds: float = WINDOW_SECONDS,
+        hop_seconds: float = 1.0,
+        loop: bool = True,
+    ) -> None:
+        import librosa
+
+        self.path = path
+        self.sample_rate = sample_rate
+        self.window_seconds = window_seconds
+        self.hop_seconds = hop_seconds
+        self.loop = loop
+
+        self._audio, _ = librosa.load(path, sr=sample_rate, mono=True)
+        self._window_samples = int(sample_rate * window_seconds)
+        self._hop_samples = int(sample_rate * hop_seconds)
+        self._total_samples = len(self._audio)
+
+        self.window_queue: queue.Queue[np.ndarray] = queue.Queue(maxsize=16)
+        self._thread: threading.Thread | None = None
+        self._running = threading.Event()
+
+    def _emit_loop(self) -> None:
+        offset = 0
+        while self._running.is_set():
+            end = offset + self._window_samples
+            if end > self._total_samples:
+                if self.loop:
+                    offset = 0
+                    continue
+                else:
+                    break
+            window = self._audio[offset:end].copy()
+            try:
+                self.window_queue.put(window, timeout=0.5)
+            except queue.Full:
+                pass
+            offset += self._hop_samples
+            # Simulate real-time pacing
+            import time
+            time.sleep(self.hop_seconds)
+
+    @property
+    def current_position_seconds(self) -> float:
+        """Approximate playback position — not perfectly precise but useful for display."""
+        return 0.0  # simplified; the thread owns the offset
+
+    def start(self) -> None:
+        self._running.set()
+        self._thread = threading.Thread(target=self._emit_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._running.clear()
+        if self._thread is not None:
+            self._thread.join(timeout=3)
+            self._thread = None
+
+    @property
+    def is_running(self) -> bool:
+        return self._running.is_set()

audio/preprocess.py

@@ -0,0 +1,94 @@
+"""Shared audio preprocessing utilities."""
+
+import numpy as np
+import librosa
+
+# ── Constants ──────────────────────────────────────────────────────────────────
+SAMPLE_RATE = 16_000          # all models normalized to 16 kHz
+WINDOW_SECONDS = 1.0          # inference window size
+SILENCE_RMS_THRESHOLD = 0.001 # skip silent frames (low for phone speaker playback)
+HOP_LENGTH = 512
+N_FFT = 1024
+N_MELS = 128
+N_MFCC = 40
+
+
+def resample(audio_np: np.ndarray, from_sr: int, to_sr: int) -> np.ndarray:
+    """Resample audio from *from_sr* to *to_sr* using librosa."""
+    if from_sr == to_sr:
+        return audio_np
+    return librosa.resample(audio_np, orig_sr=from_sr, target_sr=to_sr)
+
+
+def extract_mfcc_features(
+    audio_np: np.ndarray,
+    sr: int,
+    n_mels: int = N_MELS,
+) -> np.ndarray:
+    """Return a feature vector (MFCCs + chroma + mel + contrast + tonnetz mean).
+
+    Concatenation order matches foduucom/baby-cry-classification training code.
+    ``n_mels`` can be overridden when the SVC model was trained with a different
+    mel band count.
+    """
+    # MFCCs — 40 coeffs
+    mfcc = librosa.feature.mfcc(y=audio_np, sr=sr, n_mfcc=N_MFCC)
+    mfcc_mean = np.mean(mfcc, axis=1)  # (40,)
+
+    # Chroma — 12 bins
+    stft = np.abs(librosa.stft(audio_np, n_fft=N_FFT, hop_length=HOP_LENGTH))
+    chroma = librosa.feature.chroma_stft(S=stft, sr=sr)
+    chroma_mean = np.mean(chroma, axis=1)  # (12,)
+
+    # Mel spectrogram summary
+    mel = librosa.feature.melspectrogram(
+        y=audio_np, sr=sr, n_fft=N_FFT, hop_length=HOP_LENGTH, n_mels=n_mels,
+    )
+    mel_mean = np.mean(mel, axis=1)  # (n_mels,)
+
+    # Spectral contrast — 7 bands
+    contrast = librosa.feature.spectral_contrast(
+        S=stft, sr=sr, n_bands=6, fmin=200.0,
+    )
+    contrast_mean = np.mean(contrast, axis=1)  # (7,)
+
+    # Tonnetz — 6 dims
+    tonnetz = librosa.feature.tonnetz(
+        y=librosa.effects.harmonic(audio_np), sr=sr,
+    )
+    tonnetz_mean = np.mean(tonnetz, axis=1)  # (6,)
+
+    # Order: mfcc, chroma, mel, contrast, tonnetz (matches foduucom training)
+    return np.concatenate([mfcc_mean, chroma_mean, mel_mean, contrast_mean, tonnetz_mean])
+
+
+def extract_mel_spectrogram(audio_np: np.ndarray, sr: int) -> np.ndarray:
+    """Return a mel spectrogram of shape (128, T) as float32."""
+    mel = librosa.feature.melspectrogram(
+        y=audio_np, sr=sr, n_fft=N_FFT, hop_length=HOP_LENGTH, n_mels=N_MELS,
+    )
+    return librosa.power_to_db(mel, ref=np.max).astype(np.float32)
+
+
+def is_silent(audio_np: np.ndarray) -> bool:
+    """Return True when audio RMS is below the silence threshold."""
+    rms = np.sqrt(np.mean(audio_np ** 2))
+    return rms < SILENCE_RMS_THRESHOLD
+
+
+def compute_rms(audio_np: np.ndarray) -> float:
+    """Return the RMS energy of the audio window."""
+    return float(np.sqrt(np.mean(audio_np ** 2)))
+
+
+def normalize_audio(audio_np: np.ndarray) -> np.ndarray:
+    """Peak-normalize audio to [-1, 1].
+
+    Crucial when playing cry samples through a phone speaker → laptop mic,
+    since the captured signal can be very quiet and models perform poorly
+    on low-amplitude inputs.
+    """
+    peak = np.max(np.abs(audio_np))
+    if peak < 1e-6:
+        return audio_np
+    return audio_np / peak

display/table.py

@@ -0,0 +1,173 @@
+"""Rich-based live terminal table for displaying predictions."""
+
+from __future__ import annotations
+
+from collections import deque
+
+from rich.live import Live
+from rich.table import Table
+from rich.text import Text
+
+from models.base import LABEL_EMOJI, LABEL_MEANING, CryPrediction
+from models.ensemble import compute_consensus
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+_BAR_FULL = "█"
+_BAR_EMPTY = "░"
+_BAR_WIDTH = 5
+
+
+def confidence_bar(value: float) -> str:
+    """Render a 5-char Unicode bar for a 0.0–1.0 confidence value."""
+    filled = round(value * _BAR_WIDTH)
+    return _BAR_FULL * filled + _BAR_EMPTY * (_BAR_WIDTH - filled)
+
+
+def format_confidence(value: float) -> str:
+    """Bar + percentage string."""
+    pct = int(value * 100)
+    return f"{confidence_bar(value)} {pct:>3}%"
+
+
+# ── Display state ─────────────────────────────────────────────────────────────
+
+class CryDisplay:
+    """Manages a ``rich.live.Live`` context showing model predictions."""
+
+    def __init__(self, max_history: int = 5) -> None:
+        self._window_count = 0
+        self._rms = 0.0
+        self._yamnet_status = ""
+        self._source_label = "mic"
+        self._predictions: list[CryPrediction] = []
+        self._history: deque[str] = deque(maxlen=max_history)
+        self._live: Live | None = None
+
+    # ── Public API ────────────────────────────────────────────────────────
+
+    def start(self) -> Live:
+        self._live = Live(self._build_table(), refresh_per_second=4)
+        self._live.start()
+        return self._live
+
+    def stop(self) -> None:
+        if self._live is not None:
+            self._live.stop()
+            self._live = None
+
+    def update(
+        self,
+        predictions: list[CryPrediction],
+        rms: float,
+        source_label: str = "mic",
+        is_silent: bool = False,
+    ) -> None:
+        self._window_count += 1
+        self._rms = rms
+        self._source_label = source_label
+        self._predictions = predictions
+
+        # Update YAMNet status line
+        yamnet_preds = [p for p in predictions if p.model_name == "YAMNet-detector"]
+        if yamnet_preds:
+            yp = yamnet_preds[0]
+            icon = "✅" if yp.label == "cry" else "❌"
+            self._yamnet_status = f"YAMNet: {icon} {yp.label.upper()} ({yp.confidence:.2f})"
+        else:
+            self._yamnet_status = "YAMNet: n/a"
+
+        # History
+        if is_silent:
+            self._history.appendleft(f"#{self._window_count}  [silence]")
+        else:
+            consensus = compute_consensus(predictions)
+            tag = consensus if consensus else "—"
+            self._history.appendleft(f"#{self._window_count}  {tag}")
+
+        if self._live is not None:
+            self._live.update(self._build_table())
+
+    # ── Table builder ─────────────────────────────────────────────────────
+
+    def _build_table(self) -> Table:
+        outer = Table(
+            title=f"🍼 TotTalk Cry Eval — listening ({self._source_label})  (1s windows, 16 kHz)",
+            title_style="bold cyan",
+            show_header=False,
+            show_edge=True,
+            pad_edge=True,
+            expand=True,
+        )
+        outer.add_column(ratio=1)
+
+        # Header row
+        header = (
+            f"  RMS: {self._rms:.4f}  |  Window #{self._window_count}  "
+            f"|  {self._yamnet_status}"
+        )
+        outer.add_row(Text(header, style="dim"))
+
+        # Predictions table
+        pred_table = Table(show_edge=False, expand=True, padding=(0, 1))
+        pred_table.add_column("Model", style="bold", min_width=18)
+        pred_table.add_column("Label", min_width=14)
+        pred_table.add_column("Confidence", min_width=12)
+        pred_table.add_column("Latency", justify="right", min_width=10)
+
+        for p in self._predictions:
+            if p.error:
+                pred_table.add_row(
+                    p.model_name,
+                    Text(f"⚠️ {p.error[:30]}", style="red"),
+                    "",
+                    "",
+                )
+            else:
+                pred_table.add_row(
+                    p.model_name,
+                    p.display_label,
+                    format_confidence(p.confidence),
+                    f"{p.latency_ms:.1f} ms",
+                )
+
+        # Consensus row
+        consensus = compute_consensus(self._predictions)
+        if consensus:
+            pred_table.add_row(
+                Text("CONSENSUS", style="bold magenta"),
+                Text(consensus, style="bold"),
+                "",
+                "",
+            )
+
+        outer.add_row(pred_table)
+
+        # History
+        if self._history:
+            hist_str = "  ".join(self._history)
+            outer.add_row(Text(f"  Last detections: {hist_str}", style="dim"))
+
+        # Cry meaning legend — show meaning for the consensus / top prediction
+        shown_label = self._current_reason_label()
+        if shown_label and shown_label in LABEL_MEANING:
+            emoji = LABEL_EMOJI.get(shown_label, "")
+            outer.add_row(
+                Text(
+                    f"  {emoji} {shown_label.replace('_', ' ').title()}: "
+                    f"{LABEL_MEANING[shown_label]}",
+                    style="italic yellow",
+                )
+            )
+
+        return outer
+
+    def _current_reason_label(self) -> str | None:
+        """Return the most relevant reason label from the current predictions."""
+        for p in self._predictions:
+            if p.model_name == "YAMNet-detector":
+                continue
+            if p.error or p.label in ("no_cry", "timeout", "error"):
+                continue
+            return p.label
+        return None

main.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""TotTalk Cry Eval — real-time multi-model baby cry classifier."""
+
+from __future__ import annotations
+
+import json
+import queue
+from pathlib import Path
+
+import click
+import numpy as np
+import sounddevice as sd
+from rich.console import Console
+from rich.progress import Progress, SpinnerColumn, TextColumn
+
+from audio.capture import FileCapture, MicCapture
+from audio.preprocess import SAMPLE_RATE, compute_rms, is_silent, normalize_audio
+from display.table import CryDisplay
+from models.ensemble import EnsembleClassifier
+
+console = Console(stderr=True)
+
+
+def _print_audio_devices() -> None:
+    """Print available audio devices for reference."""
+    console.print("\n[bold]Audio devices:[/bold]")
+    try:
+        devices = sd.query_devices()
+        default_in = sd.default.device[0]
+        for i, d in enumerate(devices):
+            marker = " ← default input" if i == default_in else ""
+            if d["max_input_channels"] > 0:
+                console.print(f"  [{i}] {d['name']}  (in:{d['max_input_channels']}){marker}")
+    except Exception as exc:
+        console.print(f"  [red]Could not query devices: {exc}[/red]")
+
+
+def _load_models(ensemble: EnsembleClassifier) -> None:
+    """Load all models with a rich progress spinner."""
+    console.print()
+    with Progress(
+        SpinnerColumn(),
+        TextColumn("[progress.description]{task.description}"),
+        console=console,
+    ) as progress:
+        task = progress.add_task("Loading models…", total=None)
+        results = ensemble.load_all()
+        progress.update(task, description="Models loaded.")
+
+    for name, error in results.items():
+        if error:
+            console.print(f"  [red]✗ {name}: {error}[/red]")
+        else:
+            console.print(f"  [green]✓ {name}[/green]")
+    console.print()
+
+
+@click.command()
+@click.option(
+    "--file", "audio_file", default=None, type=click.Path(exists=True),
+    help="Path to a WAV/FLAC/MP3 file (loops in sliding windows instead of mic).",
+)
+@click.option(
+    "--models", "model_names", default=None,
+    help="Comma-separated subset of models to run: svc,hubert,kibalama,yamnet",
+)
+@click.option(
+    "--no-yamnet-gate", is_flag=True, default=False,
+    help="Disable YAMNet gating (always run reason classifiers).",
+)
+@click.option(
+    "--save-log", default=None, type=click.Path(),
+    help="Append JSONL predictions to this file.",
+)
+@click.option(
+    "--sensitivity", default=None, type=float,
+    help="Silence RMS threshold override (default 0.001). Lower = more sensitive.",
+)
+def cli(
+    audio_file: str | None,
+    model_names: str | None,
+    no_yamnet_gate: bool,
+    save_log: str | None,
+    sensitivity: float | None,
+) -> None:
+    """🍼 TotTalk Cry Eval — real-time multi-model baby cry classifier."""
+    console.print("[bold cyan]🍼 TotTalk Cry Eval[/bold cyan]")
+
+    # Override silence threshold if requested
+    if sensitivity is not None:
+        import audio.preprocess as _ap
+        _ap.SILENCE_RMS_THRESHOLD = sensitivity
+        console.print(f"[dim]Silence threshold set to {sensitivity}[/dim]")
+
+    # Parse model list
+    selected = model_names.split(",") if model_names else None
+
+    # Init ensemble
+    ensemble = EnsembleClassifier(
+        model_names=selected,
+        use_yamnet_gate=not no_yamnet_gate,
+    )
+
+    # Print device info
+    if audio_file is None:
+        _print_audio_devices()
+
+    # Load models
+    _load_models(ensemble)
+
+    # Log file handle
+    log_fh = None
+    if save_log:
+        log_fh = open(save_log, "a")  # noqa: SIM115
+
+    # Set up audio source
+    if audio_file:
+        source_label = f"file: {Path(audio_file).name}"
+        capture = FileCapture(audio_file)
+    else:
+        source_label = "mic"
+        capture = MicCapture()
+
+    # Display
+    display = CryDisplay()
+
+    try:
+        capture.start()
+        display.start()
+        console.print(f"[dim]Listening ({source_label})… Press Ctrl+C to stop.[/dim]\n")
+
+        while True:
+            try:
+                window: np.ndarray = capture.window_queue.get(timeout=3.0)
+            except queue.Empty:
+                continue
+
+            rms = compute_rms(window)
+            silent = is_silent(window)
+
+            if silent:
+                display.update([], rms, source_label=source_label, is_silent=True)
+                continue
+
+            # Peak-normalize so quiet phone playback reaches model-friendly levels
+            window = normalize_audio(window)
+
+            predictions = ensemble.predict_all(window, SAMPLE_RATE)
+            display.update(predictions, rms, source_label=source_label)
+
+            # Optional JSONL log
+            if log_fh is not None:
+                record = {
+                    "window": display._window_count,
+                    "rms": rms,
+                    "predictions": [
+                        {
+                            "model": p.model_name,
+                            "label": p.label,
+                            "confidence": p.confidence,
+                            "latency_ms": p.latency_ms,
+                            "error": p.error,
+                        }
+                        for p in predictions
+                    ],
+                }
+                log_fh.write(json.dumps(record) + "\n")
+                log_fh.flush()
+
+    except KeyboardInterrupt:
+        console.print("\n[yellow]Stopped.[/yellow]")
+    finally:
+        capture.stop()
+        display.stop()
+        if log_fh is not None:
+            log_fh.close()
+
+
+if __name__ == "__main__":
+    cli()

models/base.py

@@ -0,0 +1,78 @@
+"""Abstract base class and shared types for cry classifiers."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+import numpy as np
+
+# ── Label sets ────────────────────────────────────────────────────────────────
+LABELS_5CLASS = ["belly_pain", "burping", "discomfort", "hungry", "tired"]
+LABELS_8CLASS = [
+    "hungry", "burping", "scared", "belly_pain",
+    "discomfort", "cold_hot", "lonely", "tired",
+]
+
+LABEL_EMOJI: dict[str, str] = {
+    "belly_pain": "😣",
+    "burping": "🫧",
+    "discomfort": "😖",
+    "hungry": "🍼",
+    "tired": "😴",
+    "scared": "😨",
+    "cold_hot": "🌡️",
+    "lonely": "🥺",
+    "cry": "✅",
+    "not_cry": "❌",
+}
+
+# What each cry label means — shown in the terminal legend
+LABEL_MEANING: dict[str, str] = {
+    "belly_pain": "Baby has stomach cramps or gas — try gentle tummy massage or bicycle legs",
+    "burping":    "Baby needs to burp — hold upright and pat back gently",
+    "discomfort": "General discomfort — check diaper, clothing, temperature, or position",
+    "hungry":     "Baby is hungry — time to feed",
+    "tired":      "Baby is sleepy or overtired — needs soothing and rest",
+    "scared":     "Baby is startled or frightened — comfort and hold close",
+    "cold_hot":   "Baby is too cold or too warm — adjust clothing or room temperature",
+    "lonely":     "Baby wants attention or closeness — pick up and cuddle",
+}
+
+
+def display_label(raw: str) -> str:
+    """Return an emoji-prefixed human-friendly label."""
+    emoji = LABEL_EMOJI.get(raw, "❓")
+    name = raw.replace("_", " ").title()
+    return f"{emoji} {name}"
+
+
+# ── Prediction dataclass ─────────────────────────────────────────────────────
+@dataclass
+class CryPrediction:
+    model_name: str
+    label: str               # raw label
+    display_label: str        # emoji + human name
+    confidence: float         # 0.0 – 1.0
+    latency_ms: float         # inference time in ms
+    error: str | None = None
+
+
+# ── Abstract classifier ──────────────────────────────────────────────────────
+class CryClassifier(ABC):
+    name: str = "unnamed"
+    description: str = ""
+
+    def __init__(self) -> None:
+        self._loaded = False
+
+    @abstractmethod
+    def load(self) -> None:
+        """Download weights (if needed) and initialize the model."""
+
+    @abstractmethod
+    def predict(self, audio_np: np.ndarray, sr: int) -> CryPrediction:
+        """Run inference on a single audio window and return a prediction."""
+
+    def is_loaded(self) -> bool:
+        return self._loaded

models/ensemble.py

@@ -0,0 +1,205 @@
+"""Ensemble runner — loads all models and orchestrates per-window inference."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import threading
+from typing import Sequence
+
+import numpy as np
+from rich.console import Console
+
+from models.base import CryClassifier, CryPrediction, display_label
+from models.foduucom_svc import FoduucomSVC
+from models.kibalama import KibalamaCry
+from models.wiam_wav2vec2 import DistilHuBERTCry
+from models.yamnet import YAMNetDetector
+
+console = Console(stderr=True)
+
+# Map of short names → classes for CLI --models filtering
+MODEL_REGISTRY: dict[str, type[CryClassifier]] = {
+    "svc": FoduucomSVC,
+    "hubert": DistilHuBERTCry,
+    "kibalama": KibalamaCry,
+    "yamnet": YAMNetDetector,
+}
+
+
+class EnsembleClassifier:
+    """Loads and runs multiple cry classifiers, aggregating results."""
+
+    def __init__(
+        self,
+        model_names: Sequence[str] | None = None,
+        use_yamnet_gate: bool = True,
+    ) -> None:
+        self.use_yamnet_gate = use_yamnet_gate
+
+        # Decide which models to instantiate
+        if model_names is None:
+            names = list(MODEL_REGISTRY.keys())
+        else:
+            names = [n.lower() for n in model_names]
+
+        # Always include YAMNet if gating is enabled and it's not already in the list
+        if use_yamnet_gate and "yamnet" not in names:
+            names.insert(0, "yamnet")
+
+        self._classifiers: list[CryClassifier] = []
+        for n in names:
+            cls = MODEL_REGISTRY.get(n)
+            if cls is None:
+                console.print(f"[yellow]⚠ Unknown model '{n}' — skipping[/yellow]")
+                continue
+            self._classifiers.append(cls())
+
+        self._yamnet: YAMNetDetector | None = None
+        self._reason_classifiers: list[CryClassifier] = []
+        for c in self._classifiers:
+            if isinstance(c, YAMNetDetector):
+                self._yamnet = c
+            else:
+                self._reason_classifiers.append(c)
+
+    # ── Loading ───────────────────────────────────────────────────────────
+    def load_all(self) -> dict[str, str | None]:
+        """Load every model in parallel. Return {name: error_or_None}."""
+        results: dict[str, str | None] = {}
+        lock = threading.Lock()
+
+        def _load(clf: CryClassifier) -> None:
+            try:
+                clf.load()
+                with lock:
+                    results[clf.name] = None
+            except Exception as exc:
+                with lock:
+                    results[clf.name] = str(exc)
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=len(self._classifiers)) as pool:
+            pool.map(_load, self._classifiers)
+
+        return results
+
+    # ── Inference ─────────────────────────────────────────────────────────
+    def predict_all(
+        self,
+        audio_np: np.ndarray,
+        sr: int,
+    ) -> list[CryPrediction]:
+        predictions: list[CryPrediction] = []
+
+        # 1. YAMNet gate
+        if self._yamnet is not None and self._yamnet.is_loaded():
+            yamnet_pred = self._yamnet.predict(audio_np, sr)
+            predictions.append(yamnet_pred)
+
+            if (
+                self.use_yamnet_gate
+                and yamnet_pred.label == "not_cry"
+                and yamnet_pred.confidence < 0.4  # not_cry with cry-score < 0.4
+            ):
+                # Skip reason classifiers — no cry detected
+                for rc in self._reason_classifiers:
+                    predictions.append(
+                        CryPrediction(
+                            model_name=rc.name,
+                            label="no_cry",
+                            display_label="— No cry",
+                            confidence=0.0,
+                            latency_ms=0.0,
+                        )
+                    )
+                return predictions
+        elif self._yamnet is not None:
+            predictions.append(
+                CryPrediction(
+                    model_name=self._yamnet.name,
+                    label="error",
+                    display_label="⚠️ Load Error",
+                    confidence=0.0,
+                    latency_ms=0.0,
+                    error="Model not loaded",
+                )
+            )
+
+        # 2. Run reason classifiers
+        # SVC is sub-ms — run synchronously
+        # Transformer models (HuBERT, Kibalama) — run in threads with timeout
+        inline_results: list[CryPrediction] = []
+        thread_futures: list[tuple[CryClassifier, concurrent.futures.Future[CryPrediction]]] = []
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+            for clf in self._reason_classifiers:
+                if not clf.is_loaded():
+                    predictions.append(
+                        CryPrediction(
+                            model_name=clf.name,
+                            label="error",
+                            display_label="⚠️ Load Error",
+                            confidence=0.0,
+                            latency_ms=0.0,
+                            error="Model not loaded",
+                        )
+                    )
+                    continue
+
+                if isinstance(clf, FoduucomSVC):
+                    # Fast — run inline
+                    inline_results.append(clf.predict(audio_np, sr))
+                else:
+                    # Slow — run in a thread
+                    fut = pool.submit(clf.predict, audio_np, sr)
+                    thread_futures.append((clf, fut))
+
+            predictions.extend(inline_results)
+
+            for clf, fut in thread_futures:
+                try:
+                    result = fut.result(timeout=2.0)
+                    predictions.append(result)
+                except concurrent.futures.TimeoutError:
+                    predictions.append(
+                        CryPrediction(
+                            model_name=clf.name,
+                            label="timeout",
+                            display_label="⏳ Timeout",
+                            confidence=0.0,
+                            latency_ms=2000.0,
+                            error="Inference timed out (>2 s)",
+                        )
+                    )
+
+        return predictions
+
+    @property
+    def classifiers(self) -> list[CryClassifier]:
+        return list(self._classifiers)
+
+
+def compute_consensus(predictions: list[CryPrediction]) -> str | None:
+    """Weighted-vote consensus across *reason* classifiers (exclude YAMNet).
+
+    Each model contributes its confidence as a weight.
+    Returns the winning label string or None if no agreement / no valid votes.
+    """
+    weighted_votes: dict[str, float] = {}
+    vote_count: dict[str, int] = {}
+    total_voters = 0
+
+    for p in predictions:
+        if p.model_name == "YAMNet-detector":
+            continue
+        if p.error or p.label in ("no_cry", "timeout", "error"):
+            continue
+        total_voters += 1
+        weighted_votes[p.label] = weighted_votes.get(p.label, 0.0) + p.confidence
+        vote_count[p.label] = vote_count.get(p.label, 0) + 1
+
+    if not weighted_votes:
+        return None
+
+    top_label = max(weighted_votes, key=weighted_votes.__getitem__)
+    count = vote_count[top_label]
+    return f"{display_label(top_label)}  ({count}/{total_voters} agree)"

models/foduucom_svc.py

@@ -0,0 +1,94 @@
+"""Model 1 — foduucom/baby-cry-classification sklearn SVC."""
+
+from __future__ import annotations
+
+import os
+import time
+
+import numpy as np
+
+from models.base import CryClassifier, CryPrediction, display_label
+
+
+class FoduucomSVC(CryClassifier):
+    name = "foduucom-SVC"
+    description = "sklearn SVC trained on 5-class baby cry features (HuggingFace: foduucom/baby-cry-classification)"
+
+    REPO_ID = "foduucom/baby-cry-classification"
+    WEIGHTS_DIR = os.path.join("weights", "foduucom")
+
+    # Fixed feature dimensions: 40 MFCC + 12 chroma + 7 contrast + 6 tonnetz = 65
+    _FIXED_FEATURES = 65
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._model = None
+        self._label_encoder = None
+        self._n_mels: int = 128  # will be auto-detected from model
+
+    def load(self) -> None:
+        from huggingface_hub import hf_hub_download
+        import joblib
+
+        os.makedirs(self.WEIGHTS_DIR, exist_ok=True)
+
+        model_path = hf_hub_download(
+            repo_id=self.REPO_ID,
+            filename="model.joblib",
+            local_dir=self.WEIGHTS_DIR,
+        )
+        label_path = hf_hub_download(
+            repo_id=self.REPO_ID,
+            filename="label.joblib",
+            local_dir=self.WEIGHTS_DIR,
+        )
+
+        self._model = joblib.load(model_path)
+        self._label_encoder = joblib.load(label_path)
+
+        # Auto-detect n_mels from the model's expected feature count
+        expected = getattr(self._model, "n_features_in_", None)
+        if expected is not None:
+            self._n_mels = expected - self._FIXED_FEATURES
+            if self._n_mels < 1:
+                self._n_mels = 128  # fallback
+
+        self._loaded = True
+
+    def predict(self, audio_np: np.ndarray, sr: int) -> CryPrediction:
+        from audio.preprocess import extract_mfcc_features
+
+        t0 = time.perf_counter()
+        try:
+            features = extract_mfcc_features(audio_np, sr, n_mels=self._n_mels).reshape(1, -1)
+            pred_encoded = self._model.predict(features)
+            label_raw: str = self._label_encoder.inverse_transform(pred_encoded)[0]
+
+            # SVC doesn't natively support predict_proba.  Use decision_function
+            # distance as a rough proxy, clamped to [0, 1].
+            try:
+                decision = self._model.decision_function(features)
+                # For multi-class, decision is a matrix — take max distance
+                max_dist = float(np.max(np.abs(decision)))
+                confidence = min(max_dist / 2.0, 1.0)  # rough normalisation
+            except Exception:
+                confidence = 0.85  # fallback constant
+
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label=label_raw,
+                display_label=display_label(label_raw),
+                confidence=confidence,
+                latency_ms=latency,
+            )
+        except Exception as exc:
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label="error",
+                display_label="⚠️ Error",
+                confidence=0.0,
+                latency_ms=latency,
+                error=str(exc),
+            )

models/kibalama.py

@@ -0,0 +1,93 @@
+"""Model — Kibalama/baby_cry_classification_model (Wav2Vec2, 9-class)."""
+
+from __future__ import annotations
+
+import time
+
+import numpy as np
+
+from models.base import CryClassifier, CryPrediction, display_label
+
+
+# Map Kibalama's raw labels to our canonical label set
+_LABEL_MAP: dict[str, str | None] = {
+    "belly pain": "belly_pain",
+    "burping": "burping",
+    "cold_hot": "cold_hot",
+    "discomfort": "discomfort",
+    "hungry": "hungry",
+    "tired": "tired",
+    # Non-cry labels — skip (return as-is but with low relevance)
+    "laugh": None,
+    "noise": None,
+    "silence": None,
+}
+
+
+class KibalamaCry(CryClassifier):
+    name = "Kibalama-9c"
+    description = (
+        "Wav2Vec2 fine-tuned on 9-class baby cry dataset "
+        "(HuggingFace: Kibalama/baby_cry_classification_model)"
+    )
+
+    MODEL_ID = "Kibalama/baby_cry_classification_model"
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._pipe = None
+
+    def load(self) -> None:
+        from transformers import pipeline
+
+        self._pipe = pipeline(
+            "audio-classification",
+            model=self.MODEL_ID,
+            device="cpu",
+        )
+        self._loaded = True
+
+    def predict(self, audio_np: np.ndarray, sr: int) -> CryPrediction:
+        from audio.preprocess import SAMPLE_RATE, resample
+
+        t0 = time.perf_counter()
+        try:
+            if sr != SAMPLE_RATE:
+                audio_np = resample(audio_np, sr, SAMPLE_RATE)
+
+            results = self._pipe(
+                {"raw": audio_np, "sampling_rate": SAMPLE_RATE},
+                top_k=9,
+            )
+
+            # Pick the top *cry-related* label (skip laugh/noise/silence)
+            for res in results:
+                raw = res["label"]
+                mapped = _LABEL_MAP.get(raw, raw)
+                if mapped is not None:
+                    label_raw = mapped
+                    confidence = res["score"]
+                    break
+            else:
+                # All top results were non-cry categories
+                label_raw = "no_cry"
+                confidence = 0.0
+
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label=label_raw,
+                display_label=display_label(label_raw),
+                confidence=confidence,
+                latency_ms=latency,
+            )
+        except Exception as exc:
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label="error",
+                display_label="⚠️ Error",
+                confidence=0.0,
+                latency_ms=latency,
+                error=str(exc),
+            )

models/wiam_wav2vec2.py

@@ -0,0 +1,68 @@
+"""Model 2 — AmeerHesham/distilhubert-finetuned-baby_cry (DistilHuBERT)."""
+
+from __future__ import annotations
+
+import time
+
+import numpy as np
+
+from models.base import CryClassifier, CryPrediction, display_label
+
+
+class DistilHuBERTCry(CryClassifier):
+    name = "DistilHuBERT"
+    description = (
+        "DistilHuBERT fine-tuned for baby cry classification "
+        "(HuggingFace: AmeerHesham/distilhubert-finetuned-baby_cry)"
+    )
+
+    MODEL_ID = "AmeerHesham/distilhubert-finetuned-baby_cry"
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._pipe = None
+
+    def load(self) -> None:
+        from transformers import pipeline
+
+        self._pipe = pipeline(
+            "audio-classification",
+            model=self.MODEL_ID,
+            device="cpu",
+        )
+        self._loaded = True
+
+    def predict(self, audio_np: np.ndarray, sr: int) -> CryPrediction:
+        from audio.preprocess import SAMPLE_RATE, resample
+
+        t0 = time.perf_counter()
+        try:
+            if sr != SAMPLE_RATE:
+                audio_np = resample(audio_np, sr, SAMPLE_RATE)
+
+            results = self._pipe(
+                {"raw": audio_np, "sampling_rate": SAMPLE_RATE},
+                top_k=1,
+            )
+            top = results[0]
+            label_raw: str = top["label"]
+            confidence: float = top["score"]
+
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label=label_raw,
+                display_label=display_label(label_raw),
+                confidence=confidence,
+                latency_ms=latency,
+            )
+        except Exception as exc:
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label="error",
+                display_label="⚠️ Error",
+                confidence=0.0,
+                latency_ms=latency,
+                error=str(exc),
+            )

models/yamnet.py

@@ -0,0 +1,67 @@
+"""Model 3 — YAMNet binary baby-cry detector via TensorFlow Hub."""
+
+from __future__ import annotations
+
+import time
+
+import numpy as np
+
+from models.base import CryClassifier, CryPrediction, display_label
+
+
+# "Baby cry, infant cry" class index in the AudioSet ontology used by YAMNet.
+_BABY_CRY_CLASS_INDEX = 20
+
+
+class YAMNetDetector(CryClassifier):
+    name = "YAMNet-detector"
+    description = "YAMNet (TF Hub) binary cry detector — gates the reason classifiers"
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._model = None
+
+    def load(self) -> None:
+        import tensorflow_hub as hub
+
+        self._model = hub.load("https://tfhub.dev/google/yamnet/1")
+        self._loaded = True
+
+    def predict(self, audio_np: np.ndarray, sr: int) -> CryPrediction:
+        import tensorflow as tf
+        from audio.preprocess import SAMPLE_RATE, resample
+
+        t0 = time.perf_counter()
+        try:
+            if sr != SAMPLE_RATE:
+                audio_np = resample(audio_np, sr, SAMPLE_RATE)
+
+            waveform = tf.cast(audio_np, tf.float32)
+            scores, embeddings, spectrogram = self._model(waveform)
+
+            # scores shape: (num_frames, 521)
+            scores_np = scores.numpy()
+            cry_scores = scores_np[:, _BABY_CRY_CLASS_INDEX]
+            avg_cry_score = float(np.mean(cry_scores))
+
+            is_cry = avg_cry_score >= 0.4
+            label_raw = "cry" if is_cry else "not_cry"
+
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label=label_raw,
+                display_label=display_label(label_raw),
+                confidence=avg_cry_score,
+                latency_ms=latency,
+            )
+        except Exception as exc:
+            latency = (time.perf_counter() - t0) * 1000
+            return CryPrediction(
+                model_name=self.name,
+                label="error",
+                display_label="⚠️ Error",
+                confidence=0.0,
+                latency_ms=latency,
+                error=str(exc),
+            )

pyproject.toml

@@ -0,0 +1,24 @@
+[project]
+name = "cry-eval"
+version = "0.1.0"
+description = "Real-time multi-model baby cry classification CLI tool"
+requires-python = ">=3.11"
+dependencies = [
+    "sounddevice>=0.4.6",
+    "numpy>=1.24.0",
+    "librosa>=0.10.0",
+    "scikit-learn>=1.3.0",
+    "joblib>=1.3.0",
+    "torch>=2.1.0",
+    "torchaudio>=2.1.0",
+    "transformers>=4.38.0",
+    "tensorflow>=2.15.0",
+    "tensorflow-hub>=0.15.0",
+    "huggingface-hub>=0.20.0",
+    "rich>=13.7.0",
+    "click>=8.1.0",
+    "soundfile>=0.12.0",
+]
+
+[project.scripts]
+cry-eval = "main:cli"

weights/.gitignore

@@ -0,0 +1,3 @@
+# auto-downloaded model weights
+*
+!.gitignore