Deep-dive · Voice Pipeline cross-device¶

# requirements: pvporcupine==3.0.2, pvrecorder==1.2.3
import pvporcupine
import pvrecorder
import struct
from collections import deque
from pathlib import Path

class WakeWordDetector:
    """Detector Porcupine con buffer circolare pre-roll (250ms)."""

    SENSITIVITY = 0.6
    PRE_ROLL_FRAMES = 8

    def __init__(self, access_key: str, keyword_path: Path | None = None):
        self._porcupine = pvporcupine.create(
            access_key=access_key,
            keywords=["jarvis"] if keyword_path is None else [],
            keyword_paths=None if keyword_path is None else [str(keyword_path)],
            sensitivities=[self.SENSITIVITY],
        )
        self._recorder = pvrecorder.PvRecorder(
            frame_length=self._porcupine.frame_length, device_index=-1
        )
        self._pre_roll: deque[bytes] = deque(maxlen=self.PRE_ROLL_FRAMES)

    def listen_once(self) -> list[bytes]:
        """Blocca fino al wake-word. Restituisce i frame pre-roll."""
        self._recorder.start()
        try:
            while True:
                pcm_frame = self._recorder.read()
                self._pre_roll.append(struct.pack("h" * len(pcm_frame), *pcm_frame))
                if self._porcupine.process(pcm_frame) >= 0:
                    return list(self._pre_roll)
        finally:
            self._recorder.stop()

# requirements: faster-whisper==1.1.0, silero-vad==5.1.2, websockets==12.0
import asyncio
import io
import websockets
import numpy as np
import soundfile as sf
import torch
from collections import deque
from faster_whisper import WhisperModel

_vad_model, _vad_utils = torch.hub.load(
    repo_or_dir="snakers4/silero-vad", model="silero_vad", onnx=False
)
(_get_speech_timestamps, *_) = _vad_utils

SAMPLE_RATE = 16_000
CHUNK_SAMPLES = int(SAMPLE_RATE * 0.5)
SILENCE_THRESHOLD_S = 1.2

_whisper_model = WhisperModel(
    "distil-large-v3", device="cuda", compute_type="int8_float16"
)


async def stream_transcribe(websocket) -> None:
    """Riceve chunk PCM float32 16kHz, ritorna trascrizioni parziali real-time."""
    audio_buffer: deque[np.ndarray] = deque()
    silence_frames = 0
    max_silence = int(SILENCE_THRESHOLD_S / 0.5)

    async for message in websocket:
        chunk = np.frombuffer(message, dtype=np.float32)
        audio_buffer.append(chunk)

        speech_ts = _get_speech_timestamps(
            chunk, _vad_model, sampling_rate=SAMPLE_RATE, threshold=0.5
        )
        silence_frames = 0 if speech_ts else silence_frames + 1

        accumulated = np.concatenate(list(audio_buffer))
        if len(accumulated) >= SAMPLE_RATE * 1.5 or silence_frames >= max_silence:
            buf = io.BytesIO()
            sf.write(buf, accumulated, SAMPLE_RATE, format="WAV")
            buf.seek(0)

            segments, _ = _whisper_model.transcribe(
                buf, beam_size=1, language="it", vad_filter=False
            )
            transcript = " ".join(seg.text.strip() for seg in segments)
            if transcript:
                await websocket.send(transcript)

            if silence_frames >= max_silence:
                audio_buffer.clear()
                silence_frames = 0
                await websocket.send("__END_OF_UTTERANCE__")


async def main() -> None:
    async with websockets.serve(stream_transcribe, "0.0.0.0", 8765):
        await asyncio.Future()

# requirements: piper-tts==1.2.0 (OHF fork), sounddevice==0.4.7
import numpy as np
import sounddevice as sd
from piper.voice import PiperVoice
from pathlib import Path


class PiperStreamer:
    """TTS streaming: emette chunk audio mentre il modello genera."""

    SAMPLE_RATE = 22_050
    CHUNK_FRAMES = 1_024  # ~46ms

    def __init__(self, model_path: Path, config_path: Path):
        self._voice = PiperVoice.load(
            str(model_path), config_path=str(config_path), use_cuda=False
        )
        self._stream = sd.OutputStream(
            samplerate=self.SAMPLE_RATE, channels=1, dtype="int16"
        )

    async def speak(self, text: str) -> None:
        self._stream.start()
        try:
            buf = np.array([], dtype=np.int16)
            for audio_bytes in self._voice.synthesize_stream_raw(text):
                chunk = np.frombuffer(audio_bytes, dtype=np.int16)
                buf = np.concatenate([buf, chunk])
                while len(buf) >= self.CHUNK_FRAMES:
                    self._stream.write(buf[: self.CHUNK_FRAMES])
                    buf = buf[self.CHUNK_FRAMES :]
            if len(buf):
                self._stream.write(buf)
        finally:
            self._stream.stop()

# Event bus asyncio in-process
import asyncio
from dataclasses import dataclass
from typing import Any, Callable, Awaitable


@dataclass
class Event:
    topic: str
    payload: Any
    source: str = ""


EventHandler = Callable[[Event], Awaitable[None]]


class EventBus:
    def __init__(self) -> None:
        self._subscribers: dict[str, list[EventHandler]] = {}

    def subscribe(self, topic: str, handler: EventHandler) -> None:
        self._subscribers.setdefault(topic, []).append(handler)

    async def publish(self, event: Event) -> None:
        handlers = self._subscribers.get(event.topic, [])
        await asyncio.gather(*(h(event) for h in handlers))

class InterruptibleTTSWorker:
    def __init__(self, bus: EventBus, streamer: PiperStreamer) -> None:
        self._bus = bus
        self._streamer = streamer
        self._playing = asyncio.Event()
        self._interrupted = asyncio.Event()
        bus.subscribe("barge_in_detected", self._on_barge_in)

    async def _on_barge_in(self, event: Event) -> None:
        if self._playing.is_set():
            self._interrupted.set()

    async def speak(self, text: str) -> None:
        self._playing.set()
        self._interrupted.clear()
        speak_task = asyncio.create_task(self._streamer.speak(text))
        interrupt_task = asyncio.create_task(self._interrupted.wait())

        _, pending = await asyncio.wait(
            {speak_task, interrupt_task}, return_when=asyncio.FIRST_COMPLETED
        )
        for task in pending:
            task.cancel()
        self._playing.clear()

class PowerAwareMicController:
    INACTIVITY_TIMEOUT_S = 30.0
    WRIST_RAISE_THRESHOLD_G = 0.8

    async def run_power_loop(self) -> None:
        last_activity = time.monotonic()
        mic_active = False

        async for sensor_event in self._sensor_stream():
            if sensor_event.type == "wrist_raise" and not mic_active:
                await self._activate_mic()
                mic_active = True
            if sensor_event.type in ("wrist_raise", "wake_word_detected"):
                last_activity = time.monotonic()

            if mic_active and (time.monotonic() - last_activity) > self.INACTIVITY_TIMEOUT_S:
                await self._deactivate_mic()
                mic_active = False