En-IN Потоковая транскрипция не работает в Google STT ⇐ Python

[Anonymous]

Я использую облачные сервисы Google для транскрипции звука, поступающего через микрофон. Я создал распознаватель с параметрами: global=Long=en-In. Во время тестирования он даже не возвращает ни одной транскрипции или слова! Но с другим распознавательом с параметрами global=short=en-IN это работает. Может ли кто-нибудь помочь мне разобраться в этом?

Код: Выделить всё

import queue
import threading
import time

import audioop
import pyaudio

os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = "xxxxxxx"

# Audio recording parameters
RATE = 8000
CHUNK = int(RATE / 25)  # 20ms

class MicrophoneStream:
"""Opens a recording stream as a generator yielding the audio chunks."""

def __init__(self: object, rate: int = RATE, chunk: int = CHUNK) -> None:
"""The audio -- and generator -- is guaranteed to be on the main thread."""
self._rate = rate
self._chunk = chunk

# Create a thread-safe buffer of audio data
self._buff = queue.Queue()
self.closed = True

def __enter__(self: object) -> object:
self._audio_interface = pyaudio.PyAudio()
self._audio_stream = self._audio_interface.open(
format=pyaudio.paInt16,
# The API currently only supports 1-channel (mono) audio
google api link goes here
channels=1,
rate=self._rate,
input=True,
frames_per_buffer=self._chunk,
# Run the audio stream asynchronously to fill the buffer object.
# This is necessary so that the input device's buffer doesn't
# overflow while the calling thread makes network requests, etc.
stream_callback=self._fill_buffer,
)

self.closed = False

return self

def __exit__(
self: object,
type: object,
value: object,
traceback: object,
) -> None:
"""Closes the stream, regardless of whether the connection was lost or not."""
self._audio_stream.stop_stream()
self._audio_stream.close()
self.closed = True
# Signal the generator to terminate so that the client's
# streaming_recognize method will not block the process termination.
self._buff.put(None)
self._audio_interface.terminate()

def _fill_buffer(
self: object,
in_data: object,
frame_count: int,
time_info: object,
status_flags: object,
) -> object:
"""Continuously collect data from the audio stream, into the buffer.

Args:
in_data: The audio data as a bytes object
frame_count: The number of frames captured
time_info: The time information
status_flags: The status flags

Returns:
The audio data as a bytes object
"""
data = audioop.lin2ulaw(in_data, 2)
self._buff.put(data)
return None, pyaudio.paContinue

def generator(self: object) ->  object:
"""Generates audio chunks from the stream of audio data in chunks.

Args:
self: The MicrophoneStream object

Returns:
A generator that outputs audio chunks.
"""
count = 1
while not self.closed:
count += 1
# Use a blocking get() to ensure there's at least one chunk of
# data, and stop iteration if the chunk is None, indicating the
# end of the audio stream.
chunk = self._buff.get()
if chunk is None:
return
data = [chunk]

# if count >= 100:
#     break

# Now consume whatever other data's still buffered.
while True:
try:
chunk = self._buff.get(block=False)
if chunk is None:
return
data.append(chunk)
except queue.Empty:
break

yield b"".join(data)

def main_eng():
client1 = SpeechClient()

print("Speak Eng...")
while True:
with MicrophoneStream(RATE, CHUNK) as stream:
audio_generator = stream.generator()
audio_requests = (
cloud_speech_types.StreamingRecognizeRequest(audio=audio) for audio in audio_generator
)

adaptation = cloud_speech_types.SpeechAdaptation(
phrase_sets=[
cloud_speech_types.SpeechAdaptation.AdaptationPhraseSet(
phrase_set='projects/xxxxxxxx/locations/global/phraseSets/adaptations-en'
)
]
)

recognition_config = cloud_speech_types.RecognitionConfig(
explicit_decoding_config=cloud_speech_types.ExplicitDecodingConfig(encoding='LINEAR16',
sample_rate_hertz=8000,
audio_channel_count=1),
language_codes=["en-IN"],
model="long",
adaptation=adaptation
)

# print(recognition_config)

streaming_config = cloud_speech_types.StreamingRecognitionConfig(
config=recognition_config,
streaming_features=cloud_speech_types.StreamingRecognitionFeatures(interim_results=True)
)
config_request = cloud_speech_types.StreamingRecognizeRequest(
recognizer="projects/xxxxxxx/locations/global/recognizers/abc-en-long",
streaming_config=streaming_config,
)

def requests(config: cloud_speech_types.RecognitionConfig, audio: list) -> list:
yield config
yield from audio

# Transcribes the audio into text
responses_iterator = client1.streaming_recognize(
requests=requests(config_request, audio_requests)
)

responses = []
for response in responses_iterator:
responses.append(response)
for result in response.results:
# print(result)
if result.alternatives and result.is_final:
transcriptB = result.alternatives[0].transcript
print(f"Transcript: from Eng: {transcriptB}")

Я попробовал приведенный выше код с распознавателем en-IN с короткой моделью, он сработал. Для других языков это тоже сработало.

Подробнее здесь: https://stackoverflow.com/questions/787 ... google-stt