Streaming audio efficace basé sur Python sur WebSocket à l'aide d'Asyncio et de Threading

Combiner Asyncio et Threading pour la transcription audio en temps réel

La gestion des données audio en temps réel via une connexion WebSocket présente des difficultés distinctes, notamment lorsqu'elle inclut des API tierces telles que Google Speech-to-Text. Il devient crucial de traiter ces données de manière asynchrone lorsque des flux audio en direct sont transmis depuis une application Android vers un serveur. L’objectif est la transcription en temps réel des entrées du microphone côté client.

Le serveur est chargé de superviser la réception des trames audio dans ce projet et de fournir des transcriptions en temps réel au client. Les Pythons asyncio Le framework, qui permet des opérations asynchrones, est utilisé dans la construction du serveur. Cependant, une synchronisation minutieuse est nécessaire lors de la combinaison d'asyncio pour une transmission WebSocket non bloquante avec filetage pour gérer le traitement audio simultané.

La transcription en temps réel à l'aide de l'API Speech-to-Text de Google est une option très appréciée, mais la combiner avec une configuration de serveur asynchrone peut présenter des défis architecturaux. Rendre le système réactif dans cette configuration tout en garantissant que les composants synchrones et asynchrones fonctionnent à l'unisson présente un problème.

Cet article examine les défis de l’intégration asyncio avec filetage pour la transcription audio en temps réel et fournit des moyens pratiques pour simplifier la procédure. Nous aborderons également des sujets tels que la gestion efficace des connexions WebSocket et l'utilisation de générateurs asynchrones.

Traitement audio asynchrone avec intégration Threading et WebSocket

Les programmes mentionnés ci-dessus exploitent Python asyncio et filetage fonctionnalités pour gérer le streaming audio en temps réel via une connexion WebSocket. Les principaux objectifs sont de récupérer des données audio en direct à partir d'une application Android, de les envoyer à l'API Google Speech-to-Text pour transcription et de fournir au client des transcriptions partiellement terminées. En utilisant asyncio, le serveur est démarré et peut effectuer diverses tâches asynchrones, comme la réception de trames audio et la maintenance des connexions WebSocket. Le serveur peut gérer les données audio et d'autres opérations synchrones sans arrêter la boucle d'événements en intégrant ces tâches avec le threading.

Le Gestionnaire audio La classe, qui supervise la réception et le traitement des données audio, est le cerveau derrière la mise en œuvre. Il stocke les morceaux audio entrants dans une file d'attente. Le serveur décode l'audio une fois reçu et l'ajoute à la file d'attente. Le serveur peut désormais décharger le traitement de l'audio en introduisant ThreadPoolExécuteur, qui lit dans la file d'attente et génère des requêtes pour l'API Google Speech-to-Text. Pour une gestion et une transcription audio efficaces, l’asyncio et le threading doivent être séparés.

La nature asynchrone de la communication WebSocket par rapport au comportement synchrone requis par certains composants du processus de traitement audio présente l'un des défis majeurs de la configuration. Une approche consiste à utiliser le asyncio.run_coroutine_threadsafe commande, qui permet à une fonction asynchrone (telle que la livraison de transcriptions au client) d'être exécutée à partir d'un contexte threadé. Cela garantit que la connexion WebSocket reste réactive pendant que le traitement audio s'effectue en arrière-plan en permettant au serveur de communiquer les données de transcription au client en temps réel.

Par ailleurs, l'intégration de Google synthèse vocale est géré par des techniques asynchrones. Le script envoie des segments audio à l'API Google via le StreamingRecognizeRequest et reçoit en retour de manière asynchrone. Une boucle asynchrone est utilisée pour parcourir les réponses, garantissant que les transcriptions sont traitées et renvoyées au client dans les plus brefs délais. Grâce à l'utilisation d'asyncio pour les opérations WebSocket non bloquantes et au threading pour les processus en arrière-plan, le serveur peut gérer efficacement les flux audio en temps réel, les traiter pour la transcription et renvoyer les résultats dans un format optimal.

