feat: processing pipeline — Whisper + resemblyzer + speaker match + SQLite write

Implements process_utterance() with concurrent STT/embedding via asyncio.gather, speaker matching with configurable thresholds, utterance DB write, and WAV clip save. httpx imported lazily to keep the dev environment functional without full install. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-28 10:37:39 -05:00
parent b2c82f36d7
commit 8312db1b49
2 changed files with 200 additions and 0 deletions
@@ -0,0 +1,111 @@
+import asyncio
+import sqlite3
+import wave
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+
+from app.shared.config import Settings
+from app.pipeline.recognition import embed_audio, find_best_speaker_match, pack_embedding
+
+
+async def transcribe_audio(wav_bytes: bytes, speaches_url: str) -> str:
+    """POST audio to Speaches/Whisper, return transcript text."""
+    import httpx  # lazy import — not available in dev env without full install
+    async with httpx.AsyncClient(timeout=30.0) as client:
+        resp = await client.post(
+            f"{speaches_url}/v1/audio/transcriptions",
+            files={"file": ("audio.wav", wav_bytes, "audio/wav")},
+            data={"model": "Systran/faster-whisper-large-v3-turbo"},
+        )
+        resp.raise_for_status()
+        return resp.json()["text"].strip()
+
+
+def save_clip(audio_bytes: bytes, clips_dir: str, utterance_id: int, start_time: datetime) -> str:
+    """Save raw PCM bytes as a WAV file, return the path."""
+    date_str = start_time.strftime("%Y-%m-%d")
+    clip_dir = Path(clips_dir) / date_str
+    clip_dir.mkdir(parents=True, exist_ok=True)
+    path = clip_dir / f"{utterance_id}.wav"
+    with wave.open(str(path), "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)  # int16
+        wf.setframerate(16000)
+        wf.writeframes(audio_bytes)
+    return str(path)
+
+
+async def process_utterance(
+    room_id: int,
+    audio_bytes: bytes,
+    start_time: datetime,
+    end_time: datetime,
+    settings: Settings,
+    db: sqlite3.Connection,
+) -> Optional[int]:
+    """
+    Full pipeline for one speech segment:
+      1. Whisper STT + resemblyzer embed (concurrent)
+      2. Speaker matching
+      3. SQLite write + WAV clip save
+    Returns the new utterance row id, or None on failure.
+    """
+    try:
+        transcript, embedding = await asyncio.gather(
+            transcribe_audio(audio_bytes, settings.speaches_url),
+            asyncio.to_thread(embed_audio, audio_bytes),
+        )
+    except Exception:
+        return None
+
+    # Load all stored embeddings for matching
+    rows = db.execute("SELECT speaker_id, embedding FROM voice_embeddings").fetchall()
+    stored = [(r["speaker_id"], r["embedding"]) for r in rows]
+
+    match_status, speaker_id, confidence = find_best_speaker_match(
+        embedding,
+        stored,
+        known_threshold=settings.speaker_known_threshold,
+        ambiguous_threshold=settings.speaker_ambiguous_threshold,
+    )
+
+    # Insert utterance row first to get the id
+    cur = db.execute(
+        """INSERT INTO utterances
+           (room_id, speaker_id, transcript, embedding, match_status, match_confidence,
+            start_time, end_time)
+           VALUES (?,?,?,?,?,?,?,?)""",
+        (
+            room_id,
+            speaker_id,
+            transcript,
+            pack_embedding(embedding),
+            match_status,
+            confidence,
+            start_time.isoformat(),
+            end_time.isoformat(),
+        ),
+    )
+    db.commit()
+    utterance_id = cur.lastrowid
+
+    # Save WAV clip and update row with path
+    clip_path = save_clip(audio_bytes, settings.clips_dir, utterance_id, start_time)
+    db.execute(
+        "UPDATE utterances SET audio_clip_path=? WHERE id=?",
+        (clip_path, utterance_id),
+    )
+    db.commit()
+
+    # If known match, store embedding as an additional reference for this speaker
+    if match_status == "known" and speaker_id is not None:
+        db.execute(
+            "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?,?)",
+            (speaker_id, pack_embedding(embedding)),
+        )
+        db.commit()
+
+    return utterance_id
@@ -0,0 +1,89 @@
+import pytest
+import numpy as np
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+# Pre-import so mocker.patch can resolve the module before the test's local import
+import app.pipeline.processor  # noqa: F401
+
+
+@pytest.mark.asyncio
+async def test_process_utterance_known_speaker(mocker, settings, db):
+    # Setup: insert a room and speaker with embedding
+    db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",))
+    db.execute("INSERT INTO speakers (name) VALUES (?)", ("Jeremy",))
+    db.commit()
+    room = db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone()
+    speaker = db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone()
+
+    from app.pipeline.recognition import pack_embedding
+    ref_emb = np.array([1.0] + [0.0] * 255, dtype=np.float32)
+    db.execute(
+        "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?,?)",
+        (speaker["id"], pack_embedding(ref_emb)),
+    )
+    db.commit()
+
+    # Mock Whisper HTTP call
+    mocker.patch(
+        "app.pipeline.processor.transcribe_audio",
+        new=AsyncMock(return_value="hello world"),
+    )
+    # Mock resemblyzer — return a very similar embedding
+    query_emb = np.array([0.99] + [0.0] * 255, dtype=np.float32)
+    mocker.patch("app.pipeline.processor.embed_audio", return_value=query_emb)
+    # Mock clip saving
+    mocker.patch("app.pipeline.processor.save_clip", return_value="/data/clips/test.wav")
+
+    from app.pipeline.processor import process_utterance
+    start = datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+    end = datetime(2026, 1, 1, 10, 0, 3, tzinfo=timezone.utc)
+
+    utterance_id = await process_utterance(
+        room_id=room["id"],
+        audio_bytes=b"\x00" * 100,
+        start_time=start,
+        end_time=end,
+        settings=settings,
+        db=db,
+    )
+
+    assert utterance_id is not None
+    row = db.execute("SELECT * FROM utterances WHERE id=?", (utterance_id,)).fetchone()
+    assert row["transcript"] == "hello world"
+    assert row["match_status"] == "known"
+    assert row["speaker_id"] == speaker["id"]
+
+
+@pytest.mark.asyncio
+async def test_process_utterance_unknown_speaker(mocker, settings, db):
+    db.execute("INSERT INTO rooms (name) VALUES (?)", ("Office",))
+    db.commit()
+    room = db.execute("SELECT id FROM rooms WHERE name='Office'").fetchone()
+
+    mocker.patch(
+        "app.pipeline.processor.transcribe_audio",
+        new=AsyncMock(return_value="test transcript"),
+    )
+    mocker.patch(
+        "app.pipeline.processor.embed_audio",
+        return_value=np.zeros(256, dtype=np.float32),
+    )
+    mocker.patch("app.pipeline.processor.save_clip", return_value="/data/clips/test.wav")
+
+    from app.pipeline.processor import process_utterance
+    start = datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+    end = datetime(2026, 1, 1, 10, 0, 2, tzinfo=timezone.utc)
+
+    utterance_id = await process_utterance(
+        room_id=room["id"],
+        audio_bytes=b"\x00" * 100,
+        start_time=start,
+        end_time=end,
+        settings=settings,
+        db=db,
+    )
+
+    row = db.execute("SELECT * FROM utterances WHERE id=?", (utterance_id,)).fetchone()
+    assert row["match_status"] == "unknown"
+    assert row["speaker_id"] is None