diff --git a/.github/workflows/coverage.yaml b/.github/workflows/coverage.yaml index d65841a7d..26d03861b 100644 --- a/.github/workflows/coverage.yaml +++ b/.github/workflows/coverage.yaml @@ -37,10 +37,12 @@ jobs: uv sync --group dev \ --extra anthropic \ --extra aws \ + --extra deepgram \ --extra google \ --extra langchain \ --extra livekit \ --extra piper \ + --extra sagemaker \ --extra tracing \ --extra websocket diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index a36a2fbd0..b22d502c4 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -41,10 +41,12 @@ jobs: uv sync --group dev \ --extra anthropic \ --extra aws \ + --extra deepgram \ --extra google \ --extra langchain \ --extra livekit \ --extra piper \ + --extra sagemaker \ --extra tracing \ --extra websocket diff --git a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py index 05c92e7e2..e8094183a 100644 --- a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py @@ -7,6 +7,7 @@ import asyncio import os +from deepgram import LiveOptions from dotenv import load_dotenv from loguru import logger @@ -106,12 +107,25 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) + # NOTE: after this change, the bot will only respond if you speak Spanish await asyncio.sleep(10) - logger.info("Updating Deepgram SageMaker STT settings: language=es") + logger.info("Updating Deepgram SageMaker STT settings: language=es, punctuate=False") await task.queue_frame( - STTUpdateSettingsFrame(delta=DeepgramSageMakerSTTSettings(language=Language.ES)) + STTUpdateSettingsFrame( + delta=DeepgramSageMakerSTTSettings( + language=Language.ES, + live_options=LiveOptions(punctuate=False), + ) + ) ) + # Old-style dict update (for backward-compat testing): + # await asyncio.sleep(10) + # logger.info("Updating Deepgram SageMaker STT settings via dict: punctuate=False, filler_words=True") + # await task.queue_frame( + # STTUpdateSettingsFrame(settings={"punctuate": False, "filler_words": True}) + # ) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/examples/foundational/55a-update-settings-deepgram-stt.py b/examples/foundational/55a-update-settings-deepgram-stt.py index 39dde69e9..8808f6f4c 100644 --- a/examples/foundational/55a-update-settings-deepgram-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-stt.py @@ -7,6 +7,7 @@ import asyncio import os +from deepgram import LiveOptions from dotenv import load_dotenv from loguru import logger @@ -100,12 +101,25 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) + # NOTE: after this change, the bot will only respond if you speak Spanish await asyncio.sleep(10) - logger.info("Updating Deepgram STT settings: language=es") + logger.info("Updating Deepgram STT settings: language=es, punctuate=False") await task.queue_frame( - STTUpdateSettingsFrame(delta=DeepgramSTTSettings(language=Language.ES)) + STTUpdateSettingsFrame( + delta=DeepgramSTTSettings( + language=Language.ES, + live_options=LiveOptions(punctuate=False), + ) + ) ) + # Old-style dict update (for backward-compat testing): + # await asyncio.sleep(10) + # logger.info("Updating Deepgram STT settings via dict: punctuate=False, filler_words=True") + # await task.queue_frame( + # STTUpdateSettingsFrame(settings={"punctuate": False, "filler_words": True}) + # ) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index aa5c2ce8d..497d6aae1 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -6,8 +6,9 @@ """Deepgram speech-to-text service implementation.""" +import inspect from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, Dict, Optional +from typing import Any, AsyncGenerator, Dict, Mapping, Optional, Type from loguru import logger @@ -24,7 +25,7 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven, is_given +from pipecat.services.settings import _S, NOT_GIVEN, STTSettings, _NotGiven, is_given from pipecat.services.stt_latency import DEEPGRAM_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -48,15 +49,166 @@ except ModuleNotFoundError as e: @dataclass -class DeepgramSTTSettings(STTSettings): - """Settings for the Deepgram STT service. +class _DeepgramSTTSettingsBase(STTSettings): + """Base settings for Deepgram STT services that use ``LiveOptions``. + + Shared by ``DeepgramSTTSettings`` and ``DeepgramSageMakerSTTSettings``. + Not intended for other Deepgram services that don't use ``LiveOptions``. + + Wraps the Deepgram SDK's ``LiveOptions`` in a single ``live_options`` + field and provides delta-merge semantics: when used as a delta (e.g. + via ``STTUpdateSettingsFrame``), only the non-None fields of + ``live_options`` are merged into the stored options rather than + replacing them wholesale. + + ``model`` and ``language`` are kept in sync bidirectionally between + the top-level settings fields and the nested ``live_options``. Parameters: - live_options: Deepgram ``LiveOptions`` for detailed configuration. + live_options: Deepgram ``LiveOptions`` for STT configuration. + In delta mode only its non-None fields are merged into the + stored options. """ live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + # Valid LiveOptions __init__ parameter names (cached at class level). + _live_options_params: set[str] | None = field(default=None, init=False, repr=False) + + @classmethod + def _get_live_options_params(cls) -> set[str]: + """Return the set of valid ``LiveOptions.__init__`` parameter names.""" + if cls._live_options_params is None: + cls._live_options_params = set(inspect.signature(LiveOptions.__init__).parameters) - { + "self" + } + return cls._live_options_params + + def _merge_live_options_delta(self, delta: LiveOptions) -> Dict[str, Any]: + """Merge a ``LiveOptions`` delta into the stored ``live_options``. + + Non-None fields from *delta* overwrite corresponding fields in the + stored ``LiveOptions``. ``model`` and ``language`` are synced to + the top-level settings fields when they change. + + Args: + delta: A ``LiveOptions`` whose non-None fields are the desired + overrides. + + Returns: + Dict mapping each changed key to its **previous** value (same + contract as ``apply_update``). + """ + old_dict = self.live_options.to_dict() # type: ignore[union-attr] + delta_dict = delta.to_dict() + + # Deepgram SDK bug: model initialised to the *string* "None". + if delta_dict.get("model") == "None": + del delta_dict["model"] + + if not delta_dict: + return {} + + merged = {**old_dict, **delta_dict} + self.live_options = LiveOptions(**merged) + + # Track what changed. + changed: Dict[str, Any] = {} + for key in delta_dict: + old_val = old_dict.get(key, NOT_GIVEN) + if old_val != delta_dict[key]: + changed[key] = old_val + + # Sync model/language from live_options delta to top-level fields. + if "model" in delta_dict and delta_dict["model"] != self.model: + changed.setdefault("model", self.model) + self.model = delta_dict["model"] + if "language" in delta_dict and delta_dict["language"] != self.language: + changed.setdefault("language", self.language) + self.language = delta_dict["language"] + + return changed + + def apply_update(self: _S, delta: _S) -> Dict[str, Any]: + """Merge a delta into this store, with delta-merge for ``live_options``. + + ``live_options`` is merged field-by-field via + ``_merge_live_options_delta`` rather than being replaced wholesale. + + ``model`` and ``language`` are kept in sync bidirectionally between + the top-level settings fields and ``live_options``. + """ + # Pull live_options out of the delta so super() doesn't replace it. + delta_lo = getattr(delta, "live_options", NOT_GIVEN) + if is_given(delta_lo): + delta.live_options = NOT_GIVEN # type: ignore[assignment] + + # Let the base class handle model, language, extra. + changed = super().apply_update(delta) + + # Sync top-level model/language changes into stored live_options. + if "model" in changed: + self.live_options.model = self.model # type: ignore[union-attr] + if "language" in changed: + self.live_options.language = self.language # type: ignore[union-attr] + + # Merge live_options delta. Top-level model/language take precedence + # over conflicting values in live_options, so write them into the + # delta before merging. + if is_given(delta_lo): + if "model" in changed: + delta_lo.model = self.model + if "language" in changed: + delta_lo.language = self.language + + for key, old_val in self._merge_live_options_delta(delta_lo).items(): + changed.setdefault(key, old_val) + + return changed + + @classmethod + def from_mapping(cls: Type[_S], settings: Mapping[str, Any]) -> _S: + """Build a delta from a plain dict, routing LiveOptions keys correctly. + + Keys that are valid ``LiveOptions.__init__`` parameters (and not + top-level ``STTSettings`` fields like ``model`` / ``language``) are + collected into a ``LiveOptions`` object. ``model`` and ``language`` + are routed to the top-level settings fields. Truly unknown keys go + to ``extra``. + """ + lo_params = cls._get_live_options_params() + stt_field_names = {"model", "language"} + + kwargs: Dict[str, Any] = {} + lo_kwargs: Dict[str, Any] = {} + extra: Dict[str, Any] = {} + + for key, value in settings.items(): + canonical = cls._aliases.get(key, key) + if canonical in stt_field_names: + kwargs[canonical] = value + elif canonical in lo_params: + lo_kwargs[canonical] = value + else: + extra[key] = value + + if lo_kwargs: + kwargs["live_options"] = LiveOptions(**lo_kwargs) + + instance = cls(**kwargs) + instance.extra = extra + return instance + + +@dataclass +class DeepgramSTTSettings(_DeepgramSTTSettingsBase): + """Settings for the Deepgram STT service. + + See ``_DeepgramSTTSettingsBase`` for full documentation. + """ + + pass + class DeepgramSTTService(STTService): """Deepgram speech-to-text service. @@ -102,7 +254,9 @@ class DeepgramSTTService(STTService): base_url: Custom Deepgram API base URL. sample_rate: Audio sample rate. If None, uses default or live_options value. - live_options: Deepgram LiveOptions for detailed configuration. + live_options: Deepgram LiveOptions configuration. Treated as a + delta from a set of sensible defaults — only the fields you + set are overridden; all others keep their default values. addons: Additional Deepgram features to enable. should_interrupt: Determine whether the bot should be interrupted when Deepgram VAD events are enabled and the system detects that the user is speaking. @@ -141,34 +295,25 @@ class DeepgramSTTService(STTService): vad_events=False, ) - merged_options = default_options.to_dict() + settings = DeepgramSTTSettings( + model=default_options.model, + language=default_options.language, + live_options=default_options, + ) if live_options: - default_model = default_options.model - merged_options.update(live_options.to_dict()) - # NOTE(aleix): Fixes an in deepgram-sdk where `model` is initialized - # to the string "None" instead of the value `None`. - if "model" in merged_options and merged_options["model"] == "None": - merged_options["model"] = default_model + settings._merge_live_options_delta(live_options) - if "language" in merged_options and isinstance(merged_options["language"], Language): - merged_options["language"] = merged_options["language"].value - - merged_live_options = LiveOptions(**merged_options) super().__init__( sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, - settings=DeepgramSTTSettings( - model=merged_options.get("model"), - language=merged_options.get("language"), - live_options=merged_live_options, - ), + settings=settings, **kwargs, ) self._addons = addons self._should_interrupt = should_interrupt - if merged_live_options.vad_events: + if self._settings.live_options.vad_events: import warnings with warnings.catch_warnings(): @@ -210,43 +355,12 @@ class DeepgramSTTService(STTService): return True async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: - """Apply a settings delta, keeping ``live_options`` in sync. - - Top-level ``model`` and ``language`` are the source of truth. When - they are given in *delta* their values are propagated into - ``live_options``. When only ``live_options`` is given, its ``model`` - and ``language`` are propagated *up* to the top-level fields. - - Any change triggers a WebSocket reconnect. - """ - # Determine which top-level fields are explicitly provided. - model_given = isinstance(delta, DeepgramSTTSettings) and is_given( - getattr(delta, "model", NOT_GIVEN) - ) - language_given = isinstance(delta, DeepgramSTTSettings) and is_given( - getattr(delta, "language", NOT_GIVEN) - ) - + """Apply a settings delta and reconnect if anything changed.""" changed = await super()._update_settings(delta) if not changed: return changed - # --- Sync model -------------------------------------------------- - if model_given: - # Top-level model wins → push into live_options. - self._settings.live_options.model = self._settings.model - elif "live_options" in changed and self._settings.live_options.model is not None: - # Only live_options was given → pull model up. - self._settings.model = self._settings.live_options.model - self._sync_model_name_to_metrics() - - # --- Sync language ----------------------------------------------- - if language_given: - self._settings.live_options.language = self._settings.language - elif "live_options" in changed and self._settings.live_options.language is not None: - self._settings.language = self._settings.live_options.language - await self._disconnect() await self._connect() @@ -259,7 +373,6 @@ class DeepgramSTTService(STTService): frame: The start frame containing initialization parameters. """ await super().start(frame) - self._settings.live_options.sample_rate = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -312,9 +425,11 @@ class DeepgramSTTService(STTService): self._on_utterance_end, ) - if not await self._connection.start( - options=self._settings.live_options, addons=self._addons - ): + live_options = LiveOptions( + **{**self._settings.live_options.to_dict(), "sample_rate": self.sample_rate} + ) + + if not await self._connection.start(options=live_options, addons=self._addons): await self.push_error(error_msg=f"Unable to connect to Deepgram") else: headers = { diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index bc5eebe37..ba4b7dfda 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -14,8 +14,8 @@ languages, and various Deepgram features. import asyncio import json -from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, Optional +from dataclasses import dataclass +from typing import Any, AsyncGenerator, Dict, Optional from loguru import logger @@ -32,7 +32,8 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven, is_given +from pipecat.services.deepgram.stt import _DeepgramSTTSettingsBase +from pipecat.services.settings import STTSettings from pipecat.services.stt_latency import DEEPGRAM_SAGEMAKER_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -50,14 +51,13 @@ except ModuleNotFoundError as e: @dataclass -class DeepgramSageMakerSTTSettings(STTSettings): +class DeepgramSageMakerSTTSettings(_DeepgramSTTSettingsBase): """Settings for the Deepgram SageMaker STT service. - Parameters: - live_options: Deepgram ``LiveOptions`` for detailed configuration. + See ``_DeepgramSTTSettingsBase`` for full documentation. """ - live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + pass class DeepgramSageMakerSTTService(STTService): @@ -108,15 +108,15 @@ class DeepgramSageMakerSTTService(STTService): region: AWS region where the endpoint is deployed (e.g., "us-east-2"). sample_rate: Audio sample rate in Hz. If None, uses value from live_options or defaults to the value from StartFrame. - live_options: Deepgram LiveOptions for detailed configuration. If None, - uses sensible defaults (nova-3 model, English, interim results enabled). + live_options: Deepgram LiveOptions configuration. Treated as a + delta from a set of sensible defaults — only the fields you + set are overridden; all others keep their default values. ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark **kwargs: Additional arguments passed to the parent STTService. """ sample_rate = sample_rate or (live_options.sample_rate if live_options else None) - # Create default options similar to DeepgramSTTService default_options = LiveOptions( encoding="linear16", language=Language.EN, @@ -126,28 +126,18 @@ class DeepgramSageMakerSTTService(STTService): punctuate=True, ) - # Merge with provided options - merged_options = default_options.to_dict() + settings = DeepgramSageMakerSTTSettings( + model=default_options.model, + language=default_options.language, + live_options=default_options, + ) if live_options: - default_model = default_options.model - merged_options.update(live_options.to_dict()) - # Handle the "None" string bug from deepgram-sdk - if "model" in merged_options and merged_options["model"] == "None": - merged_options["model"] = default_model + settings._merge_live_options_delta(live_options) - # Convert Language enum to string if needed - if "language" in merged_options and isinstance(merged_options["language"], Language): - merged_options["language"] = merged_options["language"].value - - merged_live_options = LiveOptions(**merged_options) super().__init__( sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, - settings=DeepgramSageMakerSTTSettings( - model=merged_options.get("model"), - language=merged_options.get("language"), - live_options=merged_live_options, - ), + settings=settings, **kwargs, ) @@ -167,46 +157,12 @@ class DeepgramSageMakerSTTService(STTService): return True async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: - """Apply a settings delta, keeping ``live_options`` in sync. - - Top-level ``model`` and ``language`` are the source of truth. When - they are given in *delta* their values are propagated into - ``live_options``. When only ``live_options`` is given, its ``model`` - and ``language`` are propagated *up* to the top-level fields. - - Any change triggers a reconnect. - """ - # Determine which top-level fields are explicitly provided. - model_given = isinstance(delta, DeepgramSageMakerSTTSettings) and is_given( - getattr(delta, "model", NOT_GIVEN) - ) - language_given = isinstance(delta, DeepgramSageMakerSTTSettings) and is_given( - getattr(delta, "language", NOT_GIVEN) - ) - + """Apply a settings delta and warn about unhandled changes.""" changed = await super()._update_settings(delta) if not changed: return changed - # --- Sync model -------------------------------------------------- - if model_given: - # Top-level model wins → push into live_options. - self._settings.live_options.model = self._settings.model - elif "live_options" in changed and self._settings.live_options.model is not None: - # Only live_options was given → pull model up. - self._settings.model = self._settings.live_options.model - self._sync_model_name_to_metrics() - - # --- Sync language ----------------------------------------------- - if language_given: - lang = self._settings.language - if isinstance(lang, Language): - lang = lang.value - self._settings.live_options.language = lang - elif "live_options" in changed and self._settings.live_options.language is not None: - self._settings.language = self._settings.live_options.language - # TODO: someday we could reconnect here to apply updated settings. # Code might look something like the below: # await self._disconnect() @@ -223,7 +179,6 @@ class DeepgramSageMakerSTTService(STTService): frame: The start frame containing initialization parameters. """ await super().start(frame) - self._settings.live_options.sample_rate = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -269,12 +224,13 @@ class DeepgramSageMakerSTTService(STTService): """ logger.debug("Connecting to Deepgram on SageMaker...") - # Update sample rate in live_options - self._settings.live_options.sample_rate = self.sample_rate + live_options = LiveOptions( + **{**self._settings.live_options.to_dict(), "sample_rate": self.sample_rate} + ) # Build query string from live_options, converting booleans to strings query_params = {} - for key, value in self._settings.live_options.to_dict().items(): + for key, value in live_options.to_dict().items(): if value is not None: # Convert boolean values to lowercase strings for Deepgram API if isinstance(value, bool): diff --git a/tests/test_settings.py b/tests/test_settings.py index 85f89987c..47cb6e4cf 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -7,7 +7,10 @@ """Tests for the typed settings infrastructure in pipecat.services.settings.""" import pytest +from deepgram import LiveOptions +from pipecat.services.deepgram.stt import DeepgramSTTSettings +from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTSettings from pipecat.services.settings import ( NOT_GIVEN, LLMSettings, @@ -311,3 +314,219 @@ class TestRoundtrip: assert changed["model"] == "gpt-4o" assert current.model == "gpt-4o-mini" assert current.temperature == 0.9 + + +# --------------------------------------------------------------------------- +# DeepgramSTTSettings: live_options delta merge +# --------------------------------------------------------------------------- + + +class TestDeepgramSTTSettingsApplyUpdate: + def _make_store(self, **lo_kwargs) -> DeepgramSTTSettings: + """Helper to build a store-mode DeepgramSTTSettings.""" + defaults = dict( + encoding="linear16", + channels=1, + interim_results=True, + smart_format=False, + punctuate=True, + profanity_filter=True, + vad_events=False, + ) + defaults.update(lo_kwargs) + s = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions(**defaults), + ) + return s + + def test_apply_update_merges_live_options_as_delta(self): + """Only the given fields in the delta LiveOptions are merged.""" + current = self._make_store() + assert current.live_options.punctuate is True + + delta = DeepgramSTTSettings(live_options=LiveOptions(punctuate=False)) + changed = current.apply_update(delta) + + assert current.live_options.punctuate is False + assert "punctuate" in changed + # Other fields are untouched + assert current.live_options.encoding == "linear16" + assert current.live_options.channels == 1 + + def test_apply_update_syncs_model_from_live_options_to_top_level(self): + """model inside live_options delta should sync to top-level model.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings(live_options=LiveOptions(model="nova-2")) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert "model" in changed + + def test_apply_update_syncs_language_from_live_options_to_top_level(self): + """language inside live_options delta should sync to top-level language.""" + current = self._make_store() + assert current.language == "en" + + delta = DeepgramSTTSettings(live_options=LiveOptions(language="es")) + changed = current.apply_update(delta) + + assert current.language == "es" + assert "language" in changed + + def test_apply_update_syncs_top_level_model_into_live_options(self): + """Top-level model change should propagate into stored live_options.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings(model="nova-2") + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + def test_apply_update_syncs_top_level_language_into_live_options(self): + """Top-level language change should propagate into stored live_options.""" + current = self._make_store() + + delta = DeepgramSTTSettings(language="fr") + changed = current.apply_update(delta) + + assert current.language == "fr" + assert current.live_options.language == "fr" + assert "language" in changed + + def test_apply_update_no_change(self): + """Delta with same values should report no changes.""" + current = self._make_store() + delta = DeepgramSTTSettings(live_options=LiveOptions(punctuate=True)) + changed = current.apply_update(delta) + assert changed == {} + + def test_apply_update_top_level_model_takes_precedence_over_live_options(self): + """When both top-level model and live_options.model are set, top-level wins.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings( + model="nova-2", + live_options=LiveOptions(model="nova-3"), + ) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + def test_apply_update_top_level_language_takes_precedence_over_live_options(self): + """When both top-level language and live_options.language are set, top-level wins.""" + current = self._make_store() + assert current.language == "en" + + delta = DeepgramSTTSettings( + language="fr", + live_options=LiveOptions(language="es"), + ) + changed = current.apply_update(delta) + + assert current.language == "fr" + assert current.live_options.language == "fr" + assert "language" in changed + + +class TestDeepgramSTTSettingsFromMapping: + def test_routes_live_options_kwargs(self): + """LiveOptions-valid keys should be collected into live_options.""" + delta = DeepgramSTTSettings.from_mapping({"punctuate": False, "filler_words": True}) + assert is_given(delta.live_options) + assert delta.live_options.punctuate is False + assert delta.live_options.filler_words is True + + def test_routes_model_and_language_to_top_level(self): + """model and language should be top-level fields, not in live_options.""" + delta = DeepgramSTTSettings.from_mapping({"model": "nova-2", "language": "es"}) + assert delta.model == "nova-2" + assert delta.language == "es" + assert not is_given(delta.live_options) + + def test_unknown_keys_go_to_extra(self): + """Keys that aren't LiveOptions params or STT fields go to extra.""" + delta = DeepgramSTTSettings.from_mapping({"unknown_param": 42}) + assert delta.extra == {"unknown_param": 42} + assert not is_given(delta.live_options) + + def test_mixed_keys(self): + """model + LiveOptions keys + unknown keys are routed correctly.""" + delta = DeepgramSTTSettings.from_mapping( + {"model": "nova-2", "punctuate": False, "unknown": "val"} + ) + assert delta.model == "nova-2" + assert delta.live_options.punctuate is False + assert delta.extra == {"unknown": "val"} + + def test_roundtrip_from_mapping_apply_update(self): + """Simulate dict-style update: from_mapping -> apply_update.""" + current = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions( + encoding="linear16", + channels=1, + interim_results=True, + punctuate=True, + profanity_filter=True, + vad_events=False, + ), + ) + + raw = {"punctuate": False, "filler_words": True} + delta = DeepgramSTTSettings.from_mapping(raw) + changed = current.apply_update(delta) + + assert current.live_options.punctuate is False + assert current.live_options.filler_words is True + # Unchanged fields stay put + assert current.live_options.encoding == "linear16" + assert current.model == "nova-3-general" + assert "punctuate" in changed + + def test_roundtrip_model_via_dict(self): + """Dict update with model should change top-level and NOT create live_options.""" + current = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions(encoding="linear16", channels=1), + ) + + raw = {"model": "nova-2"} + delta = DeepgramSTTSettings.from_mapping(raw) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + +# --------------------------------------------------------------------------- +# DeepgramSageMakerSTTSettings: smoke test that the shared base is inherited +# --------------------------------------------------------------------------- + + +class TestDeepgramSageMakerSTTSettings: + def test_inherits_live_options_behavior(self): + """Smoke test: SageMaker settings inherit the shared base correctly.""" + store = DeepgramSageMakerSTTSettings( + model="nova-3", + language="en", + live_options=LiveOptions(encoding="linear16", channels=1, punctuate=True), + ) + delta = DeepgramSageMakerSTTSettings(live_options=LiveOptions(punctuate=False)) + changed = store.apply_update(delta) + + assert store.live_options.punctuate is False + assert store.live_options.encoding == "linear16" + assert "punctuate" in changed