"""Async vocal analysis — runs background, never blocks TTS path.""" import asyncio import io import logging import math import struct import wave from dataclasses import dataclass from typing import Callable, Optional import httpx logger = logging.getLogger(__name__) SPEECHBRAIN_URL = "http://speechbrain-service:8001" @dataclass class VocalState: emotion: str = "neutral" confidence: float = 0.0 energy: float = 0.0 pitch_hz: Optional[float] = None rate: float = 0.0 def to_tag(self) -> str: parts = [f"emotion={self.emotion}", f"energy={self.energy:.2f}"] if self.pitch_hz: parts.append(f"pitch={self.pitch_hz:.0f}hz") if self.rate: parts.append(f"rate={'fast' if self.rate > 3 else 'slow' if self.rate < 1.5 else 'normal'}") return f"[VOCAL {' '.join(parts)}]" def compute_energy(audio_bytes: bytes) -> float: if len(audio_bytes) < 2: return 0.0 samples = struct.unpack_from("<" + "h" * (len(audio_bytes) // 2), audio_bytes) rms = math.sqrt(sum(x * x for x in samples) / len(samples)) / 32768.0 return round(min(rms * 10, 1.0), 3) async def analyze_audio(audio_bytes: bytes) -> VocalState: """Call SpeechBrain for emotion/pitch. Falls back to energy-only if unavailable.""" energy = compute_energy(audio_bytes) state = VocalState(energy=energy) try: buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio_bytes) wav_bytes = buf.getvalue() async with httpx.AsyncClient(timeout=3.0) as client: r = await client.post( f"{SPEECHBRAIN_URL}/analyze", files={"audio": ("u.wav", wav_bytes, "audio/wav")}, ) r.raise_for_status() d = r.json() state.emotion = d.get("emotion", "neutral") state.confidence = d.get("emotion_confidence", 0.0) state.pitch_hz = d.get("pitch_mean_hz") state.rate = d.get("speaking_rate_approx", 0.0) except Exception as e: logger.debug("SpeechBrain unavailable, using energy-only: %s", e) return state class VocalAnalyzer: """Holds the last completed vocal state; updates async after each utterance.""" def __init__(self, on_update: Optional[Callable[[VocalState], None]] = None): self._state = VocalState() self._on_update = on_update self._audio_buffer: list[bytes] = [] self._collecting = False @property def current(self) -> VocalState: return self._state def start_collecting(self): self._collecting = True self._audio_buffer = [] def feed(self, audio_chunk: bytes): if self._collecting: self._audio_buffer.append(audio_chunk) def stop_collecting_and_analyze(self) -> asyncio.Task: self._collecting = False audio = b"".join(self._audio_buffer) self._audio_buffer = [] return asyncio.create_task(self._analyze_bg(audio)) async def _analyze_bg(self, audio_bytes: bytes): if not audio_bytes: return try: state = await analyze_audio(audio_bytes) self._state = state if self._on_update: self._on_update(state) except Exception as e: logger.warning("Vocal analysis failed: %s", e)