File size: 4,114 Bytes
1a3fc6f |
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 117 118 119 120 121 122 123 124 125 126 127 128 129 |
from deepgram import DeepgramClient, LiveTranscriptionEvents, LiveOptions
import asyncio
import os
from src.logging import initialize_logger
import logging
import threading
import time
logger = initialize_logger("transcriber", level=logging.INFO)
options = LiveOptions(
model="nova-2",
language="en-US",
smart_format=True,
punctuate=True,
# smart_format=True,
sample_rate=48000,
interim_results=True,
)
class Transcriber:
def __init__(
self,
):
self.deepgram_api_key = os.getenv("DEEPGRAM_API_KEY")
self.deepgram = None
self.dg_connection = None
self.audio_queue = asyncio.Queue()
self.stop_event = threading.Event()
def process_audio(self):
while not self.stop_event.is_set():
try:
if self.dg_connection is None:
logger.info("returned from process")
return
if self.audio_queue.empty():
time.sleep(0.1)
continue
data = self.audio_queue.get_nowait()
self.dg_connection.send(data)
self.audio_queue.task_done()
logger.info("sent data to deepgram")
except Exception as e:
logger.warning(f"Error while sending data: {e}")
break
logger.info("Audio processing thread is stopping")
def on_transcript(self, result, *args, **kwargs):
try:
sentence = result.channel.alternatives[0].transcript
logger.info(f"Transcription: {sentence}")
except Exception as e:
logger.warning(e)
def close_connection(self):
if self.dg_connection:
self.dg_connection.finish()
self.dg_connection = None
logger.info("finished deepgram connection")
def stop(self):
self.stop_event.set()
self.close_connection()
logger.info("Requested to stop the audio processing thread")
def on_close(self, *args, **kwargs):
logger.info("Deepgram connection closed")
self.dg_connection = None
def on_utterance_end(self, utterance_end, *args, **kwargs):
logger.info(f"\n\n{utterance_end}\n\n")
def on_error(self, e, *args, **kwargs):
logger.warning(f"Deepgram error received {e}")
self.dg_connection = None
def start_deepgram(self):
try:
self.deepgram = DeepgramClient(self.deepgram_api_key)
dg_connection = self.deepgram.listen.live.v("1")
except Exception as e:
logger.warning(f"Could not open socket: {e}")
return
def on_message(self, result, **kwargs):
sentence = result.channel.alternatives[0].transcript
if len(sentence) == 0:
return
logger.info(f"speaker: {sentence}")
def on_metadata(self, metadata, **kwargs):
logger.info(f"\n\n{metadata}\n\n")
def on_utterance_end(self, utterance_end, **kwargs):
logger.info(f"\n\n{utterance_end}\n\n")
def on_error(self, error, **kwargs):
logger.info(f"\n\n{error}\n\n")
def on_close(self, **kwargs):
logger.info(f"\n\nclosed\n\n")
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.Error, on_error)
dg_connection.on(LiveTranscriptionEvents.Close, on_close)
dg_connection.start(options)
self.dg_connection = dg_connection
logger.info("deepgram connection opened")
self.process_audio()
def start(self):
threading.Thread(target=self.start_deepgram).start()
def send_audio(self, data):
try:
self.audio_queue.put_nowait(data)
except Exception as e:
logger.warning(e)
|