Module livekit.plugins.rime
Rime plugin for LiveKit Agents
See https://docs.livekit.io/agents/integrations/tts/rime/ for more information.
Classes
class ChunkedStream (tts: TTS,
input_text: str,
conn_options: APIConnectOptions)-
Expand source code
class ChunkedStream(tts.ChunkedStream): """Synthesize using the chunked api endpoint""" def __init__(self, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None: self._sample_rate = tts.sample_rate self._opts = copy.deepcopy(tts._opts) self._base_url = tts._base_url self._total_timeout = tts._total_timeout super().__init__(tts=tts, input_text=input_text, conn_options=conn_options) self._tts: TTS = tts async def _run(self, output_emitter: tts.AudioEmitter) -> None: payload: dict[str, object] = { "speaker": self._opts.speaker, "text": self._input_text, "modelId": self._opts.model, **_model_params(self._opts), } format = "audio/pcm" payload["samplingRate"] = self._sample_rate if is_mist_model(self._opts.model) and self._opts.mist_options is not None: mist_opts = self._opts.mist_options if supports_reduce_latency(self._opts.model) and is_given(mist_opts.reduce_latency): payload["reduceLatency"] = mist_opts.reduce_latency try: async with self._tts._ensure_session().post( self._base_url, headers={ "accept": format, "Authorization": f"Bearer {self._tts._api_key}", "content-type": "application/json", }, json=payload, timeout=aiohttp.ClientTimeout( total=self._total_timeout, sock_connect=self._conn_options.timeout ), ) as resp: resp.raise_for_status() if not resp.content_type.startswith("audio"): content = await resp.text() logger.error("Rime returned non-audio data", extra={"lk.pii.data": content}) return output_emitter.initialize( request_id=utils.shortuuid(), sample_rate=self._sample_rate, num_channels=NUM_CHANNELS, mime_type=format, ) async for data, _ in resp.content.iter_chunks(): output_emitter.push(data) except asyncio.TimeoutError: raise APITimeoutError() from None except aiohttp.ClientResponseError as e: raise APIStatusError( message="Rime HTTP request failed", status_code=e.status, request_id=None, body=None, ) from None except Exception: raise APIConnectionError("Rime HTTP request failed") from NoneSynthesize using the chunked api endpoint
Ancestors
- livekit.agents.tts.tts.ChunkedStream
- abc.ABC
class TTS (*,
base_url: NotGivenOr[str] = NOT_GIVEN,
websocket_url: NotGivenOr[str] = NOT_GIVEN,
websocket_protocol: WebSocketProtocol = 'binary',
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
speaker: NotGivenOr[str] = NOT_GIVEN,
lang: TTSLangs | str = 'eng',
audio_format: NotGivenOr[RimeAudioFormat | str] = NOT_GIVEN,
repetition_penalty: NotGivenOr[float] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
max_tokens: NotGivenOr[int] = NOT_GIVEN,
time_scale_factor: NotGivenOr[float] = NOT_GIVEN,
speed_alpha: NotGivenOr[float] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
reduce_latency: NotGivenOr[bool] = NOT_GIVEN,
pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN,
phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
use_websocket: bool = False,
segment: NotGivenOr[str] = NOT_GIVEN,
tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
allow_custom_endpoint: bool = False)-
Expand source code
class TTS(tts.TTS): @overload def __init__( self, *, websocket_url: str, websocket_protocol: WebSocketProtocol = "binary", model: NotGivenOr[TTSModels | str] = NOT_GIVEN, speaker: NotGivenOr[str] = NOT_GIVEN, lang: TTSLangs | str = "eng", audio_format: RimeAudioFormat = DEFAULT_AUDIO_FORMAT, time_scale_factor: NotGivenOr[float] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN, phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN, allow_custom_endpoint: bool = False, ) -> None: ... @overload def __init__( self, *, base_url: NotGivenOr[str] = NOT_GIVEN, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, speaker: NotGivenOr[str] = NOT_GIVEN, lang: TTSLangs | str = "eng", repetition_penalty: NotGivenOr[float] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, max_tokens: NotGivenOr[int] = NOT_GIVEN, time_scale_factor: NotGivenOr[float] = NOT_GIVEN, speed_alpha: NotGivenOr[float] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, reduce_latency: NotGivenOr[bool] = NOT_GIVEN, pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN, phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, use_websocket: bool = False, segment: NotGivenOr[str] = NOT_GIVEN, tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN, allow_custom_endpoint: bool = False, ) -> None: ... def __init__( self, *, base_url: NotGivenOr[str] = NOT_GIVEN, websocket_url: NotGivenOr[str] = NOT_GIVEN, websocket_protocol: WebSocketProtocol = "binary", model: NotGivenOr[TTSModels | str] = NOT_GIVEN, speaker: NotGivenOr[str] = NOT_GIVEN, lang: TTSLangs | str = "eng", audio_format: NotGivenOr[RimeAudioFormat | str] = NOT_GIVEN, # Coda options repetition_penalty: NotGivenOr[float] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, max_tokens: NotGivenOr[int] = NOT_GIVEN, # Shared by Mist and Coda (HTTP and v1 WebSocket) time_scale_factor: NotGivenOr[float] = NOT_GIVEN, # Supported by HTTP and the legacy ws3 interface speed_alpha: NotGivenOr[float] = NOT_GIVEN, # Supported by all models sample_rate: NotGivenOr[int] = NOT_GIVEN, reduce_latency: NotGivenOr[bool] = NOT_GIVEN, pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN, phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN, api_key: NotGivenOr[str] = NOT_GIVEN, http_session: aiohttp.ClientSession | None = None, use_websocket: bool = False, segment: NotGivenOr[str] = NOT_GIVEN, tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN, allow_custom_endpoint: bool = False, ) -> None: websocket_v1_url = websocket_url if is_given(websocket_url) else None if websocket_v1_url is None and is_given(audio_format): raise ValueError("audio_format is only supported with the Rime v1 WebSocket interface") resolved_audio_format = ( validate_audio_format(audio_format) if is_given(audio_format) else DEFAULT_AUDIO_FORMAT ) if websocket_v1_url is not None: if is_given(base_url): raise ValueError("websocket_url cannot be used with base_url") if use_websocket: raise ValueError("websocket_url enables WebSocket streaming; omit use_websocket") if is_given(speed_alpha): raise ValueError( "speed_alpha belongs to the legacy Rime interfaces; use time_scale_factor" ) if any( is_given(value) for value in (repetition_penalty, temperature, top_p, max_tokens) ): raise ValueError( "generation controls are not supported by the Rime v1 WebSocket protocol" ) if is_given(reduce_latency) or is_given(segment): raise ValueError("websocket_url cannot be used with ws3-only options") use_websocket = True resolved_base_url = RIME_BASE_URL elif is_given(base_url): validate_endpoint_host(base_url, allow_custom_endpoint=allow_custom_endpoint) # Infer streaming mode from URL prefix; an explicit use_websocket=True still wins. use_websocket = use_websocket or base_url.startswith(("ws://", "wss://")) resolved_base_url = base_url else: resolved_base_url = RIME_WS_BASE_URL if use_websocket else RIME_BASE_URL if websocket_v1_url is not None: resolved_model = _resolve_websocket_model( websocket_v1_url, model, allow_custom_endpoint=allow_custom_endpoint, ) elif is_given(model): resolved_model = model else: resolved_model = MODEL_CODA _check_time_scale_factor_supported(resolved_model, time_scale_factor) if ( websocket_v1_url is not None and not is_mist_model(resolved_model) and any( is_given(value) for value in (pause_between_brackets, phonemize_between_brackets) ) ): raise ValueError("Mist options require a Mist model") resolved_sample_rate = ( sample_rate if is_given(sample_rate) else _default_sample_rate(resolved_model) ) super().__init__( capabilities=tts.TTSCapabilities( streaming=use_websocket, aligned_transcript=use_websocket and websocket_v1_url is None, ), sample_rate=resolved_sample_rate, num_channels=NUM_CHANNELS, ) resolved_api_key = api_key if is_given(api_key) else os.environ.get("RIME_API_KEY") if not resolved_api_key: raise ValueError( "Rime API key is required, either as argument or set RIME_API_KEY environmental variable" # noqa: E501 ) self._api_key = resolved_api_key self._allow_custom_endpoint = allow_custom_endpoint if not is_given(speaker): if is_mist_model(resolved_model): speaker = DefaultMistVoice elif resolved_model == MODEL_CODA: speaker = DefaultCodaVoice else: speaker = "astra" self._opts = _TTSOptions( model=resolved_model, speaker=speaker, language=lang, audio_format=resolved_audio_format, sample_rate=sample_rate, time_scale_factor=time_scale_factor, ) if resolved_model == MODEL_CODA: self._opts.coda_options = _CodaOptions( repetition_penalty=repetition_penalty, temperature=temperature, top_p=top_p, max_tokens=max_tokens, speed_alpha=speed_alpha, ) elif is_mist_model(resolved_model): self._opts.mist_options = _MistOptions( speed_alpha=speed_alpha, reduce_latency=reduce_latency, pause_between_brackets=pause_between_brackets, phonemize_between_brackets=phonemize_between_brackets, ) self._session = http_session self._base_url = resolved_base_url self._use_websocket = use_websocket self._segment = segment if is_given(segment) else "bySentence" self._sentence_tokenizer: tokenize.SentenceTokenizer | None = None if websocket_v1_url is None: self._sentence_tokenizer = ( tokenizer if is_given(tokenizer) else tokenize.blingfire.SentenceTokenizer() ) self._websocket_v1_adapter = ( WebSocketV1Adapter( websocket_v1_url=websocket_v1_url, websocket_protocol=websocket_protocol, api_key=self._api_key, ensure_session=self._ensure_session, sentence_tokenizer=tokenizer if is_given(tokenizer) else None, allow_custom_endpoint=allow_custom_endpoint, ) if websocket_v1_url is not None else None ) self._total_timeout = _timeout_for_model(resolved_model) self._streams: weakref.WeakSet[tts.SynthesizeStream] = weakref.WeakSet() self._legacy_websocket_adapter: LegacyWebSocketAdapter | None = None if self._use_websocket and self._websocket_v1_adapter is None: assert self._sentence_tokenizer is not None self._legacy_websocket_adapter = LegacyWebSocketAdapter( websocket_url=self._ws_url(), api_key=self._api_key, ensure_session=self._ensure_session, sentence_tokenizer=self._sentence_tokenizer, ) @property def model(self) -> str: return self._opts.model @property def provider(self) -> str: return "Rime" def _ensure_session(self) -> aiohttp.ClientSession: if not self._session: self._session = utils.http_context.http_session() return self._session def _ws_url(self) -> str: params: dict[str, object] = { "speaker": self._opts.speaker, "modelId": self._opts.model, "audioFormat": "pcm", "samplingRate": self.sample_rate, "segment": self._segment, **_model_params(self._opts), } encoded = { k: ("true" if v else "false") if isinstance(v, bool) else v for k, v in params.items() } return f"{self._base_url}/ws3?{urlencode(encoded)}" def prewarm(self) -> None: if self._websocket_v1_adapter is not None: self._websocket_v1_adapter.prewarm() elif self._legacy_websocket_adapter is not None: self._legacy_websocket_adapter.prewarm() def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.SynthesizeStream: if not self._use_websocket: raise RuntimeError( "Rime TTS streaming requires use_websocket=True at construction time" ) s: tts.SynthesizeStream if self._websocket_v1_adapter is not None: s = self._websocket_v1_adapter.stream( tts_instance=self, options=self._v1_synthesis_options(), conn_options=conn_options, ) else: assert self._legacy_websocket_adapter is not None s = self._legacy_websocket_adapter.stream( tts_instance=self, options=self._legacy_synthesis_options(), conn_options=conn_options, ) self._streams.add(s) return s def _v1_synthesis_options(self) -> V1SynthesisOptions: mist = self._opts.mist_options if is_mist_model(self._opts.model) else None return V1SynthesisOptions( model=self._opts.model, speaker=self._opts.speaker, language=(str(self._opts.language) if is_given(self._opts.language) else NOT_GIVEN), audio_format=self._opts.audio_format, sampling_rate=self.sample_rate, time_scale_factor=( self._opts.time_scale_factor if supports_time_scale_factor(self._opts.model) else NOT_GIVEN ), pause_between_brackets=(mist.pause_between_brackets if mist is not None else NOT_GIVEN), phonemize_between_brackets=( mist.phonemize_between_brackets if mist is not None else NOT_GIVEN ), ) def _legacy_synthesis_options(self) -> LegacySynthesisOptions: return LegacySynthesisOptions( model=self._opts.model, websocket_url=self._ws_url(), sample_rate=self.sample_rate, ) async def aclose(self) -> None: for s in list(self._streams): await s.aclose() self._streams.clear() if self._websocket_v1_adapter is not None: await self._websocket_v1_adapter.aclose() elif self._legacy_websocket_adapter is not None: await self._legacy_websocket_adapter.aclose() def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: if self._use_websocket: raise RuntimeError( "Rime TTS one-shot synthesize requires use_websocket=False at construction time" ) return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, speaker: NotGivenOr[str] = NOT_GIVEN, lang: NotGivenOr[TTSLangs | str] = NOT_GIVEN, # Coda parameters repetition_penalty: NotGivenOr[float] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, max_tokens: NotGivenOr[int] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, audio_format: NotGivenOr[RimeAudioFormat | str] = NOT_GIVEN, time_scale_factor: NotGivenOr[float] = NOT_GIVEN, # Mistv2 parameters speed_alpha: NotGivenOr[float] = NOT_GIVEN, reduce_latency: NotGivenOr[bool] = NOT_GIVEN, pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN, phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, websocket_url: NotGivenOr[str] = NOT_GIVEN, ) -> None: updated_audio_format: RimeAudioFormat | None = None if self._websocket_v1_adapter is not None: if is_given(audio_format): updated_audio_format = validate_audio_format(audio_format) if is_given(model) and not is_given(websocket_url): raise ValueError( "model can only be updated together with websocket_url for Rime v1" ) if is_given(base_url): raise ValueError("use websocket_url to update a Rime v1 endpoint") if is_given(speed_alpha) or is_given(reduce_latency): raise ValueError("Rime v1 cannot be updated with ws3-only options") if any( is_given(value) for value in (repetition_penalty, temperature, top_p, max_tokens) ): raise ValueError("Rime v1 does not support generation controls") effective_model = self._opts.model if is_given(websocket_url): effective_model = _resolve_websocket_model( websocket_url, model, allow_custom_endpoint=self._allow_custom_endpoint, current_model=self._opts.model, ) if not is_mist_model(effective_model) and any( is_given(value) for value in (pause_between_brackets, phonemize_between_brackets) ): raise ValueError("Mist options require a Mist model") elif is_given(websocket_url): raise ValueError("websocket_url can only update a TTS constructed with websocket_url") else: if is_given(audio_format): raise ValueError( "audio_format is only supported with the Rime v1 WebSocket interface" ) effective_model = model if is_given(model) else self._opts.model if is_given(base_url): validate_endpoint_host(base_url, allow_custom_endpoint=self._allow_custom_endpoint) _check_time_scale_factor_supported(effective_model, time_scale_factor) # Each WS3 pool is bound to one URL. Replace it when URL options change. prev_ws_url = self._ws_url() if self._legacy_websocket_adapter is not None else None if is_given(websocket_url): assert self._websocket_v1_adapter is not None self._websocket_v1_adapter.update_endpoint( websocket_url, model_changed=effective_model != self._opts.model ) self._opts.model = effective_model self._total_timeout = _timeout_for_model(effective_model) if effective_model == MODEL_CODA and self._opts.coda_options is None: self._opts.coda_options = _CodaOptions() elif is_mist_model(effective_model) and self._opts.mist_options is None: self._opts.mist_options = _MistOptions() if is_given(base_url): self._base_url = base_url if is_given(model): self._opts.model = model self._total_timeout = _timeout_for_model(model) if model == MODEL_CODA and self._opts.coda_options is None: self._opts.coda_options = _CodaOptions() elif is_mist_model(model) and self._opts.mist_options is None: self._opts.mist_options = _MistOptions() if is_given(speaker): self._opts.speaker = speaker if is_given(lang): self._opts.language = lang if is_given(sample_rate): self._opts.sample_rate = sample_rate if updated_audio_format is not None: self._opts.audio_format = updated_audio_format if is_given(time_scale_factor): self._opts.time_scale_factor = time_scale_factor if self._opts.model == MODEL_CODA and self._opts.coda_options is not None: if is_given(repetition_penalty): self._opts.coda_options.repetition_penalty = repetition_penalty if is_given(temperature): self._opts.coda_options.temperature = temperature if is_given(top_p): self._opts.coda_options.top_p = top_p if is_given(max_tokens): self._opts.coda_options.max_tokens = max_tokens if is_given(speed_alpha): self._opts.coda_options.speed_alpha = speed_alpha elif is_mist_model(self._opts.model) and self._opts.mist_options is not None: if is_given(speed_alpha): self._opts.mist_options.speed_alpha = speed_alpha if is_given(reduce_latency): self._opts.mist_options.reduce_latency = reduce_latency if is_given(pause_between_brackets): self._opts.mist_options.pause_between_brackets = pause_between_brackets if is_given(phonemize_between_brackets): self._opts.mist_options.phonemize_between_brackets = phonemize_between_brackets requested_sample_rate = self._opts.sample_rate self._sample_rate = ( requested_sample_rate if is_given(requested_sample_rate) else _default_sample_rate(self._opts.model) ) if prev_ws_url is not None and self._ws_url() != prev_ws_url: assert self._legacy_websocket_adapter is not None self._legacy_websocket_adapter.update_endpoint(self._ws_url())Helper class that provides a standard way to create an ABC using inheritance.
Ancestors
- livekit.agents.tts.tts.TTS
- abc.ABC
- EventEmitter
- typing.Generic
Instance variables
prop model : str-
Expand source code
@property def model(self) -> str: return self._opts.modelGet the model name/identifier for this TTS instance.
Returns
The model name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their model information.
prop provider : str-
Expand source code
@property def provider(self) -> str: return "Rime"Get the provider name/identifier for this TTS instance.
Returns
The provider name if available, "unknown" otherwise.
Note
Plugins should override this property to provide their provider information.
Methods
async def aclose(self) ‑> None-
Expand source code
async def aclose(self) -> None: for s in list(self._streams): await s.aclose() self._streams.clear() if self._websocket_v1_adapter is not None: await self._websocket_v1_adapter.aclose() elif self._legacy_websocket_adapter is not None: await self._legacy_websocket_adapter.aclose() def prewarm(self) ‑> None-
Expand source code
def prewarm(self) -> None: if self._websocket_v1_adapter is not None: self._websocket_v1_adapter.prewarm() elif self._legacy_websocket_adapter is not None: self._legacy_websocket_adapter.prewarm()Pre-warm connection to the TTS service
def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.agents.tts.tts.SynthesizeStream-
Expand source code
def stream( self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> tts.SynthesizeStream: if not self._use_websocket: raise RuntimeError( "Rime TTS streaming requires use_websocket=True at construction time" ) s: tts.SynthesizeStream if self._websocket_v1_adapter is not None: s = self._websocket_v1_adapter.stream( tts_instance=self, options=self._v1_synthesis_options(), conn_options=conn_options, ) else: assert self._legacy_websocket_adapter is not None s = self._legacy_websocket_adapter.stream( tts_instance=self, options=self._legacy_synthesis_options(), conn_options=conn_options, ) self._streams.add(s) return s def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.rime.tts.ChunkedStream-
Expand source code
def synthesize( self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS ) -> ChunkedStream: if self._use_websocket: raise RuntimeError( "Rime TTS one-shot synthesize requires use_websocket=False at construction time" ) return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def update_options(self,
*,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
speaker: NotGivenOr[str] = NOT_GIVEN,
lang: NotGivenOr[TTSLangs | str] = NOT_GIVEN,
repetition_penalty: NotGivenOr[float] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
max_tokens: NotGivenOr[int] = NOT_GIVEN,
sample_rate: NotGivenOr[int] = NOT_GIVEN,
audio_format: NotGivenOr[RimeAudioFormat | str] = NOT_GIVEN,
time_scale_factor: NotGivenOr[float] = NOT_GIVEN,
speed_alpha: NotGivenOr[float] = NOT_GIVEN,
reduce_latency: NotGivenOr[bool] = NOT_GIVEN,
pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN,
phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
websocket_url: NotGivenOr[str] = NOT_GIVEN) ‑> None-
Expand source code
def update_options( self, *, model: NotGivenOr[TTSModels | str] = NOT_GIVEN, speaker: NotGivenOr[str] = NOT_GIVEN, lang: NotGivenOr[TTSLangs | str] = NOT_GIVEN, # Coda parameters repetition_penalty: NotGivenOr[float] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, top_p: NotGivenOr[float] = NOT_GIVEN, max_tokens: NotGivenOr[int] = NOT_GIVEN, sample_rate: NotGivenOr[int] = NOT_GIVEN, audio_format: NotGivenOr[RimeAudioFormat | str] = NOT_GIVEN, time_scale_factor: NotGivenOr[float] = NOT_GIVEN, # Mistv2 parameters speed_alpha: NotGivenOr[float] = NOT_GIVEN, reduce_latency: NotGivenOr[bool] = NOT_GIVEN, pause_between_brackets: NotGivenOr[bool] = NOT_GIVEN, phonemize_between_brackets: NotGivenOr[bool] = NOT_GIVEN, base_url: NotGivenOr[str] = NOT_GIVEN, websocket_url: NotGivenOr[str] = NOT_GIVEN, ) -> None: updated_audio_format: RimeAudioFormat | None = None if self._websocket_v1_adapter is not None: if is_given(audio_format): updated_audio_format = validate_audio_format(audio_format) if is_given(model) and not is_given(websocket_url): raise ValueError( "model can only be updated together with websocket_url for Rime v1" ) if is_given(base_url): raise ValueError("use websocket_url to update a Rime v1 endpoint") if is_given(speed_alpha) or is_given(reduce_latency): raise ValueError("Rime v1 cannot be updated with ws3-only options") if any( is_given(value) for value in (repetition_penalty, temperature, top_p, max_tokens) ): raise ValueError("Rime v1 does not support generation controls") effective_model = self._opts.model if is_given(websocket_url): effective_model = _resolve_websocket_model( websocket_url, model, allow_custom_endpoint=self._allow_custom_endpoint, current_model=self._opts.model, ) if not is_mist_model(effective_model) and any( is_given(value) for value in (pause_between_brackets, phonemize_between_brackets) ): raise ValueError("Mist options require a Mist model") elif is_given(websocket_url): raise ValueError("websocket_url can only update a TTS constructed with websocket_url") else: if is_given(audio_format): raise ValueError( "audio_format is only supported with the Rime v1 WebSocket interface" ) effective_model = model if is_given(model) else self._opts.model if is_given(base_url): validate_endpoint_host(base_url, allow_custom_endpoint=self._allow_custom_endpoint) _check_time_scale_factor_supported(effective_model, time_scale_factor) # Each WS3 pool is bound to one URL. Replace it when URL options change. prev_ws_url = self._ws_url() if self._legacy_websocket_adapter is not None else None if is_given(websocket_url): assert self._websocket_v1_adapter is not None self._websocket_v1_adapter.update_endpoint( websocket_url, model_changed=effective_model != self._opts.model ) self._opts.model = effective_model self._total_timeout = _timeout_for_model(effective_model) if effective_model == MODEL_CODA and self._opts.coda_options is None: self._opts.coda_options = _CodaOptions() elif is_mist_model(effective_model) and self._opts.mist_options is None: self._opts.mist_options = _MistOptions() if is_given(base_url): self._base_url = base_url if is_given(model): self._opts.model = model self._total_timeout = _timeout_for_model(model) if model == MODEL_CODA and self._opts.coda_options is None: self._opts.coda_options = _CodaOptions() elif is_mist_model(model) and self._opts.mist_options is None: self._opts.mist_options = _MistOptions() if is_given(speaker): self._opts.speaker = speaker if is_given(lang): self._opts.language = lang if is_given(sample_rate): self._opts.sample_rate = sample_rate if updated_audio_format is not None: self._opts.audio_format = updated_audio_format if is_given(time_scale_factor): self._opts.time_scale_factor = time_scale_factor if self._opts.model == MODEL_CODA and self._opts.coda_options is not None: if is_given(repetition_penalty): self._opts.coda_options.repetition_penalty = repetition_penalty if is_given(temperature): self._opts.coda_options.temperature = temperature if is_given(top_p): self._opts.coda_options.top_p = top_p if is_given(max_tokens): self._opts.coda_options.max_tokens = max_tokens if is_given(speed_alpha): self._opts.coda_options.speed_alpha = speed_alpha elif is_mist_model(self._opts.model) and self._opts.mist_options is not None: if is_given(speed_alpha): self._opts.mist_options.speed_alpha = speed_alpha if is_given(reduce_latency): self._opts.mist_options.reduce_latency = reduce_latency if is_given(pause_between_brackets): self._opts.mist_options.pause_between_brackets = pause_between_brackets if is_given(phonemize_between_brackets): self._opts.mist_options.phonemize_between_brackets = phonemize_between_brackets requested_sample_rate = self._opts.sample_rate self._sample_rate = ( requested_sample_rate if is_given(requested_sample_rate) else _default_sample_rate(self._opts.model) ) if prev_ws_url is not None and self._ws_url() != prev_ws_url: assert self._legacy_websocket_adapter is not None self._legacy_websocket_adapter.update_endpoint(self._ws_url())
Inherited members