-
Notifications
You must be signed in to change notification settings - Fork 0
/
speech_to_text_chat.py
189 lines (156 loc) · 6.76 KB
/
speech_to_text_chat.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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
# imports
import time
from transformers import AutoProcessor, SeamlessM4Tv2Model
import logging
import pyaudio
import numpy as np
import torch
import pyttsx3 as pyttsx3
from langchain_cohere import ChatCohere
from langchain.schema import HumanMessage,SystemMessage,AIMessage
from langchain_core.output_parsers import StrOutputParser
import os
from dotenv import load_dotenv
import time
import eel
import speech_recognition as sr
load_dotenv() ## take environment variables from .env(langchain api key, cohere api key)
# ChatCohere llm
ChatCohere_llm = ChatCohere(cohere_api_key=os.environ["COHERE_API_KEY"], temperature=0.5)
chat_flow = [SystemMessage(content="You are a chatting AI assistant")] #store chat message history for a session
def get_ChatCohere_response(human_question):
(chat_flow).append(HumanMessage(content=human_question))
ai_answer = ChatCohere_llm(chat_flow)
(chat_flow).append(AIMessage(content=ai_answer.content))
return ai_answer.content
# speech to text processing model
# logging.info("seamless-m4t-v2.main()")
# processor = AutoProcessor.from_pretrained("facebook/seamless-m4t-v2-large")
# model = SeamlessM4Tv2Model.from_pretrained("facebook/seamless-m4t-v2-large")
# logging.basicConfig(level=logging.INFO, format='%(asctime)-15s %(levelname)s %(message)s')
@eel.expose
def welcome():
print("Welcome, how can I help you?")
eel.welcomeMessage("Welcome, how can I help you?")
response_to_audio("Welcome, how can I help you?")
@eel.expose
def goodbye():
print("Goodbye!")
eel.goodbyeMessage("Goodbye!")
# @eel.expose
# def stopListening(is_running):
# global is_listening
# is_listening = is_running
@eel.expose
def main():
# print("main")
# pass
# while True:
# eel.askMessage("Ask...")
# time.sleep(2)
# eel.assistantMessage("<h4>ASSISTANT</h4>response fron the personal assistant")
# time.sleep(3)
# eel.userMessage("<h4>YOU</h4>query asked by the user")
# time.sleep(3)
# sampling_rate = 16000
# record_time_in_seconds = 5.0
# number_of_samples = round(record_time_in_seconds * sampling_rate)
# mic_stream = pyaudio.PyAudio().open(format=pyaudio.paInt16,
# channels=1,
# rate=sampling_rate,
# input=True,
# frames_per_buffer=number_of_samples)
end_of_conversation = False
while not end_of_conversation:
# print("Ask!"
# eel.askMessage("Ask...")
# transcription = get_sentence(mic_stream, processor, model, sampling_rate)
transcription = get_sentence()
# if transcription.lower().replace('.', '').replace('!', '') == ("Thank you and goodbye").lower():
if transcription.lower().replace('.', '').replace('!', '').replace(" ", "") == ("Thankyouandgoodbye").lower():
logging.info(f"voice_assistant_service.main(): End of conversation")
end_of_conversation = True
else:
sentence_is_gibberish = False
if transcription[0] == '[':
sentence_is_gibberish = True
# for prefix in ["i'm going to ", "the first ", "it's a ", "hello", "it's ", "and "]:
# if transcription.lower().startswith(prefix):
# sentence_is_gibberish = True
if len(transcription) > 15 and not sentence_is_gibberish:
eel.userMessage(f"<h4>YOU</h4>{transcription}")
response = get_ChatCohere_response(transcription)
print("=="*40)
print(f"Cohere Assistant : {response}")
eel.assistantMessage(f"<h4>ASSISTANT</h4>{response}")
response_to_audio(response)
print("=="*40)
time.sleep(1)
goodbye()
# @eel.expose
# def getSelectedLanguage(selected_language):
# print(str(selected_language))
# return str(selected_language).lower()
@eel.expose
def get_sentence():
recognizer = sr.Recognizer()
with sr.Microphone() as audio_source:
print("Ask...")
eel.askMessage("Ask...")
audio = recognizer.listen(audio_source)
try:
# audio_to_text = recognizer.recognize_google(audio)
transcription = recognizer.recognize_whisper(audio, language=str(eel.getSelectedLanguage()()).lower(), translate=True)
if transcription:
print(transcription)
return(transcription)
# else:
# print("end of convo")
except Exception as e:
print(e)
# def get_sentence(mic_stream, stt_processor, stt_model, sampling_rate):
# chunks = []
# speech_has_started = False
# chunk_length = round(1.0 * sampling_rate)
# #logging.info(f"voice_assistant_service.get_sentence(): chunk_length = {chunk_length}")
# while not speech_has_started:
# current_chunk_arr = np.frombuffer(mic_stream.read(chunk_length), dtype=np.int16)
# std = np.std(current_chunk_arr)
# if std >= 100:
# speech_has_started = True
# chunks.append(current_chunk_arr)
# speech_has_stopped = False
# while not speech_has_stopped:
# current_chunk_arr = np.frombuffer(mic_stream.read(chunk_length), dtype=np.int16)
# std = np.std(current_chunk_arr)
# if std < 100:
# chunks.append(current_chunk_arr)
# speech_has_stopped = True
# else:
# chunks.append(current_chunk_arr)
# logging.info(f"voice_assistant_service.get_sentence(): len(chunks) = {len(chunks)}")
# speech_arr = np.concatenate(chunks)
# speech_tsr = torch.from_numpy(speech_arr)
# transcription = None
# audio_inputs = stt_processor(audios=speech_tsr, return_tensors="pt")
# output_tokens = stt_model.generate(**audio_inputs, tgt_lang="eng", generate_speech=False)
# transcription = stt_processor.decode(output_tokens[0].tolist()[0], skip_special_tokens=True)
# print("**"*40)
# print(f"YOU : {transcription}")
# # eel.userMessage(f"<h4>YOU</h4>{transcription}")
# print("**"*40)
# return transcription
@eel.expose
def response_to_audio(text):
speaker = pyttsx3.init()
speaker.say(text)
speaker.runAndWait()
@eel.expose
def text_message_response(text_message):
print(text_message)
eel.userMessage(f"<h4>YOU</h4>{text_message}")
response = get_ChatCohere_response(text_message)
eel.assistantMessage(f"<h4>ASSISTANT</h4>{response}")
# print(eel.getSelectedLanguage()())
# eel.init("web_copy_2")
# eel.start("index_copy_2.html")