Getting Started with Flux

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