generated from deepgram/oss-repo-template
/
main.py
116 lines (87 loc) · 3.58 KB
/
main.py
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
# Copyright 2023-2024 Deepgram SDK contributors. All Rights Reserved.
# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
# SPDX-License-Identifier: MIT
import httpx
from dotenv import load_dotenv
import logging, verboselogs
import threading
from deepgram import (
DeepgramClient,
DeepgramClientOptions,
LiveTranscriptionEvents,
LiveOptions,
)
load_dotenv()
# URL for the realtime streaming audio you would like to transcribe
URL = "http://stream.live.vc.bbcmedia.co.uk/bbc_world_service"
def main():
try:
# example of setting up a client config. logging values: WARNING, VERBOSE, DEBUG, SPAM
# config: DeepgramClientOptions = DeepgramClientOptions(verbose=logging.DEBUG)
# deepgram: DeepgramClient = DeepgramClient("", config)
# otherwise, use default config
deepgram: DeepgramClient = DeepgramClient()
# Create a websocket connection to Deepgram
dg_connection = deepgram.listen.live.v("1")
def on_open(self, open, **kwargs):
print(f"\n\n{open}\n\n")
def on_message(self, result, **kwargs):
sentence = result.channel.alternatives[0].transcript
if len(sentence) == 0:
return
print(f"speaker: {sentence}")
def on_metadata(self, metadata, **kwargs):
print(f"\n\n{metadata}\n\n")
def on_speech_started(self, speech_started, **kwargs):
print(f"\n\n{speech_started}\n\n")
def on_utterance_end(self, utterance_end, **kwargs):
print(f"\n\n{utterance_end}\n\n")
def on_close(self, close, **kwargs):
print(f"\n\n{close}\n\n")
def on_error(self, error, **kwargs):
print(f"\n\n{error}\n\n")
def on_unhandled(self, unhandled, **kwargs):
print(f"\n\n{unhandled}\n\n")
dg_connection.on(LiveTranscriptionEvents.Open, on_open)
dg_connection.on(LiveTranscriptionEvents.Transcript, on_message)
dg_connection.on(LiveTranscriptionEvents.Metadata, on_metadata)
dg_connection.on(LiveTranscriptionEvents.SpeechStarted, on_speech_started)
dg_connection.on(LiveTranscriptionEvents.UtteranceEnd, on_utterance_end)
dg_connection.on(LiveTranscriptionEvents.Close, on_close)
dg_connection.on(LiveTranscriptionEvents.Error, on_error)
dg_connection.on(LiveTranscriptionEvents.Unhandled, on_unhandled)
# connect to websocket
options = LiveOptions(model="nova-2", language="en-US")
print("\n\nPress Enter to stop recording...\n\n")
if dg_connection.start(options) is False:
print("Failed to start connection")
return
lock_exit = threading.Lock()
exit = False
# define a worker thread
def myThread():
with httpx.stream("GET", URL) as r:
for data in r.iter_bytes():
lock_exit.acquire()
if exit:
break
lock_exit.release()
dg_connection.send(data)
# start the worker thread
myHttp = threading.Thread(target=myThread)
myHttp.start()
# signal finished
input("")
lock_exit.acquire()
exit = True
lock_exit.release()
# Wait for the HTTP thread to close and join
myHttp.join()
# Indicate that we've finished
dg_connection.finish()
print("Finished")
except Exception as e:
print(f"Could not open socket: {e}")
return
if __name__ == "__main__":
main()