Как подружить 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
#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 = <Здесь FOLDER ID как взять смотрите в документ Яндекса>
token = <Здесь ТОКЕН как взять смотрите в документ Яндекса>
text = ""
cont = true


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:
        label .start
        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')
        if(cont):
            goto .start
    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 из исходников.
Tags:
Asterisk, eagi, python, yandex speechkit

You can't comment this post because its author is not yet a full member of the community. You will be able to contact the author only after he or she has been invited by someone in the community. Until then, author's username will be hidden by an alias.