pluto
8312db1b49
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>
112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
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
|