Module livekit.plugins.spitch

Spitch plugin for LiveKit Agents

Classes

class STT (*, language: str = 'en')

Expand source code

class STT(stt.STT):
    def __init__(self, *, language: str = "en") -> None:
        super().__init__(capabilities=stt.STTCapabilities(streaming=False, interim_results=False))

        self._opts = _STTOptions(language=language)
        self._client = AsyncSpitch()

    def update_options(self, language: str):
        self._opts.language = language or self._opts.language

    def _sanitize_options(self, *, language: str | None = None) -> _STTOptions:
        config = dataclasses.replace(self._opts)
        config.language = language or config.language
        return config

    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        try:
            config = self._sanitize_options(language=language or None)
            data = rtc.combine_audio_frames(buffer).to_wav_bytes()
            resp = await self._client.speech.transcribe(
                language=config.language,  # type: ignore
                content=data,
                timeout=httpx.Timeout(30, connect=conn_options.timeout),
            )

            return stt.SpeechEvent(
                type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                alternatives=[
                    stt.SpeechData(text=resp.text or "", language=config.language or ""),
                ],
            )
        except spitch.APITimeoutError as e:
            raise APITimeoutError() from e
        except spitch.APIStatusError as e:
            raise APIStatusError(e.message, status_code=e.status_code, body=e.body) from e
        except Exception as e:
            raise APIConnectionError() from e

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.stt.stt.STT
• abc.ABC
• EventEmitter
• typing.Generic

Methods

def update_options(self, language: str)

Expand source code

def update_options(self, language: str):
    self._opts.language = language or self._opts.language

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class TTS (*, language: str = 'en', voice: str = 'lina')

Expand source code

class TTS(tts.TTS):
    def __init__(self, *, language: str = "en", voice: str = "lina"):
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False), sample_rate=24_000, num_channels=1
        )

        self._opts = _TTSOptions(language=language, voice=voice)
        self._client = AsyncSpitch()

    def synthesize(
        self,
        text: str,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> ChunkedStream:
        return ChunkedStream(
            tts=self,
            input_text=text,
            conn_options=conn_options,
            opts=self._opts,
            client=self._client,
        )

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.tts.tts.TTS
• abc.ABC
• EventEmitter
• typing.Generic

Methods

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.spitch.tts.ChunkedStream

Expand source code

def synthesize(
    self,
    text: str,
    *,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> ChunkedStream:
    return ChunkedStream(
        tts=self,
        input_text=text,
        conn_options=conn_options,
        opts=self._opts,
        client=self._client,
    )

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once