commit 16d70bd9764c5de9228cf9f4a8d49f003563e914 Author: Magzhan Zhumabayev Date: Thu Apr 23 01:22:40 2026 +0500 Initial commit diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3d53093 --- /dev/null +++ b/.env.example @@ -0,0 +1,27 @@ +REALTIME_VOICE_HTTP_HOST=0.0.0.0 +REALTIME_VOICE_HTTP_PORT=8000 +REALTIME_VOICE_AUDIOSOCKET_HOST=0.0.0.0 +REALTIME_VOICE_AUDIOSOCKET_PORT=9092 + +OPENAI_API_KEY= +OPENAI_BASE_URL= +OPENAI_LLM_MODEL=gpt-4o-mini +OPENAI_LLM_SYSTEM_PROMPT=You are a concise voice assistant for a telecom call center. Answer clearly and briefly. +OPENAI_TIMEOUT_SECONDS=30 + +ELEVENLABS_API_KEY= +ELEVENLABS_API_BASE=https://api.elevenlabs.io +ELEVENLABS_TTS_VOICE_ID= +ELEVENLABS_TTS_MODEL_ID=eleven_flash_v2_5 +ELEVENLABS_TTS_LANGUAGE_CODE=ru +ELEVENLABS_TTS_OUTPUT_FORMAT=pcm_16000 +ELEVENLABS_STT_MODEL_ID=scribe_v2 +ELEVENLABS_STT_LANGUAGE_CODE= +ELEVENLABS_TIMEOUT_SECONDS=30 + +VAD_THRESHOLD=0.5 +VAD_NEGATIVE_THRESHOLD= +VAD_SILENCE_TIMEOUT_MS=1600 +VAD_SPEECH_PAD_MS=64 +VAD_MIN_SPEECH_DURATION_MS=0 +VAD_USE_ONNX=false diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..d5bc584 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,30 @@ +FROM python:3.11-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PYTHONPATH=/app + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + ca-certificates \ + ffmpeg \ + libasound2 \ + libgomp1 \ + libsndfile1 \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt /tmp/requirements.txt + +RUN python -m pip install --upgrade pip setuptools wheel \ + && python -m pip install -r /tmp/requirements.txt + +RUN mkdir -p /app/realtime_voice_service + +COPY . /app/realtime_voice_service + +EXPOSE 8000 9092 + +CMD ["sh", "-c", "python -m uvicorn realtime_voice_service.main:app --host 0.0.0.0 --port ${REALTIME_VOICE_HTTP_PORT:-8000}"] diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..c9c2ef6 --- /dev/null +++ b/__init__.py @@ -0,0 +1 @@ +__all__: list[str] = [] diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..d08bc33 --- /dev/null +++ b/core/__init__.py @@ -0,0 +1,4 @@ +from realtime_voice_service.core.session import CallSession, SessionState +from realtime_voice_service.core.vad import BaseVAD, SileroVADDetector, VADFrameResult + +__all__ = ["BaseVAD", "CallSession", "SessionState", "SileroVADDetector", "VADFrameResult"] diff --git a/core/session.py b/core/session.py new file mode 100644 index 0000000..032075e --- /dev/null +++ b/core/session.py @@ -0,0 +1,389 @@ +from __future__ import annotations + +import asyncio +import contextlib +import logging +import time +from collections.abc import Iterable +from enum import Enum + +from realtime_voice_service.core.vad import BaseVAD, SileroVADDetector +from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseTTS, MockLLM, MockSTT, MockTTS +from realtime_voice_service.transports.base import BaseMediaTransport + + +LOGGER = logging.getLogger("uvicorn.error") + + +class SessionState(str, Enum): + LISTENING = "LISTENING" + USER_SPEAKING = "USER_SPEAKING" + ASSISTANT_THINKING = "ASSISTANT_THINKING" + ASSISTANT_SPEAKING = "ASSISTANT_SPEAKING" + + +class GenerationInterrupted(RuntimeError): + pass + + +class TextChunker: + _BOUNDARY_CHARS = ".!?\n" + + def __init__(self) -> None: + self._buffer = "" + + def feed(self, fragment: str) -> list[str]: + if not fragment: + return [] + self._buffer += fragment + return self._extract_ready_chunks() + + def flush(self) -> list[str]: + chunks = self._extract_ready_chunks() + tail = self._buffer.strip() + self._buffer = "" + if tail: + chunks.append(tail) + return chunks + + def reset(self) -> None: + self._buffer = "" + + def _extract_ready_chunks(self) -> list[str]: + chunks: list[str] = [] + while True: + next_chunk = self._pop_next_chunk() + if next_chunk is None: + break + chunks.append(next_chunk) + return chunks + + def _pop_next_chunk(self) -> str | None: + for index, char in enumerate(self._buffer): + if char not in self._BOUNDARY_CHARS: + continue + end_index = index + 1 + while end_index < len(self._buffer) and self._buffer[end_index] in self._BOUNDARY_CHARS: + end_index += 1 + while end_index < len(self._buffer) and self._buffer[end_index].isspace(): + end_index += 1 + chunk = self._buffer[:end_index].strip() + self._buffer = self._buffer[end_index:] + if chunk: + return chunk + return None + + +class CallSession: + def __init__( + self, + *, + session_id: str, + transport: BaseMediaTransport, + vad: BaseVAD | None = None, + stt: BaseSTT | None = None, + llm: BaseLLM | None = None, + tts: BaseTTS | None = None, + ) -> None: + self.session_id = session_id + self.transport = transport + self.state = SessionState.LISTENING + self.generation_epoch = 0 + self.interruptions: list[str] = [] + self.last_latency_ms: dict[str, int] = {} + self._vad = vad or SileroVADDetector(sample_rate_hz=transport.sample_rate_hz) + self._stt = stt or MockSTT(sample_rate_hz=transport.sample_rate_hz) + self._llm = llm or MockLLM() + self._tts = tts or MockTTS(sample_rate_hz=transport.sample_rate_hz) + self._conversation: list[tuple[str, str]] = [] + self._assistant_task: asyncio.Task[None] | None = None + self._sentence_queue: asyncio.Queue[str | None] | None = None + self._closed = False + + @property + def conversation(self) -> tuple[tuple[str, str], ...]: + return tuple(self._conversation) + + async def run(self) -> None: + if self._assistant_task is not None: + raise RuntimeError("CallSession.run() can only be called once per session") + try: + await self.media_loop() + finally: + await self.stop() + + async def media_loop(self) -> None: + while not self._closed: + audio_chunk = await self.transport.receive_audio() + if audio_chunk is None: + LOGGER.info("realtime session %s transport closed", self.session_id) + break + if not audio_chunk: + continue + + vad_result = self._vad.feed(audio_chunk) + + if vad_result.is_speech and self.state != SessionState.USER_SPEAKING: + if self.state in {SessionState.ASSISTANT_THINKING, SessionState.ASSISTANT_SPEAKING}: + self.interrupt("barge-in") + self._set_state( + SessionState.USER_SPEAKING, + reason=f"speech detected prob={vad_result.speech_probability:.3f}", + ) + + if vad_result.speech_ended and vad_result.utterance_audio: + speech_end_monotonic = time.perf_counter() + self._set_state(SessionState.ASSISTANT_THINKING, reason="speech end detected") + self._start_assistant_turn( + epoch=self.generation_epoch, + utterance_audio=vad_result.utterance_audio, + speech_end_monotonic=speech_end_monotonic, + ) + + def interrupt(self, reason: str = "interrupt") -> int: + self.generation_epoch += 1 + self.interruptions.append(reason) + self._clear_sentence_queue() + if self._assistant_task is not None and not self._assistant_task.done(): + self._assistant_task.cancel() + LOGGER.info( + "realtime session %s interrupted: epoch=%s reason=%s", + self.session_id, + self.generation_epoch, + reason, + ) + return self.generation_epoch + + async def stop(self) -> None: + if self._closed: + return + self._closed = True + self._vad.reset() + self._clear_sentence_queue() + if self._assistant_task is not None and not self._assistant_task.done(): + self._assistant_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._assistant_task + await self.transport.close() + + def _start_assistant_turn( + self, + *, + epoch: int, + utterance_audio: bytes, + speech_end_monotonic: float, + ) -> None: + if self._assistant_task is not None and not self._assistant_task.done(): + self._assistant_task.cancel() + self._assistant_task = asyncio.create_task( + self._run_assistant_turn( + epoch=epoch, + utterance_audio=utterance_audio, + speech_end_monotonic=speech_end_monotonic, + ), + name=f"{self.session_id}-assistant-{epoch}", + ) + + async def _run_assistant_turn( + self, + *, + epoch: int, + utterance_audio: bytes, + speech_end_monotonic: float, + ) -> None: + sentence_queue: asyncio.Queue[str | None] | None = None + playback_task: asyncio.Task[None] | None = None + assistant_fragments: list[str] = [] + try: + transcript = await self._stt.transcribe(utterance_audio) + self._ensure_generation(epoch) + self._log_latency( + "stt_latency", + speech_end_monotonic, + epoch=epoch, + message="speech end -> transcript ready", + ) + transcript = transcript.strip() + if not transcript: + LOGGER.info("realtime session %s produced empty transcript", self.session_id) + self._set_state(SessionState.LISTENING, reason="empty transcript") + return + + self._conversation.append(("user", transcript)) + sentence_queue = asyncio.Queue() + self._sentence_queue = sentence_queue + playback_task = asyncio.create_task( + self._stream_tts_pipeline(epoch=epoch, sentence_queue=sentence_queue), + name=f"{self.session_id}-playback-{epoch}", + ) + llm_started_monotonic = time.perf_counter() + chunker = TextChunker() + first_token_seen = False + async for token in self._llm.generate_stream(transcript, list(self._conversation)): + self._ensure_generation(epoch) + if not token: + continue + if not first_token_seen: + first_token_seen = True + self._log_latency( + "ttft", + llm_started_monotonic, + epoch=epoch, + message="llm request -> first token", + ) + assistant_fragments.append(token) + await self._enqueue_chunks( + epoch=epoch, + sentence_queue=sentence_queue, + chunks=chunker.feed(token), + ) + + assistant_text = "".join(assistant_fragments).strip() + if not assistant_text: + await self._finish_sentence_queue(sentence_queue) + LOGGER.warning("realtime session %s llm produced no text", self.session_id) + self._set_state(SessionState.LISTENING, reason="empty llm response") + return + + await self._enqueue_chunks( + epoch=epoch, + sentence_queue=sentence_queue, + chunks=chunker.flush(), + ) + await self._finish_sentence_queue(sentence_queue) + if playback_task is not None: + await playback_task + self._ensure_generation(epoch) + self._conversation.append(("assistant", assistant_text)) + self._set_state(SessionState.LISTENING, reason="assistant turn completed") + except GenerationInterrupted: + LOGGER.info( + "realtime session %s ignored stale generation %s", + self.session_id, + epoch, + ) + except asyncio.CancelledError: + LOGGER.info( + "realtime session %s cancelled generation %s", + self.session_id, + epoch, + ) + raise + except Exception: + LOGGER.exception( + "realtime session %s generation %s failed", + self.session_id, + epoch, + ) + if epoch == self.generation_epoch and not self._closed: + self._set_state(SessionState.LISTENING, reason="assistant generation failed") + finally: + self._clear_sentence_queue(sentence_queue) + if playback_task is not None and not playback_task.done(): + playback_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await playback_task + current_task = asyncio.current_task() + if self._assistant_task is current_task: + self._assistant_task = None + + def _ensure_generation(self, epoch: int) -> None: + if self._closed or epoch != self.generation_epoch: + raise GenerationInterrupted(f"stale generation {epoch}") + + def _log_latency( + self, + metric_name: str, + started_monotonic: float, + *, + epoch: int, + message: str, + ) -> None: + latency_ms = max(int((time.perf_counter() - started_monotonic) * 1000.0), 0) + self.last_latency_ms[metric_name] = latency_ms + LOGGER.info( + "realtime session %s %s=%sms epoch=%s (%s)", + self.session_id, + metric_name, + latency_ms, + epoch, + message, + ) + + async def _enqueue_chunks( + self, + *, + epoch: int, + sentence_queue: asyncio.Queue[str | None], + chunks: Iterable[str], + ) -> None: + for chunk in chunks: + self._ensure_generation(epoch) + normalized = chunk.strip() + if not normalized: + continue + await sentence_queue.put(normalized) + + async def _finish_sentence_queue(self, sentence_queue: asyncio.Queue[str | None]) -> None: + await sentence_queue.put(None) + + async def _stream_tts_pipeline( + self, + *, + epoch: int, + sentence_queue: asyncio.Queue[str | None], + ) -> None: + first_audio_seen = False + first_tts_started_monotonic: float | None = None + while True: + self._ensure_generation(epoch) + sentence = await sentence_queue.get() + if sentence is None: + return + normalized = sentence.strip() + if not normalized: + continue + if first_tts_started_monotonic is None: + first_tts_started_monotonic = time.perf_counter() + async for audio_chunk in self._tts.synthesize_stream(normalized): + self._ensure_generation(epoch) + if not audio_chunk: + continue + if not first_audio_seen: + first_audio_seen = True + self._log_latency( + "ttfa", + first_tts_started_monotonic or time.perf_counter(), + epoch=epoch, + message="tts request -> first audio", + ) + self._set_state(SessionState.ASSISTANT_SPEAKING, reason="assistant playback started") + await self.transport.send_audio(audio_chunk) + + def _clear_sentence_queue(self, sentence_queue: asyncio.Queue[str | None] | None = None) -> None: + queue = sentence_queue if sentence_queue is not None else self._sentence_queue + if queue is None: + return + while True: + try: + queue.get_nowait() + except asyncio.QueueEmpty: + break + with contextlib.suppress(asyncio.QueueFull): + queue.put_nowait(None) + if queue is self._sentence_queue: + self._sentence_queue = None + + def _set_state(self, state: SessionState, *, reason: str | None = None) -> None: + if self.state == state: + return + previous_state = self.state + self.state = state + suffix = f" ({reason})" if reason else "" + LOGGER.info( + "realtime session %s state %s -> %s%s", + self.session_id, + previous_state.value, + state.value, + suffix, + ) diff --git a/core/vad.py b/core/vad.py new file mode 100644 index 0000000..ea53fd9 --- /dev/null +++ b/core/vad.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import logging +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Callable + + +LOGGER = logging.getLogger("uvicorn.error") + + +@dataclass +class VADFrameResult: + is_speech: bool = False + speech_started: bool = False + speech_ended: bool = False + speech_probability: float = 0.0 + utterance_audio: bytes | None = None + + +class BaseVAD(ABC): + @abstractmethod + def feed(self, audio_chunk: bytes) -> VADFrameResult: + raise NotImplementedError + + @abstractmethod + def flush(self) -> bytes | None: + raise NotImplementedError + + @abstractmethod + def reset(self) -> None: + raise NotImplementedError + + +class SileroVADDetector(BaseVAD): + def __init__( + self, + *, + sample_rate_hz: int = 8000, + threshold: float = 0.5, + negative_threshold: float | None = None, + speech_end_silence_ms: int = 1600, + speech_pad_ms: int = 64, + min_speech_duration_ms: int = 0, + model: Any | None = None, + prediction_fn: Callable[[bytes], float] | None = None, + use_onnx: bool = False, + ) -> None: + if sample_rate_hz not in {8000, 16000}: + raise ValueError("Silero VAD supports only 8000 Hz and 16000 Hz sample rates") + self.sample_rate_hz = sample_rate_hz + self.threshold = max(min(threshold, 1.0), 0.0) + self.negative_threshold = ( + max(min(negative_threshold, 1.0), 0.0) + if negative_threshold is not None + else max(self.threshold - 0.15, 0.01) + ) + self.speech_end_silence_ms = max(speech_end_silence_ms, 32) + self.speech_pad_ms = max(speech_pad_ms, 0) + self.min_speech_duration_ms = max(min_speech_duration_ms, 0) + self.window_samples = 512 if self.sample_rate_hz == 16000 else 256 + self.window_bytes = self.window_samples * 2 + self._speech_end_silence_samples = int(self.sample_rate_hz * self.speech_end_silence_ms / 1000.0) + self._speech_pad_bytes = int(self.sample_rate_hz * self.speech_pad_ms / 1000.0) * 2 + self._min_speech_samples = int(self.sample_rate_hz * self.min_speech_duration_ms / 1000.0) + self._model = model + self._prediction_fn = prediction_fn + self._use_onnx = use_onnx + self.reset() + + def reset(self) -> None: + self._window_buffer = bytearray() + self._pre_speech_audio = bytearray() + self._utterance_audio = bytearray() + self._triggered = False + self._silence_samples = 0 + self._speech_samples = 0 + if self._model is not None and hasattr(self._model, "reset_states"): + self._model.reset_states() + + def feed(self, audio_chunk: bytes) -> VADFrameResult: + result = VADFrameResult() + if not audio_chunk: + return result + + self._window_buffer.extend(audio_chunk) + while len(self._window_buffer) >= self.window_bytes: + window = bytes(self._window_buffer[: self.window_bytes]) + del self._window_buffer[: self.window_bytes] + speech_probability = self._predict_speech_probability(window) + result.speech_probability = speech_probability + result.is_speech = result.is_speech or self._triggered or speech_probability >= self.threshold + + if not self._triggered: + if speech_probability >= self.threshold: + self._triggered = True + self._speech_samples = self.window_samples + self._silence_samples = 0 + self._utterance_audio = bytearray(self._pre_speech_audio) + self._utterance_audio.extend(window) + self._pre_speech_audio.clear() + result.speech_started = True + result.is_speech = True + else: + self._append_pre_speech_window(window) + continue + + self._utterance_audio.extend(window) + result.is_speech = True + + if speech_probability >= self.threshold: + self._silence_samples = 0 + self._speech_samples += self.window_samples + continue + + if speech_probability >= self.negative_threshold: + self._silence_samples = 0 + continue + + self._silence_samples += self.window_samples + if self._silence_samples < self._speech_end_silence_samples: + continue + + utterance_audio = bytes(self._utterance_audio) + speech_samples = self._speech_samples + self._reset_segment() + result.is_speech = False + if speech_samples >= self._min_speech_samples: + result.speech_ended = True + result.utterance_audio = utterance_audio + + return result + + def flush(self) -> bytes | None: + if not self._triggered or not self._utterance_audio: + return None + utterance_audio = bytes(self._utterance_audio) + speech_samples = self._speech_samples + self._reset_segment() + if speech_samples < self._min_speech_samples: + return None + return utterance_audio + + def _reset_segment(self) -> None: + self._triggered = False + self._silence_samples = 0 + self._speech_samples = 0 + self._utterance_audio = bytearray() + self._pre_speech_audio.clear() + + def _append_pre_speech_window(self, window: bytes) -> None: + self._pre_speech_audio.extend(window) + if self._speech_pad_bytes <= 0: + self._pre_speech_audio.clear() + return + overflow = len(self._pre_speech_audio) - self._speech_pad_bytes + if overflow > 0: + del self._pre_speech_audio[:overflow] + + def _predict_speech_probability(self, window: bytes) -> float: + if self._prediction_fn is not None: + return max(0.0, min(float(self._prediction_fn(window)), 1.0)) + + model = self._load_model() + try: + import numpy as np + import torch + except Exception as exc: # noqa: BLE001 + raise RuntimeError( + "Silero VAD requires numpy and torch at runtime; add them to requirements.txt" + ) from exc + + pcm = np.frombuffer(window, dtype=np.int16).astype(np.float32) / 32768.0 + tensor = torch.from_numpy(pcm) + probability = model(tensor, self.sample_rate_hz).item() + return max(0.0, min(float(probability), 1.0)) + + def _load_model(self) -> Any: + if self._model is not None: + return self._model + try: + import torch + except Exception as exc: # noqa: BLE001 + raise RuntimeError( + "Silero VAD requires torch; install torch/torchaudio or use the ONNX runtime path" + ) from exc + + torch.set_num_threads(1) + try: + from silero_vad import load_silero_vad + except Exception: # noqa: BLE001 + load_silero_vad = None + + if load_silero_vad is not None: + self._model = load_silero_vad(onnx=self._use_onnx) + LOGGER.info( + "realtime voice VAD loaded via silero_vad package sample_rate=%s onnx=%s", + self.sample_rate_hz, + self._use_onnx, + ) + return self._model + + try: + self._model, _ = torch.hub.load( + repo_or_dir="snakers4/silero-vad", + model="silero_vad", + onnx=self._use_onnx, + force_reload=False, + ) + except TypeError: + self._model, _ = torch.hub.load( + repo_or_dir="snakers4/silero-vad", + model="silero_vad", + force_reload=False, + ) + LOGGER.info( + "realtime voice VAD loaded via torch.hub sample_rate=%s onnx=%s", + self.sample_rate_hz, + self._use_onnx, + ) + return self._model diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..49befc8 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,12 @@ +services: + realtime_voice: + build: + context: . + dockerfile: Dockerfile + container_name: realtime_voice + env_file: + - .env + ports: + - "${REALTIME_VOICE_HTTP_PORT:-8000}:${REALTIME_VOICE_HTTP_PORT:-8000}" + - "${REALTIME_VOICE_AUDIOSOCKET_PORT:-9092}:${REALTIME_VOICE_AUDIOSOCKET_PORT:-9092}" + restart: unless-stopped diff --git a/main.py b/main.py new file mode 100644 index 0000000..6af5cf1 --- /dev/null +++ b/main.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import asyncio +import logging +import os +import uuid +from contextlib import asynccontextmanager + +from fastapi import FastAPI, WebSocket + +from realtime_voice_service.core.session import CallSession +from realtime_voice_service.core.vad import SileroVADDetector +from realtime_voice_service.providers.llm import OpenAILLM +from realtime_voice_service.providers.stt import ElevenLabsSTT +from realtime_voice_service.providers.tts import ElevenLabsTTS +from realtime_voice_service.transports.audiosocket import AudioSocketServer +from realtime_voice_service.transports.base import BaseMediaTransport +from realtime_voice_service.transports.websocket import WebSocketMediaTransport + + +LOGGER = logging.getLogger("uvicorn.error") + + +def _int_env(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(str(raw).strip()) + except ValueError: + return default + + +def _float_env(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return default + try: + return float(str(raw).strip()) + except ValueError: + return default + + +def _optional_float_env(name: str) -> float | None: + raw = os.getenv(name) + if raw is None: + return None + normalized = str(raw).strip() + if not normalized: + return None + try: + return float(normalized) + except ValueError: + return None + + +def _bool_env(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return str(raw).strip().lower() in {"1", "true", "yes", "on"} + + +def _audiosocket_host() -> str: + return str(os.getenv("REALTIME_VOICE_AUDIOSOCKET_HOST", "0.0.0.0") or "0.0.0.0").strip() + + +def _audiosocket_port() -> int: + return max(_int_env("REALTIME_VOICE_AUDIOSOCKET_PORT", 9092), 1) + + +def _http_host() -> str: + return str(os.getenv("REALTIME_VOICE_HTTP_HOST", "0.0.0.0") or "0.0.0.0").strip() + + +def _http_port() -> int: + return max(_int_env("REALTIME_VOICE_HTTP_PORT", 8000), 1) + + +class RealtimeVoiceService: + def __init__(self) -> None: + self._stt = ElevenLabsSTT() + self._llm = OpenAILLM() + self._tts = ElevenLabsTTS() + self._audiosocket_server = AudioSocketServer( + host=_audiosocket_host(), + port=_audiosocket_port(), + session_handler=self._run_transport_session, + ) + self._active_sessions: dict[str, CallSession] = {} + self._session_lock = asyncio.Lock() + + @property + def active_session_count(self) -> int: + return len(self._active_sessions) + + async def start(self) -> None: + await self._audiosocket_server.start() + + async def stop(self) -> None: + await self._audiosocket_server.stop() + sessions = list(self._active_sessions.values()) + for session in sessions: + await session.stop() + self._active_sessions.clear() + + async def handle_websocket(self, websocket: WebSocket, *, client_id: str | None = None) -> None: + await websocket.accept() + transport = WebSocketMediaTransport( + websocket=websocket, + transport_id=client_id or str(uuid.uuid4()), + ) + await self._run_transport_session(transport) + + async def _run_transport_session(self, transport: BaseMediaTransport) -> None: + session = self._build_session(transport) + async with self._track_session(session): + LOGGER.info( + "starting realtime session %s via %s", + session.session_id, + transport.protocol, + ) + await session.run() + + def _build_session(self, transport: BaseMediaTransport) -> CallSession: + return CallSession( + session_id=transport.transport_id, + transport=transport, + vad=SileroVADDetector( + sample_rate_hz=transport.sample_rate_hz, + threshold=_float_env("VAD_THRESHOLD", 0.5), + negative_threshold=_optional_float_env("VAD_NEGATIVE_THRESHOLD"), + speech_end_silence_ms=_int_env("VAD_SILENCE_TIMEOUT_MS", 1600), + speech_pad_ms=_int_env("VAD_SPEECH_PAD_MS", 64), + min_speech_duration_ms=_int_env("VAD_MIN_SPEECH_DURATION_MS", 0), + use_onnx=_bool_env("VAD_USE_ONNX", False), + ), + stt=self._stt, + llm=self._llm, + tts=self._tts, + ) + + @asynccontextmanager + async def _track_session(self, session: CallSession): + async with self._session_lock: + self._active_sessions[session.session_id] = session + try: + yield + finally: + async with self._session_lock: + self._active_sessions.pop(session.session_id, None) + + +service = RealtimeVoiceService() + + +@asynccontextmanager +async def lifespan(_: FastAPI): + await service.start() + try: + yield + finally: + await service.stop() + + +app = FastAPI(title="realtime-voice-service", version="0.1.0", lifespan=lifespan) + + +@app.get("/health") +async def health() -> dict[str, object]: + return { + "status": "ok", + "service": "realtime-voice-service", + "active_sessions": service.active_session_count, + "audiosocket_port": _audiosocket_port(), + } + + +@app.websocket("/ws/{client_id}") +async def websocket_media(websocket: WebSocket, client_id: str) -> None: + await service.handle_websocket(websocket, client_id=client_id) + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run( + "realtime_voice_service.main:app", + host=_http_host(), + port=_http_port(), + reload=False, + ) diff --git a/providers/__init__.py b/providers/__init__.py new file mode 100644 index 0000000..95a21fa --- /dev/null +++ b/providers/__init__.py @@ -0,0 +1,16 @@ +from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseTTS, MockLLM, MockSTT, MockTTS +from realtime_voice_service.providers.llm import OpenAILLM +from realtime_voice_service.providers.stt import ElevenLabsSTT +from realtime_voice_service.providers.tts import ElevenLabsTTS + +__all__ = [ + "BaseLLM", + "BaseSTT", + "BaseTTS", + "ElevenLabsSTT", + "ElevenLabsTTS", + "MockLLM", + "MockSTT", + "MockTTS", + "OpenAILLM", +] diff --git a/providers/base.py b/providers/base.py new file mode 100644 index 0000000..75c2f85 --- /dev/null +++ b/providers/base.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import asyncio +import math +import re +import struct +from abc import ABC, abstractmethod +from collections.abc import AsyncGenerator + + +class BaseSTT(ABC): + @abstractmethod + async def transcribe(self, audio_bytes: bytes) -> str: + raise NotImplementedError + + +class BaseLLM(ABC): + @abstractmethod + async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]: + raise NotImplementedError + + +class BaseTTS(ABC): + @abstractmethod + async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]: + raise NotImplementedError + + +class MockSTT(BaseSTT): + def __init__( + self, + *, + latency_ms: int = 40, + sample_rate_hz: int = 8000, + scripted_transcripts: list[str] | None = None, + ) -> None: + self._latency_ms = max(latency_ms, 0) + self._sample_rate_hz = sample_rate_hz + self._scripted_transcripts = list(scripted_transcripts or []) + self._call_count = 0 + + async def transcribe(self, audio_bytes: bytes) -> str: + await asyncio.sleep(self._latency_ms / 1000.0) + self._call_count += 1 + if self._scripted_transcripts: + return self._scripted_transcripts.pop(0) + duration_ms = int(((len(audio_bytes) // 2) / float(self._sample_rate_hz)) * 1000.0) + return f"mock user utterance {self._call_count} ({duration_ms} ms)" + + +class MockLLM(BaseLLM): + def __init__( + self, + *, + token_delay_ms: int = 35, + scripted_responses: list[str] | None = None, + ) -> None: + self._token_delay_ms = max(token_delay_ms, 0) + self._scripted_responses = list(scripted_responses or []) + + async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]: + del context + response_text = ( + self._scripted_responses.pop(0) + if self._scripted_responses + else f"Mock reply: {text}" + ) + chunks = re.findall(r"\S+\s*", response_text) or [response_text] + for chunk in chunks: + await asyncio.sleep(self._token_delay_ms / 1000.0) + yield chunk + + +class MockTTS(BaseTTS): + def __init__( + self, + *, + sample_rate_hz: int = 8000, + chunk_duration_ms: int = 40, + chunk_delay_ms: int = 15, + tone_hz: float = 440.0, + amplitude: int = 9000, + milliseconds_per_word: int = 140, + ) -> None: + self._sample_rate_hz = sample_rate_hz + self._chunk_duration_ms = max(chunk_duration_ms, 20) + self._chunk_delay_ms = max(chunk_delay_ms, 0) + self._tone_hz = tone_hz + self._amplitude = amplitude + self._milliseconds_per_word = max(milliseconds_per_word, 40) + + async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]: + word_count = max(len(text.split()), 1) + total_ms = min(word_count * self._milliseconds_per_word, 2400) + total_samples = max(int(self._sample_rate_hz * (total_ms / 1000.0)), 1) + chunk_samples = max(int(self._sample_rate_hz * (self._chunk_duration_ms / 1000.0)), 1) + + for start in range(0, total_samples, chunk_samples): + end = min(start + chunk_samples, total_samples) + pcm = bytearray() + for index in range(start, end): + angle = 2.0 * math.pi * self._tone_hz * (index / float(self._sample_rate_hz)) + sample = int(self._amplitude * math.sin(angle)) + pcm.extend(struct.pack(" float: + raw = os.getenv("OPENAI_TIMEOUT_SECONDS") + if raw is None: + return 30.0 + try: + return max(float(raw.strip()), 1.0) + except ValueError: + return 30.0 + + +class OpenAILLM(BaseLLM): + def __init__( + self, + *, + api_key: str | None = None, + model: str | None = None, + base_url: str | None = None, + system_prompt: str | None = None, + timeout_seconds: float | None = None, + temperature: float = 0.3, + max_retries: int = 2, + ) -> None: + self._api_key = str(api_key if api_key is not None else os.getenv("OPENAI_API_KEY", "")).strip() + self._model = str(model or os.getenv("OPENAI_LLM_MODEL", "gpt-4o-mini")).strip() or "gpt-4o-mini" + self._base_url = str(base_url or os.getenv("OPENAI_BASE_URL", "")).strip() or None + self._system_prompt = str( + system_prompt + if system_prompt is not None + else os.getenv( + "OPENAI_LLM_SYSTEM_PROMPT", + "You are a concise voice assistant for a telecom call center. Answer clearly and briefly.", + ) + ).strip() + self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0) + self._temperature = max(min(float(temperature), 2.0), 0.0) + self._max_retries = max(int(max_retries), 0) + self._client: Any | None = None + self._openai_module: Any | None = None + + async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]: + if not self._api_key: + raise RuntimeError("OPENAI_API_KEY is required for OpenAI LLM") + + client = self._get_client() + messages = self._build_messages(text, context) + try: + stream = await client.chat.completions.create( + model=self._model, + messages=messages, + temperature=self._temperature, + stream=True, + ) + async for chunk in stream: + choices = getattr(chunk, "choices", None) or [] + if not choices: + continue + delta = getattr(choices[0], "delta", None) + if delta is None: + continue + content = getattr(delta, "content", None) + if content: + yield str(content) + except Exception as exc: # noqa: BLE001 + openai_module = self._openai_module + if openai_module is not None and isinstance(exc, getattr(openai_module, "APITimeoutError", ())): + raise RuntimeError("OpenAI LLM request timed out") from exc + if openai_module is not None and isinstance(exc, getattr(openai_module, "RateLimitError", ())): + raise RuntimeError("OpenAI LLM rate limit exceeded") from exc + if openai_module is not None and isinstance(exc, getattr(openai_module, "APIStatusError", ())): + status_code = getattr(exc, "status_code", "unknown") + raise RuntimeError(f"OpenAI LLM returned HTTP {status_code}") from exc + if openai_module is not None and isinstance(exc, getattr(openai_module, "APIConnectionError", ())): + raise RuntimeError("OpenAI LLM connection failed") from exc + raise RuntimeError("OpenAI LLM streaming failed") from exc + + def _get_client(self): + if self._client is not None: + return self._client + try: + import httpx + import openai + from openai import AsyncOpenAI + except Exception as exc: # noqa: BLE001 + raise RuntimeError("The `openai` package is required for OpenAI LLM") from exc + + timeout = httpx.Timeout( + self._timeout_seconds, + connect=min(self._timeout_seconds, 5.0), + write=min(self._timeout_seconds, 15.0), + read=self._timeout_seconds, + ) + self._openai_module = openai + self._client = AsyncOpenAI( + api_key=self._api_key, + base_url=self._base_url, + timeout=timeout, + max_retries=self._max_retries, + ) + return self._client + + def _build_messages(self, text: str, context: list) -> list[dict[str, str]]: + messages: list[dict[str, str]] = [] + if self._system_prompt: + messages.append({"role": "system", "content": self._system_prompt}) + + for entry in context: + role: str | None = None + content: str | None = None + if isinstance(entry, dict): + role = str(entry.get("role") or "").strip().lower() or None + content = str(entry.get("content") or "").strip() or None + elif isinstance(entry, (tuple, list)) and len(entry) >= 2: + speaker = str(entry[0]).strip().lower() + role = "assistant" if speaker == "assistant" else "user" + content = str(entry[1]).strip() or None + if role and content: + messages.append({"role": role, "content": content}) + + if text.strip(): + if not messages or messages[-1].get("role") != "user" or messages[-1].get("content") != text: + messages.append({"role": "user", "content": text}) + return messages diff --git a/providers/stt.py b/providers/stt.py new file mode 100644 index 0000000..6e3aae9 --- /dev/null +++ b/providers/stt.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import asyncio +import audioop +import io +import json +import os +import wave + +from realtime_voice_service.providers.base import BaseSTT + + +def _api_base() -> str: + return (os.getenv("ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() or "https://api.elevenlabs.io").rstrip("/") + + +def _timeout_seconds() -> float: + raw = os.getenv("ELEVENLABS_TIMEOUT_SECONDS") + if raw is None: + return 30.0 + try: + return max(float(raw.strip()), 1.0) + except ValueError: + return 30.0 + + +def _pcm16le_to_wav_bytes( + pcm_bytes: bytes, + *, + sample_rate_hz: int, + channels: int = 1, + sample_width_bytes: int = 2, +) -> bytes: + handle = io.BytesIO() + with wave.open(handle, "wb") as wav_file: + wav_file.setnchannels(channels) + wav_file.setsampwidth(sample_width_bytes) + wav_file.setframerate(sample_rate_hz) + wav_file.writeframes(pcm_bytes) + return handle.getvalue() + + +def _resample_pcm16le( + pcm_bytes: bytes, + *, + input_rate_hz: int, + output_rate_hz: int, +) -> bytes: + if not pcm_bytes or input_rate_hz == output_rate_hz: + return pcm_bytes + converted, _ = audioop.ratecv( + pcm_bytes, + 2, + 1, + input_rate_hz, + output_rate_hz, + None, + ) + return converted + + +class ElevenLabsSTT(BaseSTT): + def __init__( + self, + *, + api_key: str | None = None, + api_base: str | None = None, + model_id: str | None = None, + input_sample_rate_hz: int = 8000, + target_sample_rate_hz: int = 16000, + timeout_seconds: float | None = None, + language_code: str | None = None, + ) -> None: + self._api_key = str(api_key if api_key is not None else os.getenv("ELEVENLABS_API_KEY", "")).strip() + self._api_base = str(api_base or _api_base()).strip().rstrip("/") + self._model_id = str(model_id or os.getenv("ELEVENLABS_STT_MODEL_ID", "scribe_v2")).strip() or "scribe_v2" + self._input_sample_rate_hz = max(int(input_sample_rate_hz), 1) + self._target_sample_rate_hz = max(int(target_sample_rate_hz), 1) + self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0) + self._language_code = str(language_code or os.getenv("ELEVENLABS_STT_LANGUAGE_CODE", "")).strip() or None + + async def transcribe(self, audio_bytes: bytes) -> str: + if not audio_bytes: + return "" + if not self._api_key: + raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs STT") + + pcm_bytes, sample_rate_hz = self._extract_pcm(audio_bytes) + if sample_rate_hz != self._target_sample_rate_hz: + pcm_bytes = _resample_pcm16le( + pcm_bytes, + input_rate_hz=sample_rate_hz, + output_rate_hz=self._target_sample_rate_hz, + ) + sample_rate_hz = self._target_sample_rate_hz + + wav_bytes = _pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=sample_rate_hz) + + try: + import aiohttp + except Exception as exc: # noqa: BLE001 + raise RuntimeError("The `aiohttp` package is required for ElevenLabs STT") from exc + + form = aiohttp.FormData() + form.add_field("model_id", self._model_id) + form.add_field("timestamps_granularity", "none") + form.add_field("diarize", "false") + if self._language_code: + form.add_field("language_code", self._language_code) + form.add_field( + "file", + wav_bytes, + filename="turn.wav", + content_type="audio/wav", + ) + + timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post( + f"{self._api_base}/v1/speech-to-text", + headers={"xi-api-key": self._api_key}, + data=form, + ) as response: + payload_text = await response.text() + if response.status >= 400: + raise RuntimeError( + f"ElevenLabs STT returned HTTP {response.status}: {payload_text[:300]}" + ) + except (TimeoutError, asyncio.TimeoutError) as exc: + raise RuntimeError("ElevenLabs STT request timed out") from exc + except aiohttp.ClientError as exc: + raise RuntimeError("ElevenLabs STT request failed") from exc + + try: + payload = json.loads(payload_text) + except json.JSONDecodeError as exc: + raise RuntimeError("ElevenLabs STT returned invalid JSON") from exc + return str(payload.get("text") or payload.get("transcript") or "").strip() + + def _extract_pcm(self, audio_bytes: bytes) -> tuple[bytes, int]: + try: + with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file: + pcm_bytes = wav_file.readframes(wav_file.getnframes()) + sample_width = wav_file.getsampwidth() + channels = wav_file.getnchannels() + sample_rate_hz = int(wav_file.getframerate() or self._input_sample_rate_hz) + if sample_width != 2: + return audio_bytes, self._input_sample_rate_hz + if channels == 2: + pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5) + return pcm_bytes, sample_rate_hz + except (wave.Error, EOFError): + return audio_bytes, self._input_sample_rate_hz diff --git a/providers/tts.py b/providers/tts.py new file mode 100644 index 0000000..724d4bb --- /dev/null +++ b/providers/tts.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import asyncio +import audioop +import os +from collections.abc import AsyncGenerator + +from realtime_voice_service.providers.base import BaseTTS + + +def _api_base() -> str: + return (os.getenv("ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() or "https://api.elevenlabs.io").rstrip("/") + + +def _timeout_seconds() -> float: + raw = os.getenv("ELEVENLABS_TIMEOUT_SECONDS") + if raw is None: + return 30.0 + try: + return max(float(raw.strip()), 1.0) + except ValueError: + return 30.0 + + +def _parse_output_format(output_format: str) -> tuple[str, int]: + normalized = str(output_format or "").strip().lower() + if normalized == "ulaw_8000": + return "ulaw", 8000 + if normalized == "alaw_8000": + return "alaw", 8000 + if not normalized.startswith("pcm_"): + raise ValueError("ElevenLabsTTS expects `pcm_*`, `ulaw_8000`, or `alaw_8000` output formats") + suffix = normalized.split("_", 1)[1] + try: + return "pcm16le", int(suffix) + except ValueError as exc: + raise ValueError(f"Unsupported ElevenLabs output format: {output_format}") from exc + + +class _PCM16StreamAdapter: + def __init__(self, *, input_codec: str, input_rate_hz: int, output_rate_hz: int) -> None: + self._input_codec = input_codec + self._input_rate_hz = input_rate_hz + self._output_rate_hz = output_rate_hz + self._carry = b"" + self._state = None + + def process(self, chunk: bytes) -> bytes: + if not chunk: + return b"" + data = self._carry + chunk + sample_width_bytes = 2 if self._input_codec == "pcm16le" else 1 + usable_length = len(data) - (len(data) % sample_width_bytes) + self._carry = data[usable_length:] + if usable_length <= 0: + return b"" + pcm16 = self._decode_to_pcm16(data[:usable_length]) + if self._input_rate_hz == self._output_rate_hz: + return pcm16 + converted, self._state = audioop.ratecv( + pcm16, + 2, + 1, + self._input_rate_hz, + self._output_rate_hz, + self._state, + ) + return converted + + def flush(self) -> bytes: + if not self._carry: + return b"" + sample_width_bytes = 2 if self._input_codec == "pcm16le" else 1 + padded = self._carry + (b"\x00" * ((sample_width_bytes - len(self._carry)) % sample_width_bytes)) + self._carry = b"" + pcm16 = self._decode_to_pcm16(padded) + if self._input_rate_hz == self._output_rate_hz: + return pcm16 + converted, self._state = audioop.ratecv( + pcm16, + 2, + 1, + self._input_rate_hz, + self._output_rate_hz, + self._state, + ) + return converted + + def _decode_to_pcm16(self, chunk: bytes) -> bytes: + if self._input_codec == "pcm16le": + return chunk + if self._input_codec == "ulaw": + return audioop.ulaw2lin(chunk, 2) + if self._input_codec == "alaw": + return audioop.alaw2lin(chunk, 2) + raise ValueError(f"Unsupported input codec: {self._input_codec}") + + +class ElevenLabsTTS(BaseTTS): + def __init__( + self, + *, + api_key: str | None = None, + api_base: str | None = None, + voice_id: str | None = None, + model_id: str | None = None, + language_code: str | None = None, + output_format: str | None = None, + target_sample_rate_hz: int = 8000, + timeout_seconds: float | None = None, + stream_chunk_bytes: int = 4096, + ) -> None: + self._api_key = str(api_key if api_key is not None else os.getenv("ELEVENLABS_API_KEY", "")).strip() + self._api_base = str(api_base or _api_base()).strip().rstrip("/") + self._voice_id = str(voice_id or os.getenv("ELEVENLABS_TTS_VOICE_ID", "")).strip() + self._model_id = ( + str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_flash_v2_5")).strip() + or "eleven_flash_v2_5" + ) + self._language_code = str(language_code or os.getenv("ELEVENLABS_TTS_LANGUAGE_CODE", "ru")).strip() or None + self._output_format = str(output_format or os.getenv("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000")).strip().lower() or "pcm_16000" + self._source_codec, self._source_sample_rate_hz = _parse_output_format(self._output_format) + self._target_sample_rate_hz = max(int(target_sample_rate_hz), 1) + self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0) + self._stream_chunk_bytes = max(int(stream_chunk_bytes), 256) + + async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]: + if not text.strip(): + return + if not self._api_key: + raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs TTS") + if not self._voice_id: + raise RuntimeError("ELEVENLABS_TTS_VOICE_ID is required for ElevenLabs TTS") + + try: + import aiohttp + except Exception as exc: # noqa: BLE001 + raise RuntimeError("The `aiohttp` package is required for ElevenLabs TTS") from exc + + payload = { + "text": text, + "model_id": self._model_id, + "apply_text_normalization": "auto", + } + if self._language_code: + payload["language_code"] = self._language_code + + adapter = _PCM16StreamAdapter( + input_codec=self._source_codec, + input_rate_hz=self._source_sample_rate_hz, + output_rate_hz=self._target_sample_rate_hz, + ) + timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post( + f"{self._api_base}/v1/text-to-speech/{self._voice_id}/stream", + params={"output_format": self._output_format}, + headers={ + "xi-api-key": self._api_key, + "Content-Type": "application/json", + }, + json=payload, + ) as response: + if response.status >= 400: + error_text = await response.text() + raise RuntimeError( + f"ElevenLabs TTS returned HTTP {response.status}: {error_text[:300]}" + ) + async for chunk in response.content.iter_chunked(self._stream_chunk_bytes): + if not chunk: + continue + converted = adapter.process(chunk) + if converted: + yield converted + except (TimeoutError, asyncio.TimeoutError) as exc: + raise RuntimeError("ElevenLabs TTS request timed out") from exc + except aiohttp.ClientError as exc: + raise RuntimeError("ElevenLabs TTS request failed") from exc + + tail = adapter.flush() + if tail: + yield tail diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ac19310 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +fastapi +uvicorn[standard] +websockets +numpy +torch +torchaudio +silero-vad +onnxruntime +openai +httpx +aiohttp diff --git a/transports/__init__.py b/transports/__init__.py new file mode 100644 index 0000000..50139ea --- /dev/null +++ b/transports/__init__.py @@ -0,0 +1,28 @@ +__all__ = [ + "AudioSocketServer", + "AudioSocketTransport", + "BaseMediaTransport", + "WebSocketMediaTransport", +] + + +def __getattr__(name: str): + if name == "BaseMediaTransport": + from realtime_voice_service.transports.base import BaseMediaTransport + + return BaseMediaTransport + if name in {"AudioSocketServer", "AudioSocketTransport"}: + from realtime_voice_service.transports.audiosocket import ( + AudioSocketServer, + AudioSocketTransport, + ) + + return { + "AudioSocketServer": AudioSocketServer, + "AudioSocketTransport": AudioSocketTransport, + }[name] + if name == "WebSocketMediaTransport": + from realtime_voice_service.transports.websocket import WebSocketMediaTransport + + return WebSocketMediaTransport + raise AttributeError(name) diff --git a/transports/audiosocket.py b/transports/audiosocket.py new file mode 100644 index 0000000..59cdd1c --- /dev/null +++ b/transports/audiosocket.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import asyncio +import logging +import struct +import uuid +from typing import Awaitable, Callable + +from realtime_voice_service.transports.base import BaseMediaTransport + + +LOGGER = logging.getLogger("uvicorn.error") + +AUDIO_SOCKET_PACKET_HANGUP = 0x00 +AUDIO_SOCKET_PACKET_UUID = 0x01 +AUDIO_SOCKET_PACKET_DTMF = 0x03 +AUDIO_SOCKET_PACKET_PCM16 = 0x10 + + +def normalize_session_id(value: str | bytes) -> str: + if isinstance(value, bytes): + try: + return str(uuid.UUID(bytes=value)).lower() + except (ValueError, AttributeError): + return str(uuid.UUID(value.decode("utf-8").strip())).lower() + return str(uuid.UUID(str(value).strip())).lower() + + +def encode_packet(packet_type: int, payload: bytes = b"") -> bytes: + return struct.pack("!BH", packet_type & 0xFF, len(payload)) + payload + + +def encode_audio_packet(pcm_bytes: bytes) -> bytes: + return encode_packet(AUDIO_SOCKET_PACKET_PCM16, pcm_bytes) + + +async def read_packet( + reader: asyncio.StreamReader, + *, + timeout_seconds: float = 5.0, +) -> tuple[int, bytes]: + header = await asyncio.wait_for(reader.readexactly(3), timeout=timeout_seconds) + packet_type, payload_length = struct.unpack("!BH", header) + payload = b"" + if payload_length: + payload = await asyncio.wait_for(reader.readexactly(payload_length), timeout=timeout_seconds) + return packet_type, payload + + +class AudioSocketTransport(BaseMediaTransport): + def __init__( + self, + *, + transport_id: str, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + sample_rate_hz: int = 8000, + frame_duration_ms: int = 20, + read_timeout_seconds: float = 30.0, + ) -> None: + super().__init__( + transport_id=transport_id, + sample_rate_hz=sample_rate_hz, + frame_duration_ms=frame_duration_ms, + ) + self._reader = reader + self._writer = writer + self._read_timeout_seconds = max(read_timeout_seconds, 1.0) + self._closed = False + + @property + def protocol(self) -> str: + return "audiosocket" + + @classmethod + async def from_streams( + cls, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + *, + handshake_timeout_seconds: float = 5.0, + sample_rate_hz: int = 8000, + frame_duration_ms: int = 20, + ) -> AudioSocketTransport: + packet_type, payload = await read_packet(reader, timeout_seconds=handshake_timeout_seconds) + if packet_type != AUDIO_SOCKET_PACKET_UUID: + raise ValueError(f"expected UUID packet, got {packet_type}") + session_id = normalize_session_id(payload) + return cls( + transport_id=session_id, + reader=reader, + writer=writer, + sample_rate_hz=sample_rate_hz, + frame_duration_ms=frame_duration_ms, + ) + + async def receive_audio(self) -> bytes | None: + if self._closed: + return None + while not self._closed: + try: + packet_type, payload = await read_packet( + self._reader, + timeout_seconds=self._read_timeout_seconds, + ) + except (asyncio.IncompleteReadError, asyncio.TimeoutError, ConnectionError): + return None + + if packet_type == AUDIO_SOCKET_PACKET_HANGUP: + return None + if packet_type == AUDIO_SOCKET_PACKET_PCM16: + return payload + if packet_type in {AUDIO_SOCKET_PACKET_UUID, AUDIO_SOCKET_PACKET_DTMF}: + continue + return None + + async def send_audio(self, audio_chunk: bytes) -> None: + if self._closed or self._writer.is_closing(): + return + self._writer.write(encode_audio_packet(audio_chunk)) + await self._writer.drain() + + async def close(self) -> None: + if self._closed: + return + self._closed = True + if not self._writer.is_closing(): + self._writer.close() + await self._writer.wait_closed() + + +class AudioSocketServer: + def __init__( + self, + *, + host: str, + port: int, + session_handler: Callable[[AudioSocketTransport], Awaitable[None]], + handshake_timeout_seconds: float = 5.0, + sample_rate_hz: int = 8000, + frame_duration_ms: int = 20, + ) -> None: + self._host = host + self._port = port + self._session_handler = session_handler + self._handshake_timeout_seconds = max(handshake_timeout_seconds, 1.0) + self._sample_rate_hz = sample_rate_hz + self._frame_duration_ms = frame_duration_ms + self._server: asyncio.base_events.Server | None = None + self._connection_tasks: set[asyncio.Task[None]] = set() + + @property + def bound_port(self) -> int: + if self._server is None or not self._server.sockets: + return self._port + return int(self._server.sockets[0].getsockname()[1]) + + async def start(self) -> None: + if self._server is not None: + return + self._server = await asyncio.start_server( + self._handle_connection, + self._host, + self._port, + ) + LOGGER.info( + "AudioSocket server listening on %s:%s", + self._host, + self.bound_port, + ) + + async def stop(self) -> None: + if self._server is None: + return + self._server.close() + await self._server.wait_closed() + self._server = None + + tasks = list(self._connection_tasks) + for task in tasks: + task.cancel() + for task in tasks: + try: + await task + except asyncio.CancelledError: + pass + + async def _handle_connection( + self, + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + current_task = asyncio.current_task() + if current_task is not None: + self._connection_tasks.add(current_task) + + transport: AudioSocketTransport | None = None + peer = writer.get_extra_info("peername") + try: + transport = await AudioSocketTransport.from_streams( + reader, + writer, + handshake_timeout_seconds=self._handshake_timeout_seconds, + sample_rate_hz=self._sample_rate_hz, + frame_duration_ms=self._frame_duration_ms, + ) + LOGGER.info( + "AudioSocket client accepted: session=%s peer=%s", + transport.transport_id, + peer, + ) + await self._session_handler(transport) + except asyncio.CancelledError: + raise + except Exception: + LOGGER.exception("AudioSocket connection failed from peer=%s", peer) + finally: + if transport is not None: + await transport.close() + elif not writer.is_closing(): + writer.close() + await writer.wait_closed() + if current_task is not None: + self._connection_tasks.discard(current_task) diff --git a/transports/base.py b/transports/base.py new file mode 100644 index 0000000..020271b --- /dev/null +++ b/transports/base.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod + + +class BaseMediaTransport(ABC): + def __init__( + self, + *, + transport_id: str, + sample_rate_hz: int = 8000, + frame_duration_ms: int = 20, + ) -> None: + self._transport_id = transport_id + self._sample_rate_hz = sample_rate_hz + self._frame_duration_ms = frame_duration_ms + + @property + def transport_id(self) -> str: + return self._transport_id + + @property + def sample_rate_hz(self) -> int: + return self._sample_rate_hz + + @property + def frame_duration_ms(self) -> int: + return self._frame_duration_ms + + @property + def frame_bytes(self) -> int: + return int((self.sample_rate_hz * self.frame_duration_ms / 1000.0) * 2) + + @property + @abstractmethod + def protocol(self) -> str: + raise NotImplementedError + + @abstractmethod + async def receive_audio(self) -> bytes | None: + raise NotImplementedError + + @abstractmethod + async def send_audio(self, audio_chunk: bytes) -> None: + raise NotImplementedError + + @abstractmethod + async def close(self) -> None: + raise NotImplementedError diff --git a/transports/websocket.py b/transports/websocket.py new file mode 100644 index 0000000..7d0fac7 --- /dev/null +++ b/transports/websocket.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import base64 +import json + +from fastapi import WebSocket +from starlette.websockets import WebSocketState + +from realtime_voice_service.transports.base import BaseMediaTransport + + +class WebSocketMediaTransport(BaseMediaTransport): + def __init__( + self, + *, + websocket: WebSocket, + transport_id: str, + sample_rate_hz: int = 8000, + frame_duration_ms: int = 20, + ) -> None: + super().__init__( + transport_id=transport_id, + sample_rate_hz=sample_rate_hz, + frame_duration_ms=frame_duration_ms, + ) + self._websocket = websocket + self._closed = False + + @property + def protocol(self) -> str: + return "websocket" + + async def receive_audio(self) -> bytes | None: + if self._closed: + return None + while not self._closed: + message = await self._websocket.receive() + message_type = message.get("type") + if message_type == "websocket.disconnect": + return None + + binary_audio = message.get("bytes") + if binary_audio is not None: + return binary_audio + + text_payload = message.get("text") + if text_payload is None: + continue + decoded = self._decode_text_frame(text_payload) + if decoded is None: + continue + return decoded + return None + + async def send_audio(self, audio_chunk: bytes) -> None: + if self._closed: + return + await self._websocket.send_bytes(audio_chunk) + + async def close(self) -> None: + if self._closed: + return + self._closed = True + if ( + self._websocket.application_state == WebSocketState.DISCONNECTED + or self._websocket.client_state == WebSocketState.DISCONNECTED + ): + return + try: + await self._websocket.close(code=1000) + except RuntimeError: + return + + @staticmethod + def _decode_text_frame(text_payload: str) -> bytes | None: + try: + payload = json.loads(text_payload) + except json.JSONDecodeError: + return None + + message_type = str(payload.get("type") or "").strip().lower() + if message_type in {"close", "disconnect", "hangup"}: + return None + if message_type != "audio": + return None + + encoded_audio = payload.get("pcm16") + if not isinstance(encoded_audio, str): + return None + return base64.b64decode(encoded_audio)