From 16cfaaea6a0e90b27766fae77fd0bb6bdfef09da Mon Sep 17 00:00:00 2001 From: recipes-bot Date: Mon, 6 Jul 2026 06:02:10 +0000 Subject: [PATCH] =?UTF-8?q?feat(python):=20speech-to-text=20v1=20=E2=80=94?= =?UTF-8?q?=20streaming-opus-encoding?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🤖 Generated with [Claude Code](https://claude.ai/code) --- .../v1/streaming-opus-encoding/README.md | 50 +++++++++++++++++++ .../v1/streaming-opus-encoding/example.py | 47 +++++++++++++++++ .../streaming-opus-encoding/example_test.py | 16 ++++++ 3 files changed, 113 insertions(+) create mode 100644 recipes/python/speech-to-text/v1/streaming-opus-encoding/README.md create mode 100644 recipes/python/speech-to-text/v1/streaming-opus-encoding/example.py create mode 100644 recipes/python/speech-to-text/v1/streaming-opus-encoding/example_test.py diff --git a/recipes/python/speech-to-text/v1/streaming-opus-encoding/README.md b/recipes/python/speech-to-text/v1/streaming-opus-encoding/README.md new file mode 100644 index 00000000..3d8298d5 --- /dev/null +++ b/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 +``` diff --git a/recipes/python/speech-to-text/v1/streaming-opus-encoding/example.py b/recipes/python/speech-to-text/v1/streaming-opus-encoding/example.py new file mode 100644 index 00000000..0ce09fe4 --- /dev/null +++ b/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() diff --git a/recipes/python/speech-to-text/v1/streaming-opus-encoding/example_test.py b/recipes/python/speech-to-text/v1/streaming-opus-encoding/example_test.py new file mode 100644 index 00000000..170b2556 --- /dev/null +++ b/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"