Getting Started with Flux

1
from deepgram import AsyncDeepgramClient
2
3
client = AsyncDeepgramClient()
4
5
# SDK automatically uses /v2/listen endpoint
6
async with client.listen.v2.connect(
7
    model="flux-general-en",
8
    encoding="linear16",
9
    sample_rate="16000"
10
) as connection:
11
    # Your code here
12
    pass

1
 # Install the Deepgram Python SDK
2
 # https://github.com/deepgram/deepgram-python-sdk
3
 pip install deepgram-sdk

1
# Install python-dotenv to protect your API key
2
pip install python-dotenv

1
import asyncio
2
import subprocess
3
from dotenv import load_dotenv
4
5
# Load environment variables from .env file
6
load_dotenv()
7
8
from deepgram import AsyncDeepgramClient
9
from deepgram.core.events import EventType
10
from deepgram.extensions.types.sockets import ListenV2SocketClientResponse
11
12
# URL for the realtime streaming audio to transcribe
13
STREAM_URL = "http://stream.live.vc.bbcmedia.co.uk/bbc_world_service"
14
15
# Terminal color codes
16
class Colors:
17
    GREEN = '\033[92m'    # 0.90-1.00
18
    YELLOW = '\033[93m'   # 0.80-0.90
19
    ORANGE = '\033[91m'   # 0.70-0.80 (using red as orange isn't standard)
20
    RED = '\033[31m'      # <=0.69
21
    RESET = '\033[0m'     # Reset to default
22
23
def get_confidence_color(confidence: float) -> str:
24
    """Return the appropriate color code based on confidence score"""
25
    if confidence >= 0.90:
26
        return Colors.GREEN
27
    elif confidence >= 0.80:
28
        return Colors.YELLOW
29
    elif confidence >= 0.70:
30
        return Colors.ORANGE
31
    else:
32
        return Colors.RED

1
async def main():
2
    """Main async function to handle URL streaming to Deepgram Flux"""
3
4
    # Create the Deepgram async client
5
    client = AsyncDeepgramClient() # The API key retrieval happens automatically in the constructor
6
7
    try:
8
        # Connect to Flux with auto-detection for streaming audio
9
        # SDK automatically connects to: wss://api.deepgram.com/v2/listen?model=flux-general-en&encoding=linear16&sample_rate=16000
10
        async with client.listen.v2.connect(
11
            model="flux-general-en",
12
            encoding="linear16",
13
            sample_rate="16000"
14
        ) as connection:
15
16
            # Define message handler function
17
            def on_message(message: ListenV2SocketClientResponse) -> None:
18
                msg_type = getattr(message, "type", "Unknown")
19
20
                # Show transcription results
21
                if hasattr(message, 'transcript') and message.transcript:
22
                    print(f"🎤 {message.transcript}")
23
24
                    # Show word-level confidence with color coding
25
                    if hasattr(message, 'words') and message.words:
26
                        colored_words = []
27
                        for word in message.words:
28
                            color = get_confidence_color(word.confidence)
29
                            colored_words.append(f"{color}{word.word}({word.confidence:.2f}){Colors.RESET}")
30
                        words_info = " | ".join(colored_words)
31
                        print(f"   📝 {words_info}")
32
                elif msg_type == "Connected":
33
                    print(f"✅ Connected to Deepgram Flux - Ready for audio!")
34
35
            # Set up event handlers
36
            connection.on(EventType.OPEN, lambda _: print("Connection opened"))
37
            connection.on(EventType.MESSAGE, on_message)
38
            connection.on(EventType.CLOSE, lambda _: print("Connection closed"))
39
            connection.on(EventType.ERROR, lambda error: print(f"Caught: {error}"))
40
41
            # Start the connection listening in background (it's already async)
42
            deepgram_task = asyncio.create_task(connection.start_listening())
43
44
            # Convert BBC stream to linear16 PCM using ffmpeg
45
            print(f"Starting to stream and convert audio from: {STREAM_URL}")
46
47
            # Use ffmpeg to convert the compressed BBC stream to linear16 PCM at 16kHz
48
            ffmpeg_cmd = [
49
                'ffmpeg',
50
                '-i', STREAM_URL,           # Input: BBC World Service stream
51
                '-f', 's16le',              # Output format: 16-bit little-endian PCM (linear16)
52
                '-ar', '16000',             # Sample rate: 16kHz
53
                '-ac', '1',                 # Channels: mono
54
                '-'                         # Output to stdout
55
            ]
56
57
            try:
58
                # Start ffmpeg process
59
                process = await asyncio.create_subprocess_exec(
60
                    *ffmpeg_cmd,
61
                    stdout=asyncio.subprocess.PIPE,
62
                    stderr=asyncio.subprocess.PIPE
63
                )
64
65
                print(f"✅ Audio conversion started (BBC → linear16 PCM)")
66
67
                # Read converted PCM data and send to Deepgram
68
                # Note: 1024 bytes = ~32ms of audio at 16kHz linear16
69
                # For optimal performance, consider using ~2560 bytes (~80ms at 16kHz)
70
                while True:
71
                    chunk = await process.stdout.read(1024)
72
                    if not chunk:
73
                        break
74
75
                    # Send converted linear16 PCM data to Flux
76
                    await connection._send(chunk)
77
78
                await process.wait()
79
80
            except Exception as e:
81
                print(f"Error during audio conversion: {e}")
82
                if 'process' in locals():
83
                    stderr = await process.stderr.read()
84
                    print(f"FFmpeg error: {stderr.decode()}")
85
86
            # Wait for Deepgram task to complete (or cancel after timeout)
87
            try:
88
                await asyncio.wait_for(deepgram_task, timeout=60)
89
            except asyncio.TimeoutError:
90
                print("Stream timeout after 60 seconds")
91
                deepgram_task.cancel()
92
93
    except Exception as e:
94
        print(f"Caught: {e}")
95
96
if __name__ == "__main__":
97
    asyncio.run(main())