From 0623c6c79bd7b146b3c03c0b3d7d1ed80172ec97 Mon Sep 17 00:00:00 2001 From: Varun Singh <382354+vr000m@users.noreply.github.com> Date: Fri, 1 May 2026 13:11:03 +0300 Subject: [PATCH] feat(deepgram): expose mip_opt_out on TTS services --- src/pipecat/services/deepgram/tts.py | 20 ++- tests/test_deepgram_tts.py | 181 +++++++++++++++++++++++++++ 2 files changed, 197 insertions(+), 4 deletions(-) create mode 100644 tests/test_deepgram_tts.py diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 1a1908628..229619fb1 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -12,7 +12,7 @@ for generating speech from text using various voice models. import json from collections.abc import AsyncGenerator -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any import aiohttp @@ -27,7 +27,7 @@ from pipecat.frames.frames import ( TTSAudioRawFrame, TTSStoppedFrame, ) -from pipecat.services.settings import TTSSettings +from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven, is_given from pipecat.services.tts_service import TTSService, WebsocketTTSService from pipecat.utils.tracing.service_decorators import traced_tts @@ -44,9 +44,13 @@ except ModuleNotFoundError as e: @dataclass class DeepgramTTSSettings(TTSSettings): - """Settings for DeepgramTTSService and DeepgramHttpTTSService.""" + """Settings for DeepgramTTSService and DeepgramHttpTTSService. - pass + Parameters: + mip_opt_out: Opt out of Deepgram's Model Improvement Program. + """ + + mip_opt_out: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) class DeepgramTTSService(WebsocketTTSService): @@ -102,6 +106,7 @@ class DeepgramTTSService(WebsocketTTSService): model=None, voice="aura-2-helena-en", language=None, + mip_opt_out=None, ) # 2. Apply direct init arg overrides (deprecated) @@ -221,6 +226,10 @@ class DeepgramTTSService(WebsocketTTSService): params.append(f"model={self._settings.voice}") params.append(f"encoding={self._encoding}") params.append(f"sample_rate={self.sample_rate}") + if is_given(self._settings.mip_opt_out) and self._settings.mip_opt_out is not None: + params.append( + f"mip_opt_out={'true' if self._settings.mip_opt_out else 'false'}" + ) url = f"{self._base_url}/v1/speak?{'&'.join(params)}" @@ -405,6 +414,7 @@ class DeepgramHttpTTSService(TTSService): model=None, voice="aura-2-helena-en", language=None, + mip_opt_out=None, ) # 2. Apply direct init arg overrides (deprecated) @@ -464,6 +474,8 @@ class DeepgramHttpTTSService(TTSService): "sample_rate": self.sample_rate, "container": "none", } + if is_given(self._settings.mip_opt_out) and self._settings.mip_opt_out is not None: + params["mip_opt_out"] = "true" if self._settings.mip_opt_out else "false" payload = { "text": text, diff --git a/tests/test_deepgram_tts.py b/tests/test_deepgram_tts.py new file mode 100644 index 000000000..7cf753711 --- /dev/null +++ b/tests/test_deepgram_tts.py @@ -0,0 +1,181 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from unittest.mock import AsyncMock, MagicMock, patch + +import aiohttp +import pytest + +from pipecat.services.deepgram.tts import DeepgramHttpTTSService, DeepgramTTSService + + +def _make_ws_service(**settings_kwargs) -> DeepgramTTSService: + settings = ( + DeepgramTTSService.Settings(**settings_kwargs) if settings_kwargs else None + ) + service = DeepgramTTSService(api_key="test-key", settings=settings) + # Bypass start() lifecycle: sample_rate is the only field _connect_websocket reads. + service._sample_rate = 16000 + return service + + +@pytest.mark.asyncio +async def test_ws_mip_opt_out_true_in_url(): + service = _make_ws_service(mip_opt_out=True) + + fake_ws = MagicMock() + fake_ws.response.headers = {} + + with patch( + "pipecat.services.deepgram.tts.websocket_connect", + new=AsyncMock(return_value=fake_ws), + ) as mock_connect: + await service._connect_websocket() + + url = mock_connect.call_args.args[0] + assert "mip_opt_out=true" in url + + +@pytest.mark.asyncio +async def test_ws_mip_opt_out_false_in_url(): + service = _make_ws_service(mip_opt_out=False) + + fake_ws = MagicMock() + fake_ws.response.headers = {} + + with patch( + "pipecat.services.deepgram.tts.websocket_connect", + new=AsyncMock(return_value=fake_ws), + ) as mock_connect: + await service._connect_websocket() + + url = mock_connect.call_args.args[0] + assert "mip_opt_out=false" in url + + +@pytest.mark.asyncio +async def test_ws_mip_opt_out_default_absent(): + service = _make_ws_service() + + fake_ws = MagicMock() + fake_ws.response.headers = {} + + with patch( + "pipecat.services.deepgram.tts.websocket_connect", + new=AsyncMock(return_value=fake_ws), + ) as mock_connect: + await service._connect_websocket() + + url = mock_connect.call_args.args[0] + assert "mip_opt_out" not in url + + +@pytest.mark.asyncio +async def test_ws_explicit_empty_settings_omits_mip_opt_out(): + """Explicit Settings() with no kwargs must not leak the NOT_GIVEN sentinel.""" + service = DeepgramTTSService( + api_key="test-key", settings=DeepgramTTSService.Settings() + ) + # Bypass start() lifecycle: sample_rate is the only field _connect_websocket reads. + service._sample_rate = 16000 + + fake_ws = MagicMock() + fake_ws.response.headers = {} + + with patch( + "pipecat.services.deepgram.tts.websocket_connect", + new=AsyncMock(return_value=fake_ws), + ) as mock_connect: + await service._connect_websocket() + + url = mock_connect.call_args.args[0] + assert "mip_opt_out" not in url + + +class _FakeResponse: + def __init__(self): + self.status = 200 + self.content = MagicMock() + + async def _empty_iter(_chunk_size): + return + yield # unreachable; makes this an async generator + + self.content.iter_chunked = _empty_iter + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + +def _make_http_service(**settings_kwargs) -> DeepgramHttpTTSService: + settings = ( + DeepgramHttpTTSService.Settings(**settings_kwargs) if settings_kwargs else None + ) + session = MagicMock(spec=aiohttp.ClientSession) + service = DeepgramHttpTTSService( + api_key="test-key", aiohttp_session=session, settings=settings + ) + # Bypass start() lifecycle: sample_rate is the only field run_tts reads. + service._sample_rate = 16000 + service._session.post = MagicMock(return_value=_FakeResponse()) + return service + + +async def _drain(gen): + async for _ in gen: + pass + + +@pytest.mark.asyncio +async def test_http_mip_opt_out_true_in_params(): + service = _make_http_service(mip_opt_out=True) + + await _drain(service.run_tts("hello", "ctx")) + + params = service._session.post.call_args.kwargs["params"] + assert params["mip_opt_out"] == "true" + + +@pytest.mark.asyncio +async def test_http_mip_opt_out_false_in_params(): + service = _make_http_service(mip_opt_out=False) + + await _drain(service.run_tts("hello", "ctx")) + + params = service._session.post.call_args.kwargs["params"] + assert params["mip_opt_out"] == "false" + + +@pytest.mark.asyncio +async def test_http_mip_opt_out_default_absent(): + service = _make_http_service() + + await _drain(service.run_tts("hello", "ctx")) + + params = service._session.post.call_args.kwargs["params"] + assert "mip_opt_out" not in params + + +@pytest.mark.asyncio +async def test_http_explicit_empty_settings_omits_mip_opt_out(): + """Explicit Settings() with no kwargs must not leak the NOT_GIVEN sentinel.""" + session = MagicMock(spec=aiohttp.ClientSession) + service = DeepgramHttpTTSService( + api_key="test-key", + aiohttp_session=session, + settings=DeepgramHttpTTSService.Settings(), + ) + # Bypass start() lifecycle: sample_rate is the only field run_tts reads. + service._sample_rate = 16000 + service._session.post = MagicMock(return_value=_FakeResponse()) + + await _drain(service.run_tts("hello", "ctx")) + + params = service._session.post.call_args.kwargs["params"] + assert "mip_opt_out" not in params