-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_deepgram.py
More file actions
82 lines (65 loc) · 2.53 KB
/
test_deepgram.py
File metadata and controls
82 lines (65 loc) · 2.53 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import os
import asyncio
from dotenv import load_dotenv
from deepgram import DeepgramClient, LiveOptions, LiveTranscriptionEvents, DeepgramClientOptions
# Load environment variables
load_dotenv()
# Get API key
api_key = os.getenv("DEEPGRAM_API_KEY")
if not api_key:
print("ERROR: DEEPGRAM_API_KEY not found in environment variables")
exit(1)
# Print masked API key for verification
masked_key = api_key[:4] + "*" * (len(api_key) - 4) if len(api_key) > 4 else "****"
print(f"Using Deepgram API key: {masked_key}")
async def main():
try:
# Initialize Deepgram client
config = DeepgramClientOptions(options={"keepalive": "true"})
deepgram = DeepgramClient(api_key, config)
# Define event handlers
def on_open(open):
print(f"Connection opened: {open}")
def on_message(result):
try:
transcript = result.channel.alternatives[0].transcript
if transcript:
print(f"Transcript: {transcript}")
except Exception as e:
print(f"Error processing transcript: {e}")
def on_error(error):
print(f"Error: {error}")
def on_close(close):
print(f"Connection closed: {close}")
# Create connection
dg_connection = deepgram.listen.live.v("1")
# Register event handlers
dg_connection.on(LiveTranscriptionEvents.Open, on_open)
dg_connection.on(LiveTranscriptionEvents.Transcript, on_message)
dg_connection.on(LiveTranscriptionEvents.Error, on_error)
dg_connection.on(LiveTranscriptionEvents.Close, on_close)
# Define options
options = LiveOptions(
model="nova-2",
language="ko-KR",
punctuate=True,
channels=1,
sample_rate=16000
)
# Start connection
print("Starting connection...")
connection_started = dg_connection.start(options)
if connection_started:
print("Connection started successfully")
# Wait for a moment to see if connection establishes
await asyncio.sleep(5)
# Close connection
print("Closing connection...")
dg_connection.finish()
print("Connection closed")
else:
print("Failed to start connection")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(main())