Apply ruff format

changelog/4399.added.md

@@ -0,0 +1 @@
+- Added `mip_opt_out` to `DeepgramTTSSettings` (used by both `DeepgramTTSService` and `DeepgramHttpTTSService`) for opting out of Deepgram's Model Improvement Program. Pass it via `settings=DeepgramTTSService.Settings(mip_opt_out=True)` to mirror the existing flag on `DeepgramSTTService`.

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,8 @@ 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 +412,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 +472,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,

tests/test_deepgram_tts.py

@@ -0,0 +1,173 @@
+#
+# 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