Я хочу транслировать аудио из Интернета и преобразовывать его в текст с помощью Python Google-cloud-speech API. Я интегрировал это в свой код каналов Django.
Для интерфейса я напрямую скопировал этот code, и этот код есть в бэкенде (см. ниже). Теперь, что касается проблемы, я не получаю никаких исключений или ошибок, но я не получал никаких результатов от API Google.
Что я пробовал:
Я помещаю точки отладки внутри цикла for функции
process, управление никогда не попадает внутрь цикла.Я просмотрел код Java здесь и пытался это понять. У меня есть настройка этого java-кода на моем локальном компьютере и его отладка. Я понял одну вещь: в java-коде метод
onWebSocketBinaryполучает целочисленный массив, из внешнего интерфейса мы отправляем его вот так.socket.send(Int16Array.from(floatSamples.map(function (n) {return n * MAX_INT;})));В java они конвертируются в байтовую строку, а затем отправляются в Google. Тогда как в Django я поставил точки отладки и заметил, что получаю данные в двоичной строке. Так что я чувствовал, что мне не нужно ничего делать с этим. но я попробовал несколько способов, преобразовав это в целочисленный массив, но это не сработало, потому что Google ожидает в байтах (вы можете увидеть прокомментированный код ниже).
Я просмотрел этот пример кода и это от Google и я делаю то же самое, я не понял, что я делаю здесь неправильно.
Код Джанго:
import json
from channels.generic.websocket import WebsocketConsumer
# Imports the Google Cloud client library
from google.cloud import speech
from google.cloud.speech import enums
from google.cloud.speech import types
# Instantiates a client
client = speech.SpeechClient()
language_code = "en-US"
streaming_config = None
class SpeechToTextConsumer(WebsocketConsumer):
def connect(self):
self.accept()
def disconnect(self, close_code):
pass
def process(self, streaming_recognize_response: types.StreamingRecognitionResult):
for response in streaming_recognize_response:
if not response.results:
continue
result = response.results[0]
self.send(text_data=json.dumps(result))
def receive(self, text_data=None, bytes_data=None):
global streaming_config
if text_data:
data = json.loads(text_data)
rate = data["sampleRate"]
config = types.RecognitionConfig(
encoding=enums.RecognitionConfig.AudioEncoding.LINEAR16,
sample_rate_hertz=rate,
language_code=language_code,
)
streaming_config = types.StreamingRecognitionConfig(
config=config, interim_results=True, single_utterance=False
)
types.StreamingRecognizeRequest(streaming_config=streaming_config)
self.send(text_data=json.dumps({"message": "processing..."}))
if bytes_data:
# bytes_data = bytes_data[math.floor(len(bytes_data) / 2) :]
# bytes_data = bytes_data.lstrip(b"\x00")
# bytes_data = int.from_bytes(bytes_data, "little")
stream = [bytes_data]
requests = (
types.StreamingRecognizeRequest(audio_content=chunk) for chunk in stream
)
responses = client.streaming_recognize(streaming_config, requests)
self.process(responses)
