Overview

I have a server which has an an open WebSocket connection with one client application. This client application (android app) can send live microphone audio data. What the server has to do in response to receiving this data is to reply with partial transcriptions so that the user can see what they are saying be transcribed in real time. I use Google Speech-to-text API for this.

I am also well aware that android has a built in speech recogniser which achieves exactly this.

The server is launched using asyncio.run and incoming data is passed to handlers which all use asynchronous methods. These are the methods that are given the responsibility of handling the reception of an audio frame:

elif action == util.ActionMessages.AUDIO_FRAME:
    audio_id, audio = content["id"], content["audio"]
    await self._audio_handler.receive_audio(audio, audio_id)


# Audio handler method
class AudioHandler:
    def __init__(self, client_handler: ClientHandler):
        self._client_handler = client_handler


        self._audio_finished = dict()

        self._is_streaming = False
        self._audio_queue = queue.Queue()
        self._languages = "en-US"

        self._speech_client = speech.SpeechClient()
        config = speech.RecognitionConfig(...)
        self._streaming_config = speech.StreamingRecognitionConfig(...)

        self._executor = ThreadPoolExecutor(max_workers=1)
        self._request_built = False


    async def receive_audio(self, content: str | None, audio_id: str):

        is_audio_complete = self._audio_finished.setdefault(audio_id, False)
        if content and not is_audio_complete:
            self._is_streaming = True
            content = base64.b64decode(content)
            self._audio_queue.put(content)

            if not self._request_built:
                future = self._executor.submit(self._build_requests)
                future.add_done_callback(lambda f: self._on_audio_processing_complete(f, audio_id))
                self._request_built = True

        elif is_audio_complete:
            # TODO: Implement audio processing complete like clean up dictionary
            pass

        else:
            self._request_built = False
            self._is_streaming = False
            self._audio_queue.put(None)



    def _on_audio_processing_complete(self, future, audio_id):
        self._audio_finished[audio_id] = True
        self._request_built = False

    def _read_audio(self):
        while self._is_streaming:
            chunk = self._audio_queue.get()
            if chunk is None:
                return
            data = [chunk]

            while True:
                try:
                    chunk = self._audio_queue.get_nowait()
                    if chunk is None:
                        return
                    data.append(chunk)
                except queue.Empty:
                    break

            yield b"".join(data)

    def _listen_print_loop(self, responses):
        num_chars_printed = 0
        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))

            # Send transcript clients
            print(transcript + overwrite_chars)

            if not result.is_final:
                num_chars_printed = len(transcript)

            else:
                self._is_streaming = False
                return

    def _build_requests(self):
        audio_generator = self._read_audio()
        requests = (
            speech.StreamingRecognizeRequest(audio_content=content)
            for content in audio_generator
        )

        responses = self._speech_client.streaming_recognize(self._streaming_config, requests)
        self._listen_print_loop(responses)

When audio comes in with the utterance id (audio id) it fills up a queue. On first arrival a new thread is launched which instantiates a generator which reads audio samples from the queue and converts them to the appropriate type. The google speech client uses this generator to perform the transcription. This speech client returns a responses generator which is the used by the _listen_print_loop method to (for now) print the responses/transcriptions.

The logic for using the Google speech api is largely based on their docs.

The issue

As you can imagine, printing the transcription server side is not what I want. I would like to send these partial transcriptions to my client application. However, the method I use to send messages through the socket is async and thus in this implementation cannot be sent from the _listen_print_loop method as it itself is not async. Here is what I mean:

def _listen_print_loop(self, responses):
        num_chars_printed = 0
        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))

            # Cannot do this!
            await send_to_client(transcript + overwrite_chars)

            if not result.is_final:
                num_chars_printed = len(transcript)

            else:
                self._is_streaming = False
                return

I would like to know what the best solution for this is. Is it switching from using threading to only asyncio? If so, would that not mean I would have to implement an async generator function? Would that not cause issues with the speech client?

I am relatively new to asyncio, any pointers would be greatly appreciated!

Edit: Using asyncio.run_coroutine_threadsafe()

I have tried the following to no avail:

def _listen_print_loop(self, responses):
        num_chars_printed = 0
        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))

            # Send on current even loop passed to client handler
            print(transcript + overwrite_chars)
            asyncio.run_coroutine_threadsafe(
                send_to_clients(transcript + overwrite_chars),
                self._client_handler.loop,
            )

            if not result.is_final:
                num_chars_printed = len(transcript)

            else:
                self._is_streaming = False
                return

In this change I use asyncio’s asyncio.run_coroutine_threadsafe() to run the coroutine send_to_clients. The loop variable is set on launch as follows:

async def launch_server():
    # For threads
    client_handler.loop = asyncio.get_running_loop()

    ip_address = "0.0.0.0"
    port = int(os.getenv("PORT"))
    server = await websockets.serve(
        websocket_server,
        ip_address,
        port,
        process_request=health_check
    )
   
    await asyncio.shield(server.wait_closed())


if __name__ == "__main__":
    asyncio.run(launch_server())

This solution does not work as when it reaches the line to send the transcript the WebSocket closes with the following exception:

WebSocket connection closed: no close frame received or sent.

Edit: Using speech.SpeechAsyncClient

I have discovered the the speech module offers the speech.SpeechAsyncClient (link) which I have used as follows:

async def receive_audio(self, content: str | None, audio_id: str):
    is_audio_complete = self._audio_finished.setdefault(audio_id, False)
    if content and not is_audio_complete:
        self._is_streaming = True
        content = base64.b64decode(content)
        await self._audio_queue.put(content)

        if not self._request_built:
            self._request_built = True
            await self._build_requests()

    elif is_audio_complete:
        pass

    else:
        self._request_built = False
        self._is_streaming = False
        await self._audio_queue.put(None)

async def _read_audio(self):
    print("Reading audio")

    config_request = speech.StreamingRecognizeRequest()
    config_request.streaming_config = self._streaming_config
    yield config_request

    while self._is_streaming:
        chunk = await self._audio_queue.get()
        if chunk is None:
            return
        data = [chunk]

        while True:
            try:
                chunk = await self._audio_queue.get_nowait()
                if chunk is None:
                    return
                data.append(chunk)
            except queue.Empty:
                break

        request = speech.StreamingRecognizeRequest()
        request.audio_content = b"".join(data)
        yield request

async def _build_requests(self):
    audio_generator = self._read_audio()
    responses = await self._speech_client.streaming_recognize(
        requests=audio_generator,
    )
    print("Listening for audio")
    await self._listen_print_loop(responses)

This does not cause any error, however, for some reason the program hangs when await the streaming_recognize(...) method. More specifically, the generator _read_audio() is never called meaning no audio is ever processed.

Edit 2.1: Forgot to mention that asyncio.Queue is being used here

Edit 2.2: I have implemented the feature in a scratch file using this method (and my microphone directly) and it works. The issue here is still that the generator is never called (print(“Reading audio”) never reached. This leads me to believe it is the way I am handling asyncio.