Как подружить 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 потоков
Asterisk agi python example
The AGI script we’ll be writing in Python, called “The Subtraction Game,” was inspired by a Perl program written by Ed Guy and discussed by him at the 2004 AstriCon conference. Ed described his enthusiasm for the power and simplicity of Asterisk when he found he could write a quick Perl script to help his young daughter improve her math skills.
Since we’ve already written a Perl program using AGI, and Ed has already written the math program in Perl, we figured we’d take a stab at it in Python!
Let’s go through our Python script:
This line tells the system to run this script in the Python interpreter. For small scripts, you may consider adding the — u option to this line, which will run Python in unbuffered mode. This is not recommended, however, for larger or frequently used AGI scripts, as it can affect system performance.
import sys import re import time import random
Here, we import several libraries that we’ll be using in our AGI script.
# Read and ignore AGI environment (read until blank line) env = <> tests = 0; while 1: line = sys.stdin.readline().strip() if line == '': break key,data = line.split(':') if key[:4] <> 'agi_': #skip input that doesn't begin with agi_ sys.stderr.write("Did not work!\n"); sys.stderr.flush() continue key = key.strip() data = data.strip() if key <> '': envAsterisk agi python example = data sys.stderr.write("AGI Environment Dump:\n"); sys.stderr.flush() for key in env.keys(): sys.stderr.write(" -- %s = %s\n" % (key, envAsterisk agi python example)) sys.stderr.flush() This section of code reads in the variables that are passed to our script from Asterisk, and saves them into a dictionary named env . These values are then written to STDERR for debugging purposes.
def checkresult (params): params = params.rstrip() if re.search('^200',params): result = re.search('result=(\d+)',params) if (not result): sys.stderr.write("FAIL ('%s')\n" % params) sys.stderr.flush() return -1 else: result = result.group(1) #debug("Result:%s Params:%s" % (result, params)) sys.stderr.write("PASS (%s)\n" % result) sys.stderr.flush() return result else: sys.stderr.write("FAIL (unexpected result '%s')\n" % params) sys.stderr.flush() return -2 The checkresult function is almost identical in purpose to the checkresult subroutine in the sample Perl AGI script we covered earlier in the chapter. It reads in the result of an Asterisk command, parses the answer, and reports whether or not the command was successful.
def sayit (params): sys.stderr.write("STREAM FILE %s \"\"\n" % str(params)) sys.stderr.flush() sys.stdout.write("STREAM FILE %s \"\"\n" % str(params)) sys.stdout.flush() result = sys.stdin.readline().strip() checkresult(result) The sayit function is a simple wrapper around the STREAM FILE command.
def saynumber (params): sys.stderr.write("SAY NUMBER %s \"\"\n" % params) sys.stderr.flush() sys.stdout.write("SAY NUMBER %s \"\"\n" % params) sys.stdout.flush() result = sys.stdin.readline().strip() checkresult(result) The saynumber function is a simple wrapper around the SAY NUMBER command.
def getnumber (prompt, timelimit, digcount): sys.stderr.write("GET DATA %s %d %d\n" % (prompt, timelimit, digcount)) sys.stderr.flush() sys.stdout.write("GET DATA %s %d %d\n" % (prompt, timelimit, digcount)) sys.stdout.flush() result = sys.stdin.readline().strip() result = checkresult(result) sys.stderr.write("digits are %s\n" % result) sys.stderr.flush() if result: return result else: result = -1 The getnumber function calls the GET DATA command to get DTMF input from the caller. It is used in our program to get the caller’s answers to the subtraction problems.
limit=20 digitcount=2 score=0 count=0 ttanswer=5000
Here, we initialize a few variables that we’ll be using in our program.
starttime = time.time() t = time.time() - starttime
In these lines we set the starttime variable to the current time and initialize t to 0. We’ll use the t variable to keep track of the number of seconds that have elapsed since the AGI script was started.
sayit("subtraction-game-welcome") Next, we welcome the caller to the subtraction game.
This is the heart of the AGI script. We loop through this section of code and give subtraction problems to the caller until 180 seconds have elapsed. Near the top of the loop, we calculate two random numbers and their difference. We then present the problem to the caller, and read in the caller’s response. If the caller answers incorrectly, we give the correct answer.
pct = float(score)/float(count)*100; sys.stderr.write("Percentage correct is %d\n" % pct) sys.stderr.flush() sayit("subtraction-game-timesup") saynumber(score) sayit("subtraction-game-right") saynumber(count) sayit("subtraction-game-pct") saynumber(pct) After the user is done answering the subtraction problems, she is given her score.
As you have seen, the basics you should remember when writing AGI scripts in Python are:
- Flush the output buffer after every write. This will ensure that your AGI program won’t hang while Asterisk is waiting for the buffer to fill and Python is waiting for the response from Asterisk.
- Read data from Asterisk with the sys.stdin.readline command.
- Write commands to Asterisk with the sys.stdout.write command. Don’t forget to call sys.stdout.flush after writing.
The Python AGI Library
If you are planning on writing lot of Python AGI code, you may want to check out Karl Putland’s Python module, Pyst . You can find it at http://www.sourceforge.net/projects/pyst/.
Asterisk agi python example
I’m learning about AGI today, and naturally I decided to attack it with python and no library.
If you’re confused about what AGI can do, don’t be. As it’s name implies it’s not a terribly different concept from CGI, except for the fact that two way communication is extremely pivotal in AGI. Like CGI, It’s all stdin and stdout.
A crude explanation of AGI is a something like this:
- Step 1. Your dialplan invokes an AGI script
- Step 2. Your AGI script reads stdin until it finds an empty line, to get all the AGI variables (you needn’t get this confused with os.getenv — they’re all coming in through stdn in the format “property: value\n”)
- Step (X). Your AGI script sends an AGI command through stdout
- Step (X+1). Your AGI script reads a line on stdin to obtain the response and result from its last command (this is also how you allow your script wait for its command to complete).
- Step (X+2). Do something with that result?
- Finally, Your script exits.
This is what I came up with, the script simply announces your caller ID (as a number), plays a beep, and then accepts up to 5 digits, and then plays those digits back to you (as a number, again).
Extension (somewhere in extensions*.conf…):
exten => 234234,1,Answer
exten => 234234,n,AGI(andy.agi)
exten => 234234,2,Hangup
#!/usr/bin/python import sys,os,datetime def log(msg): open('/tmp/andy.agi.log','ab+',512).write( "%s: %s\n"%(datetime.datetime.now(),msg) ) def get(): res = sys.stdin.readline() log(res) res = res.strip() response,delim,result=res.partition(' ') result=result.split('=')[1].strip() result,delim,data = result.partition(' ') return response,result,data def send(data): log(data) sys.stdout.write("%s \n"%data) sys.stdout.flush() def getDigits(count=1): i=0 digits='' digit = None while (i env = "" while(env != "\n"): env = sys.stdin.readline() log("Looping: env = \"%s\""%env) envdata = env.split(":") if len(envdata)==2: AGIENV[envdata[0].strip()]=envdata[1].strip() log(AGIENV) log ("os.environ: %s"%os.environ) send("SAY NUMBER %s \"\""%AGIENV['agi_callerid']) res = get() send("STREAM FILE beep \"\""); res = get() send('SAY NUMBER %s "*#"\n'%getDigits(5)) res = get() Of course this is completely non-realworld. Anything going to log() was just for me to get a handle on what was actually happening, but I left it in here in case somebody else was going through the same thing. “tail -f” before or during the call gives you the best overview of what’s happening.
Other things
You can pass arguments to your AGI script from the dialplan. Arguments are separated by the | character.
There are certain environmental variables you can access through getenv(), but they tend to be more related to asterisk and not so much AGI specifically.
There’s a great reference of agi variables, env variables, and agi commands available at voip-info, see my sources below.
I should point out that I would also likely be using something like pyst for my real-world applications, but I think it’s important to know all the nitty-gritty before jumping into a valuable library.