[Python] Google Speech V2 real time streaming from mic

I can't seem to find anywhere in the documentation how to use google's speech V2 API. For some reason V2 seems to be cheaper than V1 (as per google's speech pricing table -- although I have no idea why an outdated version seems to be more expensive), and also V2 supports automatic language detection.

Here's a working example with V1 to "listen" from the microphone and transcribe in real time:

import queue
import re
import sys
import os

from google.cloud import speech


import pyaudio

# Audio recording parameters
RATE = 16000
CHUNK = int(RATE / 10) # 100ms

os.environ["GOOGLE_APPLICATION_CREDENTIALS"]="key_google.json"

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
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
"""
self._buff.put(in_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.
"""
while not self.closed:
# 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]

# 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 listen_print_loop(responses: object) -> None: # Changed the return type to None
num_chars_printed = 0
all_transcripts = [] # To store all transcripts

for response in responses:
if not response.results:
continue

result = response.results[0]
if not result.alternatives:
continue

transcript = result.alternatives[0].transcript
overwrite_chars = " " * (num_chars_printed - len(transcript))

if not result.is_final:
sys.stdout.write(transcript + overwrite_chars + "\r")
sys.stdout.flush()
num_chars_printed = len(transcript)
else:
print(transcript + overwrite_chars)
all_transcripts.append(transcript) # Storing the transcript

if re.search(r"\b(exit|quit)\b", transcript, re.I):
print("Exiting..")
print("All Transcripts: ", all_transcripts) # Print all transcripts if needed
break

num_chars_printed = 0


def main() -> None:
"""Transcribe speech from audio file."""
# See http://g.co/cloud/speech/docs/languages
# for a list of supported languages.
language_code = "es" # a BCP-47 language tag

client = speech.SpeechClient()
config = speech.RecognitionConfig(
encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
sample_rate_hertz=RATE,
language_code=language_code,
)

streaming_config = speech.StreamingRecognitionConfig(
config=config, interim_results=True
)

with MicrophoneStream(RATE, CHUNK) as stream:
audio_generator = stream.generator()
requests = (
speech.StreamingRecognizeRequest(audio_content=content)
for content in audio_generator
)

responses = client.streaming_recognize(streaming_config, requests)

# Now, put the transcription responses to use.
listen_print_loop(responses)


if __name__ == "__main__":
main()


However, I have no idea how to convert the same code to use V2 instead. Here's the only example google has, but that code over there is basically to read a .wav file or any other recorded file, but I need to so it in real time. Anyone knows how to use V2 for real time stream transcribing?

This is my attempt (but it's not working at all):

import os
import queue
from google.cloud.speech_v2 import SpeechClient
from google.cloud.speech_v2.types import cloud_speech
import pyaudio

# Audio recording parameters
RATE = 16000
CHUNK = int(RATE / 10) # 100ms

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "key_google.json"


class MicrophoneStream:
def __init__(self, rate, chunk):
self._rate = rate
self._chunk = chunk
self._buff = queue.Queue()
self.closed = True

def __enter__(self):
self._audio_interface = pyaudio.PyAudio()
self._audio_stream = self._audio_interface.open(
format=pyaudio.paInt16,
channels=1,
rate=self._rate,
input=True,
frames_per_buffer=self._chunk,
stream_callback=self._fill_buffer,
)
self.closed = False
return self

def __exit__(self, type, value, traceback):
self._audio_stream.stop_stream()
self._audio_stream.close()
self.closed = True
self._buff.put(None)
self._audio_interface.terminate()

def _fill_buffer(self, in_data, frame_count, time_info, status_flags):
self._buff.put(in_data)
return None, pyaudio.paContinue

def generator(self):
while not self.closed:
chunk = self._buff.get()
if chunk is None:
return
data = [chunk]
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():
project_id = "stellar-cumulus-379717"

client = SpeechClient()

recognition_config = cloud_speech.RecognitionConfig(
auto_decoding_config=cloud_speech.AutoDetectDecodingConfig(),
language_codes=["en-US"],
model="long",
)

streaming_config = cloud_speech.StreamingRecognitionConfig(config=recognition_config)
config_request = cloud_speech.StreamingRecognizeRequest(
recognizer=f"projects/{project_id}/locations/global/recognizers/_",
streaming_config=streaming_config,
)

with MicrophoneStream(RATE, CHUNK) as stream:
audio_generator = stream.generator()
audio_requests = (
cloud_speech.StreamingRecognizeRequest(audio_content=content)
for content in audio_generator
)

def requests():
yield config_request
yield from audio_requests

responses = client.streaming_recognize(requests=requests())

for response in responses:
for result in response.results:
print(f"Transcript: {result.alternatives[0].transcript}")


if __name__ == "__main__":
main()


I am getting a bunch of errors:

Traceback (most recent call last):
File "C:\Python3\Lib\site-packages\google\api_core\grpc_helpers.py", line 162, in error_remapped_callable
return _StreamingResponseIterator(
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python3\Lib\site-packages\google\api_core\grpc_helpers.py", line 88, in __init__
self._stored_first_result = next(self._wrapped)
^^^^^^^^^^^^^^^^^^^
File "C:\Python3\Lib\site-packages\grpc\_channel.py", line 541, in __next__
return self._next()
^^^^^^^^^^^^
File "C:\Python3\Lib\site-packages\grpc\_channel.py", line 967, in _next
raise self
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.UNKNOWN
details = "Exception iterating requests!"
debug_error_string = "None"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "D:\AI_workers\pocketchat\stt2.py", line 98, in <module>
main()
File "D:\AI_workers\pocketchat\stt2.py", line 90, in main
responses = client.streaming_recognize(requests=requests())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python3\Lib\site-packages\google\cloud\speech_v2\services\speech\client.py", line 1639, in streaming_recognize
response = rpc(
^^^^
File "C:\Python3\Lib\site-packages\google\api_core\gapic_v1\method.py", line 113, in __call__
return wrapped_func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Python3\Lib\site-packages\google\api_core\grpc_helpers.py", line 166, in error_remapped_callable
raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.Unknown: None Exception iterating requests!

Continue reading...