feat(python): speech-to-text v1 — streaming-opus-encoding

recipes/python/speech-to-text/v1/streaming-opus-encoding/README.md

@@ -0,0 +1,50 @@
+# Streaming Transcription with Opus Encoding (Speech-to-Text v1)
+
+Stream Opus-encoded audio over WebSocket instead of raw PCM to dramatically reduce bandwidth while maintaining identical transcription quality.
+
+## What it does
+
+Deepgram's streaming WebSocket API accepts multiple audio encodings — not just raw PCM. By sending Opus-encoded audio (with `encoding=opus`), you can reduce bandwidth by 10–20x compared to uncompressed linear16 PCM. This is critical for mobile apps, browser-based voice pipelines, and bandwidth-constrained environments. Transcription quality remains identical because Deepgram decodes the Opus audio server-side before running the model.
+
+This recipe downloads a WAV file, converts it to Opus using ffmpeg, then streams the compressed audio over WebSocket. It prints the byte-size comparison so you can see the bandwidth savings.
+
+## Key parameters
+
+| Parameter | Value | Description |
+|-----------|-------|-------------|
+| `model` | `"nova-3"` | Transcription model |
+| `encoding` | `"opus"` | Tells Deepgram the incoming audio is Opus-encoded |
+| `sample_rate` | `48000` | Sample rate of the Opus audio (48 kHz is standard for Opus) |
+| `smart_format` | `True` | Format numbers, dates, etc. |
+
+## Example output
+
+```
+PCM: 4379648 bytes → Opus: 220184 bytes (19.9x smaller)
+Yeah, as much as it's worth celebrating the 50th anniversary of the spacewalk,
+it's also worth noting that we've come a long way since then...
+```
+
+## When to use Opus vs PCM
+
+- **Opus**: mobile apps, browser MediaRecorder, bandwidth-limited networks, cost-sensitive streaming
+- **PCM (linear16)**: lowest latency, no encoding overhead, local/server-side processing
+
+## Prerequisites
+
+- Python 3.10+
+- [ffmpeg](https://ffmpeg.org/) installed and on PATH
+- Set `DEEPGRAM_API_KEY` environment variable
+- Install: `pip install -r recipes/python/requirements.txt`
+
+## Run
+
+```bash
+python example.py
+```
+
+## Test
+
+```bash
+pytest example_test.py -v
+```

recipes/python/speech-to-text/v1/streaming-opus-encoding/example.py

@@ -0,0 +1,47 @@
+"""Stream Opus-encoded audio over WebSocket for 10-20x bandwidth savings."""
+
+import subprocess
+import threading
+import time
+import urllib.request
+
+from deepgram import DeepgramClient
+from deepgram.core.events import EventType
+from deepgram.listen.v1.types import ListenV1Results
+
+AUDIO_URL = "https://dpgr.am/spacewalk.wav"
+
+def main():
+    client = DeepgramClient()
+    wav_data = urllib.request.urlopen(AUDIO_URL).read()
+    opus_data = subprocess.run(
+        ["ffmpeg", "-i", "pipe:0", "-c:a", "libopus", "-ar", "48000",
+         "-ac", "1", "-f", "opus", "pipe:1"],
+        input=wav_data, capture_output=True, check=True,
+    ).stdout
+    print(f"PCM: {len(wav_data)} bytes → Opus: {len(opus_data)} bytes ({len(wav_data)/len(opus_data):.1f}x smaller)")
+
+    with client.listen.v1.connect(
+        model="nova-3", encoding="opus", sample_rate=48000, smart_format=True,
+    ) as connection:
+        def on_message(message) -> None:
+            if isinstance(message, ListenV1Results):
+                txt = message.channel.alternatives[0].transcript
+                if txt:
+                    print(txt)
+
+        connection.on(EventType.MESSAGE, on_message)
+
+        def send_audio():
+            for i in range(0, len(opus_data), 4096):
+                connection.send_media(opus_data[i : i + 4096])
+                time.sleep(0.01)
+            time.sleep(2)
+            connection.send_close_stream()
+
+        threading.Thread(target=send_audio, daemon=True).start()
+        connection.start_listening()
+
+
+if __name__ == "__main__":
+    main()

recipes/python/speech-to-text/v1/streaming-opus-encoding/example_test.py

@@ -0,0 +1,16 @@
+import subprocess
+from pathlib import Path
+
+def test_example_runs():
+    """Runs the streaming-opus-encoding example and verifies it produces output."""
+    example = Path(__file__).parent / "example.py"
+    result = subprocess.run(
+        ["python", str(example)],
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    assert result.returncode == 0, (
+        f"Example failed\nSTDOUT: {result.stdout}\nSTDERR: {result.stderr}"
+    )
+    assert result.stdout.strip(), "Example produced no output"