-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfw_realtime_scripty.py
181 lines (151 loc) · 7.3 KB
/
fw_realtime_scripty.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
# Note W+T: This code was copied from https://github.com/davabase/whisper_real_time
# and translated to fast-whisper using ChatGPT.
import argparse
import os
import numpy as np
import datetime
import speech_recognition as sr
import torch
from datetime import datetime, timedelta
from queue import Queue
from time import sleep
from sys import platform
from faster_whisper import WhisperModel
import os
os.environ["KMP_DUPLICATE_LIB_OK"]="TRUE"
def print_header():
print("Real-time Speech-to-Text Transcription")
print("Author: Adrian Schneider, W+T")
def check_environment():
# Check if CUDA is available
cuda_available = torch.cuda.is_available()
if cuda_available:
cuda_version = torch.version.cuda
print(f"CUDA is available. Version: {cuda_version}")
else:
print("CUDA is not available.")
# Print the versions of the main libraries
print(f"PyTorch version: {torch.__version__}")
print(f"NumPy version: {np.__version__}")
def main():
print_header()
parser = argparse.ArgumentParser()
parser.add_argument("--model", default="medium", help="Model to use",
choices=["tiny", "base", "small", "medium", "large", "large-v2", "large-v3"])
parser.add_argument("--non_english", action='store_true',
help="Don't use the english model.")
parser.add_argument("--energy_threshold", default=1000,
help="Energy level for mic to detect.", type=int)
parser.add_argument("--record_timeout", default=2,
help="How real time the recording is in seconds.", type=float)
parser.add_argument("--phrase_timeout", default=3,
help="How much empty space between recordings before we "
"consider it a new line in the transcription.", type=float)
parser.add_argument("--out_file", default="out.txt",
help="Path of the output file.", type=str)
if 'linux' in platform:
parser.add_argument("--default_microphone", default='pulse',
help="Default microphone name for SpeechRecognition. "
"Run this with 'list' to view available Microphones.", type=str)
args = parser.parse_args()
check_environment()
# The last time a recording was retrieved from the queue.
phrase_time = None
# Thread safe Queue for passing data from the threaded recording callback.
data_queue = Queue()
# We use SpeechRecognizer to record our audio because it has a nice feature where it can detect when speech ends.
recorder = sr.Recognizer()
recorder.energy_threshold = args.energy_threshold
# Definitely do this, dynamic energy compensation lowers the energy threshold dramatically to a point where the SpeechRecognizer never stops recording.
recorder.dynamic_energy_threshold = False
# Important for linux users.
# Prevents permanent application hang and crash by using the wrong Microphone
if 'linux' in platform:
mic_name = args.default_microphone
if not mic_name or mic_name == 'list':
print("Available microphone devices are: ")
for index, name in enumerate(sr.Microphone.list_microphone_names()):
print(f"Microphone with name \"{name}\" found")
return
else:
for index, name in enumerate(sr.Microphone.list_microphone_names()):
if mic_name in name:
source = sr.Microphone(sample_rate=16000, device_index=index)
break
else:
source = sr.Microphone(sample_rate=16000)
# Load / Download model
model_name = args.model
audio_model = WhisperModel(model_name, device="cuda" if torch.cuda.is_available() else "cpu")
if audio_model is not None:
print("Model loaded successfully -> ", model_name)
else:
print("Model failed to load -> ", model_name)
record_timeout = args.record_timeout
phrase_timeout = args.phrase_timeout
transcription = ['']
with source:
recorder.adjust_for_ambient_noise(source)
def record_callback(_, audio: sr.AudioData) -> None:
"""
Threaded callback function to receive audio data when recordings finish.
audio: An AudioData containing the recorded bytes.
"""
# Grab the raw bytes and push it into the thread safe queue.
data = audio.get_raw_data()
data_queue.put(data)
# Create a background thread that will pass us raw audio bytes.
# We could do this manually but SpeechRecognizer provides a nice helper.
recorder.listen_in_background(source, record_callback, phrase_time_limit=record_timeout)
# Cue the user that we're ready to go.
while True:
try:
now = datetime.utcnow()
# Pull raw recorded audio from the queue.
if not data_queue.empty():
phrase_complete = False
# If enough time has passed between recordings, consider the phrase complete.
# Clear the current working audio buffer to start over with the new data.
if phrase_time and now - phrase_time > timedelta(seconds=phrase_timeout):
phrase_complete = True
# This is the last time we received new audio data from the queue.
phrase_time = now
# Combine audio data from queue
audio_data = b''.join(data_queue.queue)
data_queue.queue.clear()
# Convert in-ram buffer to something the model can use directly without needing a temp file.
# Convert data from 16 bit wide integers to floating point with a width of 32 bits.
# Clamp the audio stream frequency to a PCM wavelength compatible default of 32768hz max.
audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
# Transcribe the audio
segments, _ = audio_model.transcribe(audio_np)
# Collect the recognized text from segments
text = "".join([segment.text for segment in segments]).strip()
# If we detected a pause between recordings, add a new item to our transcription.
# Otherwise edit the existing one.
if phrase_complete:
transcription.append(text)
print(text)
# write to out-file
with open(args.out_file, 'a', encoding="utf-8") as f:
current_time = datetime.now()
time_str = current_time.strftime("%Y-%m-%d %H:%M:%S")
f.write(time_str + ": " + text + "\n")
else:
transcription[-1] = text
# Clear the console to reprint the updated transcription.
# os.system('cls' if os.name == 'nt' else 'clear')
#for line in transcription:
#print(line)
# Flush stdout.
print('', end='', flush=True)
else:
# Infinite loops are bad for processors, must sleep.
sleep(0.25)
except KeyboardInterrupt:
break
print("\n\nTranscription:")
for line in transcription:
print(line)
if __name__ == "__main__":
main()