wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3817 from pipecat-ai/pk/service-settings-fix-back-compat-for-nested-external-sdk-types

Browse Source 
Flatten `LiveOptions` into individual fields on `DeepgramSTTSettings`…
This commit is contained in:
kompfner
2026-02-26 11:08:27 -05:00
committed by GitHub
parent b63ca524f5 faed775d90
commit 7fe458fe59
7 changed files with 451 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/coverage.yaml vendored
View File

@@ -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

2
.github/workflows/tests.yaml vendored
View File

@@ -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

18
examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py
View File

@@ -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")

18
examples/foundational/55a-update-settings-deepgram-stt.py
View File

@@ -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")

233
src/pipecat/services/deepgram/stt.py
View File

@@ -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 = {

88
src/pipecat/services/deepgram/stt_sagemaker.py
View File

@@ -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):

219
tests/test_settings.py
View File

@@ -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