#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim: set et sw=4 fenc=utf-8:
"""
.. module:: agi
:synopsis: This module contains functions and classes to implment AGI scripts in python.
pyvr
{'agi_callerid' : 'mars.putland.int',
'agi_channel' : 'IAX[kputland@kputland]/119',
'agi_context' : 'default',
'agi_dnid' : '1000',
'agi_enhanced' : '0.0',
'agi_extension': '1000',
'agi_language' : 'en',
'agi_priority' : '1',
'agi_rdnis' : '',
'agi_request' : 'pyst',
'agi_type' : 'IAX'}
Specification
-------------
"""
import sys
import pprint
import re
import signal
from six import PY3
DEFAULT_TIMEOUT = 2000 # 2sec timeout used as default for functions that take timeouts
DEFAULT_RECORD = 20000 # 20sec record time
re_code = re.compile(r'(^\d*)\s*(.*)')
re_kv = re.compile(r'(?P<key>\w+)=(?P<value>[^\s]+)\s*(?:\((?P<data>.*)\))*')
[docs]class AGIException(Exception):
pass
[docs]class AGIError(AGIException):
pass
[docs]class AGIUnknownError(AGIError):
pass
[docs]class AGIAppError(AGIError):
pass
# there are several different types of hangups we can detect
# they all are derrived from AGIHangup
[docs]class AGIHangup(AGIAppError):
pass
[docs]class AGISIGHUPHangup(AGIHangup):
pass
[docs]class AGISIGPIPEHangup(AGIHangup):
pass
[docs]class AGIResultHangup(AGIHangup):
pass
[docs]class AGIDBError(AGIAppError):
pass
[docs]class AGIUsageError(AGIError):
pass
[docs]class AGIInvalidCommand(AGIError):
pass
[docs]class AGI:
"""
This class encapsulates communication between Asterisk an a python script.
It handles encoding commands to Asterisk and parsing responses from
Asterisk.
"""
def __init__(self, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr):
self.stdin = stdin
self.stdout = stdout
self.stderr = stderr
self._got_sighup = False
signal.signal(signal.SIGHUP, self._handle_sighup) # handle SIGHUP
self.stderr.write('ARGS: ')
self.stderr.write(str(sys.argv))
self.stderr.write('\n')
self.env = {}
self._get_agi_env()
def _get_agi_env(self):
while 1:
line = self.stdin.readline().strip()
if PY3:
if type(line) is bytes: line = line.decode('utf8')
self.stderr.write('ENV LINE: ')
self.stderr.write(line)
self.stderr.write('\n')
if line == '':
#blank line signals end
break
key, data = line.split(':')[0], ':'.join(line.split(':')[1:])
key = key.strip()
data = data.strip()
if key != '':
self.env[key] = data
self.stderr.write('class AGI: self.env = ')
self.stderr.write(pprint.pformat(self.env))
self.stderr.write('\n')
def _quote(self, string):
""" provides double quotes to string, converts int/bool to string """
if isinstance(string, int):
string = str(string)
if isinstance(string, float):
string = str(string)
if PY3:
return ''.join(['"', string, '"'])
else:
return ''.join(['"', string.encode('utf8', 'ignore'), '"'])
def _handle_sighup(self, signum, frame):
"""Handle the SIGHUP signal"""
self._got_sighup = True
[docs] def test_hangup(self):
"""This function throws AGIHangup if we have recieved a SIGHUP"""
if self._got_sighup:
raise AGISIGHUPHangup("Received SIGHUP from Asterisk")
def execute(self, command, *args):
self.test_hangup()
try:
self.send_command(command, *args)
return self.get_result()
except IOError as e:
if e.errno == 32:
# Broken Pipe * let us go
raise AGISIGPIPEHangup("Received SIGPIPE")
else:
raise
[docs] def send_command(self, command, *args):
"""Send a command to Asterisk"""
command = command.strip()
command = '%s %s' % (command, ' '.join(map(str, args)))
command = command.strip()
if command[-1] != '\n':
command += '\n'
self.stderr.write(' COMMAND: %s' % command)
if PY3:
self.stdout.write(command.encode('utf8'))
else:
self.stdout.write(command)
self.stdout.flush()
[docs] def get_result(self, stdin=sys.stdin):
"""Read the result of a command from Asterisk"""
code = 0
result = {'result': ('', '')}
line = self.stdin.readline().strip()
if PY3:
if type(line) is bytes: line = line.decode('utf8')
self.stderr.write(' RESULT_LINE: %s\n' % line)
m = re_code.search(line)
if m:
code, response = m.groups()
code = int(code)
if code == 200:
for key, value, data in re_kv.findall(response):
result[key] = (value, data)
# If user hangs up... we get 'hangup' in the data
if data == 'hangup':
raise AGIResultHangup("User hungup during execution")
if key == 'result' and value == '-1':
raise AGIAppError("Error executing application, or hangup")
self.stderr.write(' RESULT_DICT: %s\n' % pprint.pformat(result))
return result
elif code == 510:
raise AGIInvalidCommand(response)
elif code == 520:
usage = [line]
line = self.stdin.readline().strip()
if PY3:
if type(line) is bytes: line = line.decode('utf8')
while line[:3] != '520':
usage.append(line)
line = self.stdin.readline().strip()
if PY3:
if type(line) is bytes: line = line.decode('utf8')
usage.append(line)
usage = '%s\n' % '\n'.join(usage)
raise AGIUsageError(usage)
else:
raise AGIUnknownError(code, 'Unhandled code or undefined response')
def _process_digit_list(self, digits):
if type(digits) is list:
digits = ''.join(map(str, digits))
return self._quote(digits)
[docs] def answer(self):
"""agi.answer() --> None
Answer channel if not already in answer state.
"""
self.execute('ANSWER')['result'][0]
[docs] def wait_for_digit(self, timeout=DEFAULT_TIMEOUT):
"""agi.wait_for_digit(timeout=DEFAULT_TIMEOUT) --> digit
Waits for up to 'timeout' milliseconds for a channel to receive a DTMF
digit. Returns digit dialed
Throws AGIError on channel falure
"""
res = self.execute('WAIT FOR DIGIT', timeout)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except ValueError:
raise AGIError('Unable to convert result to digit: %s' % res)
[docs] def send_text(self, text=''):
"""agi.send_text(text='') --> None
Sends the given text on a channel. Most channels do not support the
transmission of text.
Throws AGIError on error/hangup
"""
self.execute('SEND TEXT', self._quote(text))['result'][0]
[docs] def receive_char(self, timeout=DEFAULT_TIMEOUT):
"""agi.receive_char(timeout=DEFAULT_TIMEOUT) --> chr
Receives a character of text on a channel. Specify timeout to be the
maximum time to wait for input in milliseconds, or 0 for infinite. Most channels
do not support the reception of text.
"""
res = self.execute('RECEIVE CHAR', timeout)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def tdd_mode(self, mode='off'):
"""agi.tdd_mode(mode='on'|'off') --> None
Enable/Disable TDD transmission/reception on a channel.
Throws AGIAppError if channel is not TDD-capable.
"""
res = self.execute('TDD MODE', mode)['result'][0]
if res == '0':
raise AGIAppError('Channel %s is not TDD-capable')
[docs] def stream_file(self, filename, escape_digits='', sample_offset=0):
"""agi.stream_file(filename, escape_digits='', sample_offset=0) --> digit
Send the given file, allowing playback to be interrupted by the given
digits, if any. escape_digits is a string '12345' or a list of
ints [1,2,3,4,5] or strings ['1','2','3'] or mixed [1,'2',3,'4']
If sample offset is provided then the audio will seek to sample
offset before play starts. Returns digit if one was pressed.
Throws AGIError if the channel was disconnected. Remember, the file
extension must not be included in the filename.
"""
escape_digits = self._process_digit_list(escape_digits)
response = self.execute(
'STREAM FILE', filename, escape_digits, sample_offset)
res = response['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def control_stream_file(self, filename, escape_digits='', skipms=3000, fwd='', rew='', pause=''):
"""
Send the given file, allowing playback to be interrupted by the given
digits, if any. escape_digits is a string '12345' or a list of
ints [1,2,3,4,5] or strings ['1','2','3'] or mixed [1,'2',3,'4']
If sample offset is provided then the audio will seek to sample
offset before play starts. Returns digit if one was pressed.
Throws AGIError if the channel was disconnected. Remember, the file
extension must not be included in the filename.
"""
escape_digits = self._process_digit_list(escape_digits)
response = self.execute('CONTROL STREAM FILE', self._quote(filename), escape_digits, self._quote(skipms), self._quote(fwd), self._quote(rew), self._quote(pause))
res = response['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def send_image(self, filename):
"""agi.send_image(filename) --> None
Sends the given image on a channel. Most channels do not support the
transmission of images. Image names should not include extensions.
Throws AGIError on channel failure
"""
res = self.execute('SEND IMAGE', filename)['result'][0]
if res != '0':
raise AGIAppError('Channel falure on channel %s' %
self.env.get('agi_channel', 'UNKNOWN'))
[docs] def say_digits(self, digits, escape_digits=''):
"""agi.say_digits(digits, escape_digits='') --> digit
Say a given digit string, returning early if any of the given DTMF digits
are received on the channel.
Throws AGIError on channel failure
"""
digits = self._process_digit_list(digits)
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('SAY DIGITS', digits, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_number(self, number, escape_digits=''):
"""agi.say_number(number, escape_digits='') --> digit
Say a given digit string, returning early if any of the given DTMF digits
are received on the channel.
Throws AGIError on channel failure
"""
number = self._process_digit_list(number)
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('SAY NUMBER', number, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_alpha(self, characters, escape_digits=''):
"""agi.say_alpha(string, escape_digits='') --> digit
Say a given character string, returning early if any of the given DTMF
digits are received on the channel.
Throws AGIError on channel failure
"""
characters = self._process_digit_list(characters)
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('SAY ALPHA', characters, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_phonetic(self, characters, escape_digits=''):
"""agi.say_phonetic(string, escape_digits='') --> digit
Phonetically say a given character string, returning early if any of
the given DTMF digits are received on the channel.
Throws AGIError on channel failure
"""
characters = self._process_digit_list(characters)
escape_digits = self._process_digit_list(escape_digits)
res = self.execute(
'SAY PHONETIC', characters, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_date(self, seconds, escape_digits=''):
"""agi.say_date(seconds, escape_digits='') --> digit
Say a given date, returning early if any of the given DTMF digits are
pressed. The date should be in seconds since the UNIX Epoch (Jan 1, 1970 00:00:00)
"""
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('SAY DATE', seconds, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_time(self, seconds, escape_digits=''):
"""agi.say_time(seconds, escape_digits='') --> digit
Say a given time, returning early if any of the given DTMF digits are
pressed. The time should be in seconds since the UNIX Epoch (Jan 1, 1970 00:00:00)
"""
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('SAY TIME', seconds, escape_digits)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def say_datetime(self, seconds, escape_digits='', format='', zone=''):
"""agi.say_datetime(seconds, escape_digits='', format='', zone='') --> digit
Say a given date in the format specfied (see voicemail.conf), returning
early if any of the given DTMF digits are pressed. The date should be
in seconds since the UNIX Epoch (Jan 1, 1970 00:00:00).
"""
escape_digits = self._process_digit_list(escape_digits)
if format:
format = self._quote(format)
res = self.execute(
'SAY DATETIME', seconds, escape_digits, format, zone)['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def get_data(self, filename, timeout=DEFAULT_TIMEOUT, max_digits=255):
"""agi.get_data(filename, timeout=DEFAULT_TIMEOUT, max_digits=255) --> digits
Stream the given file and receive dialed digits
"""
result = self.execute('GET DATA', filename, timeout, max_digits)
res, value = result['result']
return res
[docs] def get_option(self, filename, escape_digits='', timeout=0):
"""agi.get_option(filename, escape_digits='', timeout=0) --> digit
Send the given file, allowing playback to be interrupted by the given
digits, if any. escape_digits is a string '12345' or a list of
ints [1,2,3,4,5] or strings ['1','2','3'] or mixed [1,'2',3,'4']
Returns digit if one was pressed.
Throws AGIError if the channel was disconnected. Remember, the file
extension must not be included in the filename.
"""
escape_digits = self._process_digit_list(escape_digits)
if timeout:
response = self.execute(
'GET OPTION', filename, escape_digits, timeout)
else:
response = self.execute('GET OPTION', filename, escape_digits)
res = response['result'][0]
if res == '0':
return ''
else:
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to char: %s' % res)
[docs] def set_context(self, context):
"""agi.set_context(context)
Sets the context for continuation upon exiting the application.
No error appears to be produced. Does not set exten or priority
Use at your own risk. Ensure that you specify a valid context.
"""
self.execute('SET CONTEXT', context)
[docs] def set_extension(self, extension):
"""agi.set_extension(extension)
Sets the extension for continuation upon exiting the application.
No error appears to be produced. Does not set context or priority
Use at your own risk. Ensure that you specify a valid extension.
"""
self.execute('SET EXTENSION', extension)
[docs] def set_priority(self, priority):
"""agi.set_priority(priority)
Sets the priority for continuation upon exiting the application.
No error appears to be produced. Does not set exten or context
Use at your own risk. Ensure that you specify a valid priority.
"""
self.execute('set priority', priority)
def goto_on_exit(self, context='', extension='', priority=''):
context = context or self.env['agi_context']
extension = extension or self.env['agi_extension']
priority = priority or self.env['agi_priority']
self.set_context(context)
self.set_extension(extension)
self.set_priority(priority)
[docs] def record_file(self, filename, format='gsm', escape_digits='#', timeout=DEFAULT_RECORD, offset=0, beep='beep', silence=0):
"""agi.record_file(filename, format, escape_digits, timeout=DEFAULT_TIMEOUT, offset=0, beep='beep', silence=0) --> None
Record to a file until a given dtmf digit in the sequence is received. Returns
'-1' on hangup or error. The format will specify what kind of file will be
recorded. The <timeout> is the maximum record time in milliseconds, or '-1'
for no <timeout>. <offset samples> is optional, and, if provided, will seek
to the offset without exceeding the end of the file. <silence> is the number
of seconds of silence allowed before the function returns despite the lack
of dtmf digits or reaching <timeout>. <silence> value must be preceded by
's=' and is also optional.
"""
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('RECORD FILE', self._quote(filename), format,
escape_digits, timeout, offset, beep, ('s=%s' % silence))['result'][0]
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to digit: %s' % res)
[docs] def set_autohangup(self, secs):
"""agi.set_autohangup(secs) --> None
Cause the channel to automatically hangup at <secs> seconds in the
future. Of course it can be hungup before then as well. Setting to
0 will cause the autohangup feature to be disabled on this channel.
"""
self.execute('SET AUTOHANGUP', secs)
[docs] def hangup(self, channel=''):
"""agi.hangup(channel='')
Hangs up the specified channel.
If no channel name is given, hangs up the current channel
"""
self.execute('HANGUP', channel)
[docs] def appexec(self, application, options=''):
"""agi.appexec(application, options='')
Executes <application> with given <options>.
Returns whatever the application returns, or -2 on failure to find
application
"""
result = self.execute('EXEC', application, self._quote(options))
res = result['result'][0]
if res == '-2':
raise AGIAppError('Unable to find application: %s' % application)
return res
[docs] def set_callerid(self, number):
"""agi.set_callerid(number) --> None
Changes the callerid of the current channel.
"""
self.execute('SET CALLERID', number)
[docs] def channel_status(self, channel=''):
"""agi.channel_status(channel='') --> int
Returns the status of the specified channel. If no channel name is
given the returns the status of the current channel.
Return values:
0 Channel is down and available
1 Channel is down, but reserved
2 Channel is off hook
3 Digits (or equivalent) have been dialed
4 Line is ringing
5 Remote end is ringing
6 Line is up
7 Line is busy
"""
try:
result = self.execute('CHANNEL STATUS', channel)
except AGIHangup:
raise
except AGIAppError:
result = {'result': ('-1', '')}
return int(result['result'][0])
[docs] def set_variable(self, name, value):
"""Set a channel variable.
"""
self.execute('SET VARIABLE', self._quote(name), self._quote(value))
[docs] def get_variable(self, name):
"""Get a channel variable.
This function returns the value of the indicated channel variable. If
the variable is not set, an empty string is returned.
"""
try:
result = self.execute('GET VARIABLE', self._quote(name))
except AGIResultHangup:
result = {'result': ('1', 'hangup')}
res, value = result['result']
return value
[docs] def get_full_variable(self, name, channel=None):
"""Get a channel variable.
This function returns the value of the indicated channel variable. If
the variable is not set, an empty string is returned.
"""
try:
if channel:
result = self.execute('GET FULL VARIABLE',
self._quote(name), self._quote(channel))
else:
result = self.execute('GET FULL VARIABLE', self._quote(name))
except AGIResultHangup:
result = {'result': ('1', 'hangup')}
res, value = result['result']
return value
[docs] def verbose(self, message, level=1):
"""agi.verbose(message='', level=1) --> None
Sends <message> to the console via verbose message system.
<level> is the the verbose level (1-4)
"""
self.execute('VERBOSE', self._quote(message), level)
[docs] def database_get(self, family, key):
"""agi.database_get(family, key) --> str
Retrieves an entry in the Asterisk database for a given family and key.
Returns 0 if <key> is not set. Returns 1 if <key>
is set and returns the variable in parenthesis
example return code: 200 result=1 (testvariable)
"""
family = '"%s"' % family
key = '"%s"' % key
result = self.execute(
'DATABASE GET', self._quote(family), self._quote(key))
res, value = result['result']
if res == '0':
raise AGIDBError('Key not found in database: family=%s, key=%s' %
(family, key))
elif res == '1':
return value
else:
raise AGIError('Unknown exception for : family=%s, key=%s, result=%s' % (family, key, pprint.pformat(result)))
[docs] def database_put(self, family, key, value):
"""agi.database_put(family, key, value) --> None
Adds or updates an entry in the Asterisk database for a
given family, key, and value.
"""
result = self.execute('DATABASE PUT', self._quote(
family), self._quote(key), self._quote(value))
res, value = result['result']
if res == '0':
raise AGIDBError('Unable to put vaule in databale: family=%s, key=%s, value=%s' % (family, key, value))
[docs] def database_del(self, family, key):
"""agi.database_del(family, key) --> None
Deletes an entry in the Asterisk database for a
given family and key.
"""
result = self.execute(
'DATABASE DEL', self._quote(family), self._quote(key))
res, value = result['result']
if res == '0':
raise AGIDBError('Unable to delete from database: family=%s, key=%s' % (family, key))
[docs] def database_deltree(self, family, key=''):
"""agi.database_deltree(family, key='') --> None
Deletes a family or specific keytree with in a family
in the Asterisk database.
"""
result = self.execute(
'DATABASE DELTREE', self._quote(family), self._quote(key))
res, value = result['result']
if res == '0':
raise AGIDBError('Unable to delete tree from database: family=%s, key=%s' % (family, key))
[docs] def noop(self):
"""agi.noop() --> None
Does nothing
"""
self.execute('NOOP')
[docs] def exec_command(self, command, *args):
"""Send an arbitrary asterisk command with args (even not AGI commands)"""
# The arguments of the command should be prepared as comma delimited, that's the way the EXEC works
args = ','.join(map(str, args))
return self.execute('EXEC', command, args)
if __name__ == '__main__':
agi = AGI()
#agi.appexec('festival','Welcome to Klass Technologies. Thank you for calling.')
#agi.appexec('festival','This is a test of the text to speech engine.')
#agi.appexec('festival','Press 1 for sales ')
#agi.appexec('festival','Press 2 for customer support ')
#agi.hangup()
#agi.goto_on_exit(extension='1234', priority='1')
#sys.exit(0)
#agi.say_digits('123', [4,'5',6])
#agi.say_digits([4,5,6])
#agi.say_number('1234')
#agi.say_number('01234') # 668
#agi.say_number('0xf5') # 245
agi.get_data('demo-congrats')
agi.hangup()
sys.exit(0)
#agi.record_file('pyst-test') #FAILS
#agi.stream_file('demo-congrats', [1,2,3,4,5,6,7,8,9,0,'#','*'])
#agi.appexec('background','demo-congrats')
try:
agi.appexec('backgrounder', 'demo-congrats')
except AGIAppError:
sys.stderr.write(
"Handled exception for missing application backgrounder\n")
agi.set_variable('foo', 'bar')
agi.get_variable('foo')
try:
agi.get_variable('foobar')
except AGIAppError:
sys.stderr.write("Handled exception for missing variable foobar\n")
try:
agi.database_put('foo', 'bar', 'foobar')
agi.database_put('foo', 'baz', 'foobaz')
agi.database_put('foo', 'bat', 'foobat')
v = agi.database_get('foo', 'bar')
sys.stderr.write('DBVALUE foo:bar = %s\n' % v)
v = agi.database_get('bar', 'foo')
sys.stderr.write('DBVALUE foo:bar = %s\n' % v)
agi.database_del('foo', 'bar')
agi.database_deltree('foo')
except AGIDBError:
sys.stderr.write(
"Handled exception for missing database entry bar:foo\n")
agi.hangup()