diff --git a/app/pipeline/processor.py b/app/pipeline/processor.py new file mode 100644 index 0000000..d56a461 --- /dev/null +++ b/app/pipeline/processor.py @@ -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 diff --git a/tests/test_processor.py b/tests/test_processor.py new file mode 100644 index 0000000..3af8343 --- /dev/null +++ b/tests/test_processor.py @@ -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