Как подружить Asterisk с потоковым распознаванием от Яндекс SpeechKit через EAGI и Python

До этого самого времени, я никогда не писал код на Pyhton и Node JS. И поэтому мне было очень сложно скрестить эти системы. И поэтому решил написать об этом пост, так как готовых примеров в Яндексе нет за исключением MRCP про которого было немало нелестных отзывов от пользователей этой системы, но пруфы в данное время предоставить не могу. Я решил не пользоваться этим костылем и изобрести велосипед на костылях сам. Для этого мне в помощь прослужила сама документация с Яндекса и некоторые примеры с StackOverflow.

Код распознает и повторяет что вы говорили на русском некий эхо.

Сначала читаем документ Яндекса о потоковом распознавании следуем все четко по документации выбираем язык Python, так как с Node JS у меня не получилось читать поток из Asteriska EAGI file descrition 3, а так в принципе можете сами попробовать и, если получится написать об этом статью в хабре и дать мне на него ссылку.

Далее пишем экстенш. Здесь все просто с телефона звоним 444 и по идее мы должны слышать то, что говорили, но не сейчас это в конце если у вас все получится.

;;YANDEX SPEACH exten = 444,1,Answer exten = 444,n,EAGI(test.py) exten = 444,n,Hangup 

Затем меняем расположение папки AGI, в файле /etc/asterisk/asterisk.conf

На ту папку, где установлен у нас Яндекс cloudapi/output (из документации яндекса). Изменения в этом файле начинают действовать, как я заметил, только после перезагрузки Asterisk, через службы или kill.

Внимание, был подвох со знаком! его надо было удалить.

Далее сам код, который лежит в папке cloudapi output test.py

#!/usr/bin/python3.6 #coding=utf8 import argparse from asterisk.agi import *  import requests import grpc import os import yandex.cloud.ai.stt.v2.stt_service_pb2 as stt_service_pb2 import yandex.cloud.ai.stt.v2.stt_service_pb2_grpc as stt_service_pb2_grpc agi = AGI() CHUNK_SIZE = 4000 folder = "берете в яндексе" token = "берете в яндексе" text = ""   def gen(folder_id):     # Задать настройки распознавания.     specification = stt_service_pb2.RecognitionSpec(         language_code='ru-RU',         profanity_filter=True,         model='general',         partial_results=True,         audio_encoding='LINEAR16_PCM',         sample_rate_hertz=8000     )      streaming_config = stt_service_pb2.RecognitionConfig(specification=specification, folder_id=folder_id)      # Отправить сообщение с настройками распознавания.     yield stt_service_pb2.StreamingRecognitionRequest(config=streaming_config)      # Прочитать аудиофайл и отправить его содержимое порциями.     with os.fdopen(3, 'rb') as f:         data = f.read(CHUNK_SIZE)         while data != b'':             yield stt_service_pb2.StreamingRecognitionRequest(audio_content=data)             data = f.read(CHUNK_SIZE)   def run(folder_id, iam_token):     # Установить соединение с сервером.     cred = grpc.ssl_channel_credentials()     channel = grpc.secure_channel('stt.api.cloud.yandex.net:443', cred)     stub = stt_service_pb2_grpc.SttServiceStub(channel)      # Отправить данные для распознавания.     it = stub.StreamingRecognize(gen(folder_id), metadata=(('authorization', 'Bearer %s' % iam_token),))      # Обработать ответы сервера и вывести результат в консоль.     try:         cont = True         while cont:                 agi.verbose('START START START')                 for r in it:                     try:                         agi.verbose('Start chunk: ')                         for alternative in r.chunks[0].alternatives:                             agi.verbose(alternative.text)                             if(alternative.text == "привет"):                                 cont = true                                 break                         if r.chunks[0].final:                             agi.verbose("FINAL")                             with open("/var/lib/asterisk/agi-bin/cloudapi/output/echo123.raw", "wb") as f:                                 for audio_content in synthesize(alternative.text):                                     f.write(audio_content)                             agi.stream_file("/var/lib/asterisk/agi-bin/cloudapi/output/echo123")                     except LookupError:                         agi.verbose('Not available chunks')     except grpc._channel._Rendezvous as err:         agi.verbose('Error code %s, message: %s' % (err._state.code, err._state.details))   def synthesize(text):     url = 'https://tts.api.cloud.yandex.net/speech/v1/tts:synthesize'     headers = {         'Authorization': 'Bearer '+token     }      data = {         'text': text,         'lang': 'ru-RU',         'folderId': folder,         'sampleRateHertz': '8000',         'format': 'lpcm'     }      with requests.post(url, headers=headers, data=data, stream=True) as resp:         if resp.status_code != 200:             raise RuntimeError("Invalid response received: code: %d, message: %s" % (resp.status_code, resp.text))          for chunk in resp.iter_content(chunk_size=None):             yield chunk    if __name__ == '__main__':     run(folder,token)  

У меня нет желания писать, как устанавливать питон и его зависимости все просто через pip install. Так как эта статья не для самых маленьких. Кстати, Asterisk распространяемый в Ubuntu репозиториях, но я не смог из него получить поток звука, через EAGI. Получил только после собирания Asterisk из исходников с соответствующей настройкой menuconfig включением eagi потоков