import asyncio
import websockets
import base64
from concurrent.futures import ThreadPoolExecutor
from google.cloud import speech
import queue
class AudioHandler:
    def __init__(self, client_handler):
        self._client_handler = client_handler
        self._audio_queue = queue.Queue()
        self._is_streaming = False
        self._speech_client = speech.SpeechClient()
        self._executor = ThreadPoolExecutor(max_workers=1)
    async def receive_audio(self, content, audio_id):
        self._is_streaming = True
        audio_data = base64.b64decode(content)
        self._audio_queue.put(audio_data)
        if not self._request_built:
            future = self._executor.submit(self._build_requests)
            future.add_done_callback(lambda f: self._on_audio_complete(f, audio_id))
    def _build_requests(self):
        audio_generator = self._read_audio()
        requests = (speech.StreamingRecognizeRequest(audio_content=chunk) for chunk in audio_generator)
        responses = self._speech_client.streaming_recognize(config, requests)
        self._listen_print_loop(responses)
    def _read_audio(self):
        while self._is_streaming:
            chunk = self._audio_queue.get()
            yield chunk
    def _listen_print_loop(self, responses):
        for response in responses:
            for result in response.results:
                if result.is_final:
                    asyncio.run_coroutine_threadsafe(self._client_handler.send_transcription(result), self._client_handler.loop)

import asyncio
import websockets
import base64
from google.cloud import speech
from asyncio import Queue
class AsyncAudioHandler:
    def __init__(self, client_handler):
        self._client_handler = client_handler
        self._audio_queue = Queue()
        self._speech_client = speech.SpeechAsyncClient()
        self._is_streaming = False
    async def receive_audio(self, content, audio_id):
        self._is_streaming = True
        await self._audio_queue.put(base64.b64decode(content))
        if not self._request_built:
            self._request_built = True
            await self._build_requests()
    async def _read_audio(self):
        while self._is_streaming:
            chunk = await self._audio_queue.get()
            yield speech.StreamingRecognizeRequest(audio_content=chunk)
    async def _build_requests(self):
        async for response in self._speech_client.streaming_recognize(requests=self._read_audio()):
            await self._listen_print_loop(response)
    async def _listen_print_loop(self, responses):
        for response in responses:
            if response.results:
                result = response.results[0]
                if result.is_final:
                    await self._client_handler.send_transcription(result.alternatives[0].transcript)

Amélioration du streaming audio en temps réel avec gestion des erreurs et optimisation des performances

Robuste gestion des erreurs et l'optimisation de la vitesse sont essentielles pour le traitement audio en temps réel sur les connexions WebSocket, mais elles sont souvent ignorées. Des plantages ou un comportement inhabituel peuvent survenir lors du traitement des flux audio et des transcriptions en direct en raison de pannes de réseau, de surcharge du serveur ou même d'une utilisation inappropriée de l'API. Il est crucial de s'assurer que les erreurs telles que la perte de connexion ou les échecs d'API sont gérées correctement par le serveur WebSocket. Pour garantir la stabilité, des blocs try-sauf peuvent être inclus autour de fonctions cruciales, telles que la lecture de la file d'attente audio ou le traitement des réponses de l'API Google Speech-to-Text.

Maintenir la réactivité du système face à de lourdes charges de travail est un autre élément crucial. Plusieurs images peuvent être diffusées rapidement lors du traitement de l'audio en direct, ce qui pourrait submerger le serveur ou le fournisseur de transcription. L’utilisation d’un système tampon dans la file d’attente, où le serveur peut réguler le flux des blocs de données, est une tactique efficace. Le maintien d'un niveau de performance optimal peut également être obtenu en mettant en œuvre des méthodes de temporisation et de contre-pression au sein du asyncio boucle d'événements, qui garantira que l'audio est traité et transcrit sans aucun retard ni perte de données.

La sécurité est un problème en plus de la performance. La protection de la communication WebSocket est essentielle pour gérer les données sensibles en temps réel, telles que la parole. Garantir les flux de données cryptés entre le serveur et le client est possible en implémentant SSL/TLS pour la connexion WebSocket. De plus, l’injection de données nuisibles peut être évitée en vérifiant d’abord l’intégrité et l’authenticité des données audio entrantes avant de les traiter. L'ensemble du système de streaming et de transcription audio peut être rendu plus fiable, évolutif et sécurisé en accordant la même importance à la sécurité et aux performances.