ukamnya/pyst2
1
0 Fork 0
Code Issues Pull Requests Releases Wiki Activity

- copy original 0.2 release from tags/0.2

Browse Source 
git-svn-id: https://pyst.svn.sourceforge.net/svnroot/pyst/pyst/trunk@2 01a3061f-1c3a-49da-a2a0-fa5697faa6a0
This commit is contained in:
ralf committed 2010-06-17 16:16:28 +00:00
1 parent dd33d0c724
commit 0cfa2e82ef
19 files changed
+2024
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

asterisk/__init__.py
+14
View File
@@ -0,0 +1,14 @@
""" pyst - A set of interfaces and libraries to allow programming of asterisk from python.
The pyst project includes several python modules to assist in programming
asterisk with python:
agi - python wrapper for agi
agitb - a module to assist in agi debugging, like cgitb
config - a module for parsing asterisk config files
manager - a module for interacting with the asterisk manager interface
"""
__all__ = ['agi', 'agitb', 'config', 'manager']
asterisk/agi.py
+666
View File
@@ -0,0 +1,666 @@
#!/usr/bin/env python2
# vim: set et sw=4:
"""agi
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' : '666',
'agi_enhanced' : '0.0',
'agi_extension': '666',
'agi_language' : 'en',
'agi_priority' : '1',
'agi_rdnis' : '',
'agi_request' : 'pyst',
'agi_type' : 'IAX'}
"""
import sys, pprint, re
from types import ListType
import signal
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>.*)\))*')
class AGIException(Exception): pass
class AGIError(AGIException): pass
class AGIUnknownError(AGIError): pass
class AGIAppError(AGIError): pass
# there are several different types of hangups we can detect
# they all are derrived from AGIHangup
class AGIHangup(AGIAppError): pass
class AGISIGHUPHangup(AGIHangup): pass
class AGISIGPIPEHangup(AGIHangup): pass
class AGIResultHangup(AGIHangup): pass
class AGIDBError(AGIAppError): pass
class AGIUsageError(AGIError): pass
class AGIInvalidCommand(AGIError): pass
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):
self._got_sighup = False
signal.signal(signal.SIGHUP, self._handle_sighup) # handle SIGHUP
sys.stderr.write('ARGS: ')
sys.stderr.write(str(sys.argv))
sys.stderr.write('\n')
self.env = {}
self._get_agi_env()
def _get_agi_env(self):
while 1:
line = sys.stdin.readline().strip()
sys.stderr.write('ENV LINE: ')
sys.stderr.write(line)
sys.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
sys.stderr.write('class AGI: self.env = ')
sys.stderr.write(pprint.pformat(self.env))
sys.stderr.write('\n')
def _quote(self, string):
return ''.join(['"', str(string), '"'])
def _handle_sighup(self, signum, frame):
"""Handle the SIGHUP signal"""
self._got_sighup = True
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,e:
if e.errno == 32:
# Broken Pipe * let us go
raise AGISIGPIPEHangup("Received SIGPIPE")
else:
raise
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'
sys.stderr.write(' COMMAND: %s' % command)
sys.stdout.write(command)
sys.stdout.flush()
def get_result(self, stdin=sys.stdin):
"""Read the result of a command from Asterisk"""
code = 0
result = {'result':('','')}
line = stdin.readline().strip()
sys.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")
sys.stderr.write(' RESULT_DICT: %s\n' % pprint.pformat(result))
return result
elif code == 510:
raise AGIInvalidCommand(response)
elif code == 520:
usage = [line]
line = stdin.readline().strip()
while line[:3] != '520':
usage.append(line)
line = stdin.readline().strip()
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) == ListType:
digits = ''.join(map(str, digits))
return self._quote(digits)
def answer(self):
"""agi.answer() --> None
Answer channel if not already in answer state.
"""
self.execute('ANSWER')['result'][0]
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)
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]
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)
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')
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)
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)
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'))
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)
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)
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)
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)
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)
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)
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)
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
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)
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)
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)
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)
def record_file(self, filename, format='gsm', escape_digits='#', timeout=DEFAULT_RECORD, offset=0, beep='beep'):
"""agi.record_file(filename, format, escape_digits, timeout=DEFAULT_TIMEOUT, offset=0, beep='beep') --> None
Record to a file until a given dtmf digit in the sequence is received
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
"""
escape_digits = self._process_digit_list(escape_digits)
res = self.execute('RECORD FILE', self._quote(filename), format, escape_digits, timeout, offset, beep)['result'][0]
try:
return chr(int(res))
except:
raise AGIError('Unable to convert result to digit: %s' % res)
def set_autohangup(self, secs):
"""agi.set_autohangup(secs) --> None
Cause the channel to automatically hangup at <time> 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', time)
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)
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
def set_callerid(self, number):
"""agi.set_callerid(number) --> None
Changes the callerid of the current channel.
"""
self.execute('SET CALLERID', number)
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])
def set_variable(self, name, value):
"""Set a channel variable.
"""
self.execute('SET VARIABLE', self._quote(name), self._quote(value))
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
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
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)
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)))
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))
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))
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))
def noop(self):
"""agi.noop() --> None
Does nothing
"""
self.execute('NOOP')
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()
asterisk/agitb.py
+216
View File
@@ -0,0 +1,216 @@
"""More comprehensive traceback formatting for Python scripts.
To enable this module, do:
import asterisk.agitb, asterisk.agi
asterisk.agitb.enable(display = False, logdir = '/var/log/asterisk/')
agi = asterisk.agi.AGI()
asterisk.agitb.enable(agi, False, '/var/log/asterisk')
at the top of your script. The optional arguments to enable() are:
agi - the agi handle to write verbose messages to
display - if true, tracebacks are displayed on the asterisk console
(used with the agi option)
logdir - if set, tracebacks are written to files in this directory
context - number of lines of source code to show for each stack frame
By default, tracebacks are displayed but not saved, and the context is 5 lines.
You may want to add a logdir if you call agitb.enable() before you have
an agi.AGI() handle.
Alternatively, if you have caught an exception and want agitb to display it
for you, call agitb.handler(). The optional argument to handler() is a
3-item tuple (etype, evalue, etb) just like the value of sys.exc_info().
If you do not pass anything to handler() it will use sys.exc_info().
This script was adapted from Ka-Ping Yee's cgitb.
"""
__author__ = 'Matthew Nicholson'
__version__ = '0.1.0'
import sys
__UNDEF__ = [] # a special sentinel object
def lookup(name, frame, locals):
"""Find the value for a given name in the given environment."""
if name in locals:
return 'local', locals[name]
if name in frame.f_globals:
return 'global', frame.f_globals[name]
if '__builtins__' in frame.f_globals:
builtins = frame.f_globals['__builtins__']
if type(builtins) is type({}):
if name in builtins:
return 'builtin', builtins[name]
else:
if hasattr(builtins, name):
return 'builtin', getattr(builtins, name)
return None, __UNDEF__
def scanvars(reader, frame, locals):
"""Scan one logical line of Python and look up values of variables used."""
import tokenize, keyword
vars, lasttoken, parent, prefix, value = [], None, None, '', __UNDEF__
for ttype, token, start, end, line in tokenize.generate_tokens(reader):
if ttype == tokenize.NEWLINE: break
if ttype == tokenize.NAME and token not in keyword.kwlist:
if lasttoken == '.':
if parent is not __UNDEF__:
value = getattr(parent, token, __UNDEF__)
vars.append((prefix + token, prefix, value))
else:
where, value = lookup(token, frame, locals)
vars.append((token, where, value))
elif token == '.':
prefix += lasttoken + '.'
parent = value
else:
parent, prefix = None, ''
lasttoken = token
return vars
def text((etype, evalue, etb), context=5):
"""Return a plain text document describing a given traceback."""
import os, types, time, traceback, linecache, inspect, pydoc
if type(etype) is types.ClassType:
etype = etype.__name__
pyver = 'Python ' + sys.version.split()[0] + ': ' + sys.executable
date = time.ctime(time.time())
head = "%s\n%s\n%s\n" % (str(etype), pyver, date) + '''
A problem occurred in a Python script. Here is the sequence of
function calls leading up to the error, in the order they occurred.
'''
frames = []
records = inspect.getinnerframes(etb, context)
for frame, file, lnum, func, lines, index in records:
file = file and os.path.abspath(file) or '?'
args, varargs, varkw, locals = inspect.getargvalues(frame)
call = ''
if func != '?':
call = 'in ' + func + \
inspect.formatargvalues(args, varargs, varkw, locals,
formatvalue=lambda value: '=' + pydoc.text.repr(value))
highlight = {}
def reader(lnum=[lnum]):
highlight[lnum[0]] = 1
try: return linecache.getline(file, lnum[0])
finally: lnum[0] += 1
vars = scanvars(reader, frame, locals)
rows = [' %s %s' % (file, call)]
if index is not None:
i = lnum - index
for line in lines:
num = '%5d ' % i
rows.append(num+line.rstrip())
i += 1
done, dump = {}, []
for name, where, value in vars:
if name in done: continue
done[name] = 1
if value is not __UNDEF__:
if where == 'global': name = 'global ' + name
elif where == 'local': name = name
else: name = where + name.split('.')[-1]
dump.append('%s = %s' % (name, pydoc.text.repr(value)))
else:
dump.append(name + ' undefined')
rows.append('\n'.join(dump))
frames.append('\n%s\n' % '\n'.join(rows))
exception = ['%s: %s' % (str(etype), str(evalue))]
if type(evalue) is types.InstanceType:
for name in dir(evalue):
value = pydoc.text.repr(getattr(evalue, name))
exception.append('\n%s%s = %s' % (" "*4, name, value))
import traceback
return head + ''.join(frames) + ''.join(exception) + '''
The above is a description of an error in a Python program. Here is
the original traceback:
%s
''' % ''.join(traceback.format_exception(etype, evalue, etb))
class Hook:
"""A hook to replace sys.excepthook that shows tracebacks in HTML."""
def __init__(self, display=1, logdir=None, context=5, file=None,
agi=None):
self.display = display # send tracebacks to browser if true
self.logdir = logdir # log tracebacks to files if not None
self.context = context # number of source code lines per frame
self.file = file or sys.stderr # place to send the output
self.agi = agi
def __call__(self, etype, evalue, etb):
self.handle((etype, evalue, etb))
def handle(self, info=None):
info = info or sys.exc_info()
try:
doc = text(info, self.context)
except: # just in case something goes wrong
import traceback
doc = ''.join(traceback.format_exception(*info))
if self.display:
if self.agi: # print to agi
for line in doc.split('\n'):
self.agi.verbose(line, 4)
else:
self.file.write(doc + '\n')
if self.agi:
self.agi.verbose('A problem occured in a python script', 4)
else:
self.file.write('A problem occured in a python script\n')
if self.logdir is not None:
import os, tempfile
(fd, path) = tempfile.mkstemp(suffix='.txt', dir=self.logdir)
try:
file = os.fdopen(fd, 'w')
file.write(doc)
file.close()
msg = '%s contains the description of this error.' % path
except:
msg = 'Tried to save traceback to %s, but failed.' % path
if self.agi:
self.agi.verbose(msg, 4)
else:
self.file.write(msg + '\n')
try:
self.file.flush()
except: pass
handler = Hook().handle
def enable(agi=None, display=1, logdir=None, context=5):
"""Install an exception handler that formats tracebacks as HTML.
The optional argument 'display' can be set to 0 to suppress sending the
traceback to the browser, and 'logdir' can be set to a directory to cause
tracebacks to be written to files there."""
except_hook = Hook(display=display, logdir=logdir,
context=context, agi=agi)
sys.excepthook = except_hook
global handler
handler = except_hook.handle
asterisk/config.py
+162
View File
@@ -0,0 +1,162 @@
#!/usr/bin/env python
# vim: set expandtab:
"""
Parse Asterisk configuration files.
This module provides parsing functionality for asterisk config files.
import asterisk.config
import sys
# load and parse the config file
try:
config = asterisk.config.Config('/etc/asterisk/extensions.conf')
except asterisk.config.ParseError, (line, reason):
print "Parse Error line: %s: %s" % (line, reason)
sys.exit(1)
except IOError, reason:
print "Error opening file: %s" % reason
sys.exit(1)
# print our parsed output
for category in config.categories:
print '[%s]' % category.name # print the current category
for item in category.items:
print ' %s = %s' % (item.name, item.value)
"""
import sys
class ParseError(Exception): pass
class Line(object):
def __init__(self, line, number):
self.line = ''
self.comment = ''
line = line.strip() # I guess we don't preserve indentation
self.number = number
parts = line.split(';')
if len(parts) >= 2:
self.line = parts[0].strip()
self.comment = ';'.join(parts[1:]) #Just in case the comment contained ';'
else:
self.line = line
def __str__(self):
return self.get_line()
def get_line(self):
if self.comment and self.line:
return '%s\t;%s' % (self.line, self.comment)
elif self.comment and not self.line:
return ';%s' % self.comment
return self.line
class Category(Line):
def __init__(self, line='', num=-1, name=None):
Line.__init__(self, line, num)
if self.line:
if (self.line[0] != '[' or self.line[-1] != ']'):
raise ParseError(self.number, "Missing '[' or ']' in category definition")
self.name = self.line[1:-1]
elif name:
self.name = name
else:
raise Exception("Must provide name or line representing a category")
self.items = []
self.comments = []
def get_line(self):
if self.comment:
return '[%s]\t;%s' % (self.name, self.comment)
return '[%s]' % self.name
def append(self, item):
self.items.append(item)
def insert(self, index, item):
self.items.insert(index, item)
def pop(self, index=-1):
self.items.pop(index)
def remove(self, item):
self.items.remove(item)
class Item(Line):
def __init__(self, line='', num=-1, name=None, value=None):
Line.__init__(self, line, num)
self.style = ''
if self.line:
self.parse()
elif (name and value):
self.name = name
self.value = value
else:
raise Exception("Must provide name or value representing an item")
def parse(self):
try:
name, value = self.line.split('=', 1)
except ValueError:
if self.line.strip()[-1] == ']':
raise ParseError(self.number, "Category name missing '['")
else:
raise ParseError(self.number, "Item must be in name = value pairs")
if value and value[0] == '>':
self.style = '>' #preserve the style of the original
value = value[1:].strip()
self.name = name.strip()
self.value = value
def get_line(self):
if self.comment:
return '%s =%s %s\t;%s' % (self.name, self.style, self.value, self.comment)
return '%s =%s %s' % (self.name, self.style, self.value)
class Config(object):
def __init__(self, filename):
self.filename = filename
self.raw_lines = [] # Holds the raw strings
self.lines = [] # Holds things in order
self.categories = []
# load and parse the file
self.load()
self.parse()
def load(self):
self.raw_lines = open(self.filename).readlines()
#try:
#self.raw_lines = open(self.filename).readlines()
#except IOError:
#sys.stderr.write('WARNING: error opening filename: %s No data read. Starting new file?' % self.filename)
#self.raw_lines = []
def parse(self):
cat = None
num = 0
for line in self.raw_lines:
num += 1
line = line.strip()
if not line or line[0] == ';':
item = Line(line or '', num)
self.lines.append(item)
if cat: cat.comments.append(item)
continue
elif line[0] == '[':
cat = Category(line, num)
self.lines.append(cat)
self.categories.append(cat)
continue
else:
item = Item(line, num)
self.lines.append(item)
if cat: cat.append(item)
continue
asterisk/manager.py
+597
View File
@@ -0,0 +1,597 @@
#!/usr/bin/env python
# vim: set expandtab shiftwidth=4:
"""
Python Interface for Asterisk Manager
This module provides a Python API for interfacing with the asterisk manager.
import asterisk.manager
import sys
def handle_shutdown(event, manager):
print "Recieved shutdown event"
manager.close()
# we could analize the event and reconnect here
def handle_event(event, manager):
print "Recieved event: %s" % event.name
manager = asterisk.manager.Manager()
try:
# connect to the manager
try:
manager.connect('host')
manager.login('user', 'secret')
# register some callbacks
manager.register_event('Shutdown', handle_shutdown) # shutdown
manager.register_event('*', handle_event) # catch all
# get a status report
response = manager.status()
manager.logoff()
except asterisk.manager.ManagerSocketException, (errno, reason):
print "Error connecting to the manager: %s" % reason
sys.exit(1)
except asterisk.manager.ManagerAuthException, reason:
print "Error logging in to the manager: %s" % reason
sys.exit(1)
except asterisk.manager.ManagerException, reason:
print "Error: %s" % reason
sys.exit(1)
finally:
# remember to clean up
manager.close()
Remember all header, response, and event names are case sensitive.
Not all manager actions are implmented as of yet, feel free to add them
and submit patches.
"""
import sys,os
import socket
import threading
import Queue
import re
from cStringIO import StringIO
from types import *
from time import sleep
EOL = '\r\n'
class ManagerMsg(object):
"""A manager interface message"""
def __init__(self, response):
self.response = response # the raw response, straight from the horse's mouth
self.data = ''
self.headers = {}
# parse the response
self.parse(response)
if not self.headers:
# Bad app not returning any headers. Let's fake it
# this could also be the inital greeting
self.headers['Response'] = 'Generated Header'
# 'Response:'
def parse(self, response):
"""Parse a manager message"""
response.seek(0)
data = []
# read the response line by line
for line in response.readlines():
line = line.rstrip() # strip trailing whitespace
if not line: continue # don't process if this is not a message
# locate the ':' in our message, if there is one
if line.find(':') > -1:
item = [x.strip() for x in line.split(':',1)]
# if this is a header
if len(item) == 2:
# store the header
self.headers[item[0]] = item[1]
# otherwise it is just plain data
else:
data.append(line)
# if there was no ':', then we have data
else:
data.append(line)
# store the data
self.data = '%s\n' % '\n'.join(data)
def has_header(self, hname):
"""Check for a header"""
return self.headers.has_key(hname)
def get_header(self, hname):
"""Return the specfied header"""
return self.headers[hname]
def __getitem__(self, hname):
"""Return the specfied header"""
return self.headers[hname]
def __repr__(self):
return self.headers['Response']
class Event(object):
"""Manager interface Events, __init__ expects and 'Event' message"""
def __init__(self, message):
# store all of the event data
self.message = message
self.data = message.data
self.headers = message.headers
# if this is not an event message we have a problem
if not message.has_header('Event'):
raise ManagerException('Trying to create event from non event message')
# get the event name
self.name = message.get_header('Event')
def has_header(self, hname):
"""Check for a header"""
return self.headers.has_key(hname)
def get_header(self, hname):
"""Return the specfied header"""
return self.headers[hname]
def __getitem__(self, hname):
"""Return the specfied header"""
return self.headers[hname]
def __repr__(self):
return self.headers['Event']
def get_action_id(self):
return self.headers.get('ActionID',0000)
class Manager(object):
def __init__(self):
self._sock = None # our socket
self._connected = threading.Event()
self._running = threading.Event()
# our hostname
self.hostname = socket.gethostname()
# our queues
self._message_queue = Queue.Queue()
self._response_queue = Queue.Queue()
self._event_queue = Queue.Queue()
# callbacks for events
self._event_callbacks = {}
self._reswaiting = [] # who is waiting for a response
# sequence stuff
self._seqlock = threading.Lock()
self._seq = 0
# some threads
self.message_thread = threading.Thread(target=self.message_loop)
self.event_dispatch_thread = threading.Thread(target=self.event_dispatch)
self.message_thread.setDaemon(True)
self.event_dispatch_thread.setDaemon(True)
def __del__(self):
self.close()
def connected(self):
"""
Check if we are connected or not.
"""
return self._connected.isSet()
def next_seq(self):
"""Return the next number in the sequence, this is used for ActionID"""
self._seqlock.acquire()
try:
return self._seq
finally:
self._seq += 1
self._seqlock.release()
def send_action(self, cdict={}, **kwargs):
"""
Send a command to the manager
If a list is passed to the cdict argument, each item in the list will
be sent to asterisk under the same header in the following manner:
cdict = {"Action": "Originate",
"Variable": ["var1=value", "var2=value"]}
send_action(cdict)
...
Action: Originate
Variable: var1=value
Variable: var2=value
"""
if not self._connected.isSet():
raise ManagerException("Not connected")
# fill in our args
cdict.update(kwargs)
# set the action id
if not cdict.has_key('ActionID'): cdict['ActionID'] = '%s-%08x' % (self.hostname, self.next_seq())
clist = []
# generate the command
for key, value in cdict.items():
if isinstance(value, list):
for item in value:
item = tuple([key, item])
clist.append('%s: %s' % item)
else:
item = tuple([key, value])
clist.append('%s: %s' % item)
clist.append(EOL)
command = EOL.join(clist)
# lock the soket and send our command
try:
self._sock.sendall(command)
except socket.error, (errno, reason):
raise ManagerSocketException(errno, reason)
self._reswaiting.insert(0,1)
response = self._response_queue.get()
self._reswaiting.pop(0)
if not response:
raise ManagerSocketException(0, 'Connection Terminated')
return response
def _receive_data(self):
"""
Read the response from a command.
"""
# loop while we are sill running and connected
while self._running.isSet() and self._connected.isSet():
lines = []
try:
try:
# if there is data to be read
# read a message
while self._connected.isSet():
line = []
# read a line, one char at a time
while self._connected.isSet():
c = self._sock.recv(1)
if not c: # the other end closed the connection
self._sock.close()
self._connected.clear()
break
line.append(c) # append the character to our line
# is this the end of a line?
if c == '\n':
line = ''.join(line)
break
# if we are no longer connected we probably did not
# recieve a full message, don't try to handle it
if not self._connected.isSet(): break
# make sure our line is a string
assert type(line) in StringTypes
lines.append(line) # add the line to our message
# if the line is our EOL marker we have a complete message
if line == EOL:
break
# check to see if this is the greeting line
if line.find('/') >= 0 and line.find(':') < 0:
self.title = line.split('/')[0].strip() # store the title of the manager we are connecting to
self.version = line.split('/')[1].strip() # store the version of the manager we are connecting to
break
#sleep(.001) # waste some time before reading another line
except socket.error:
self._sock.close()
self._connected.clear()
break
finally:
# if we have a message append it to our queue
if lines and self._connected.isSet():
self._message_queue.put(StringIO(''.join(lines)))
else:
self._message_queue.put(None)
def register_event(self, event, function):
"""
Register a callback for the specfied event.
If a callback function returns True, no more callbacks for that
event will be executed.
"""
# get the current value, or an empty list
# then add our new callback
current_callbacks = self._event_callbacks.get(event, [])
current_callbacks.append(function)
self._event_callbacks[event] = current_callbacks
def unregister_event(self, event, function):
"""
Unregister a callback for the specified event.
"""
current_callbacks = self._event_callbacks.get(event, [])
current_callbacks.remove(function)
self._event_callbacks[event] = current_callbacks
def message_loop(self):
"""
The method for the event thread.
This actually recieves all types of messages and places them
in the proper queues.
"""
# start a thread to recieve data
t = threading.Thread(target=self._receive_data)
t.setDaemon(True)
t.start()
try:
# loop getting messages from the queue
while self._running.isSet():
# get/wait for messages
data = self._message_queue.get()
# if we got None as our message we are done
if not data:
# notify the other queues
self._event_queue.put(None)
for waiter in self._reswaiting:
self._response_queue.put(None)
break
# parse the data
message = ManagerMsg(data)
# check if this is an event message
if message.has_header('Event'):
self._event_queue.put(Event(message))
# check if this is a response
elif message.has_header('Response'):
self._response_queue.put(message)
# this is an unknown message
else:
print 'No clue what we got\n%s' % message.data
finally:
# wait for our data receiving thread to exit
t.join()
def event_dispatch(self):
"""This thread is responsible fore dispatching events"""
# loop dispatching events
while self._running.isSet():
# get/wait for an event
ev = self._event_queue.get()
# if we got None as an event, we are finished
if not ev:
break
# dispatch our events
# first build a list of the functions to execute
callbacks = self._event_callbacks.get(ev.name, [])
callbacks.extend(self._event_callbacks.get('*', []))
# now execute the functions
for callback in callbacks:
if callback(ev, self):
break
def connect(self, host, port=5038):
"""Connect to the manager interface"""
if self._connected.isSet():
raise ManagerException('Already connected to manager')
# make sure host is a string
assert type(host) in StringTypes
port = int(port) # make sure port is an int
# create our socket and connect
try:
self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._sock.connect((host,port))
except socket.error, (errno, reason):
raise ManagerSocketException(errno, reason)
# we are connected and running
self._connected.set()
self._running.set()
# start the event thread
self.message_thread.start()
# start the event dispatching thread
self.event_dispatch_thread.start()
# get our initial connection response
return self._response_queue.get()
def close(self):
"""Shutdown the connection to the manager"""
# if we are still running, logout
if self._running.isSet() and self._connected.isSet():
self.logoff()
if self._running.isSet():
# put None in the message_queue to kill our threads
self._message_queue.put(None)
# wait for the event thread to exit
self.message_thread.join()
# make sure we do not join our self (when close is called from event handlers)
if threading.currentThread() != self.event_dispatch_thread:
# wait for the dispatch thread to exit
self.event_dispatch_thread.join()
self._running.clear()
def login(self, username, secret):
"""Login to the manager, throws ManagerAuthException when login falis"""
cdict = {'Action':'Login'}
cdict['Username'] = username
cdict['Secret'] = secret
response = self.send_action(cdict)
if response.get_header('Response') == 'Error':
raise ManagerAuthException(response.get_header('Message'))
return response
def ping(self):
"""Send a ping action to the manager"""
cdict = {'Action':'Ping'}
response = self.send_action(cdict)
return response
def logoff(self):
"""Logoff from the manager"""
cdict = {'Action':'Logoff'}
response = self.send_action(cdict)
return response
def hangup(self, channel):
"""Hanup the specfied channel"""
cdict = {'Action':'Hangup'}
cdict['Channel'] = channel
response = self.send_action(cdict)
return response
def status(self, channel = ''):
"""Get a status message from asterisk"""
cdict = {'Action':'Status'}
cdict['Channel'] = channel
response = self.send_action(cdict)
return response
def redirect(self, channel, exten, priority='1', extra_channel='', context=''):
"""Redirect a channel"""
cdict = {'Action':'Redirect'}
cdict['Channel'] = channel
cdict['Exten'] = exten
cdict['Priority'] = priority
if context: cdict['Context'] = context
if extra_channel: cdict['ExtraChannel'] = extra_channel
response = self.send_action(cdict)
return response
def originate(self, channel, exten, context='', priority='', timeout='', caller_id='', async=False, account='', variables={}):
"""Originate a call"""
cdict = {'Action':'Originate'}
cdict['Channel'] = channel
cdict['Exten'] = exten
if context: cdict['Context'] = context
if priority: cdict['Priority'] = priority
if timeout: cdict['Timeout'] = timeout
if caller_id: cdict['CallerID'] = caller_id
if async: cdict['Async'] = 'yes'
if account: cdict['Account'] = account
# join dict of vairables together in a string in the form of 'key=val|key=val'
# with the latest CVS HEAD this is no longer necessary
# if variables: cdict['Variable'] = '|'.join(['='.join((str(key), str(value))) for key, value in variables.items()])
if variables: cdict['Variable'] = ['='.join((str(key), str(value))) for key, value in variables.items()]
response = self.send_action(cdict)
return response
def mailbox_status(self, mailbox):
"""Get the status of the specfied mailbox"""
cdict = {'Action':'MailboxStatus'}
cdict['Mailbox'] = mailbox
response = self.send_action(cdict)
return response
def command(self, command):
"""Execute a command"""
cdict = {'Action':'Command'}
cdict['Command'] = command
response = self.send_action(cdict)
return response
def extension_state(self, exten, context):
"""Get the state of an extension"""
cdict = {'Action':'ExtensionState'}
cdict['Exten'] = exten
cdict['Context'] = context
response = self.send_action(cdict)
return response
def absolute_timeout(self, channel, timeout):
"""Set an absolute timeout on a channel"""
cdict = {'Action':'AbsoluteTimeout'}
cdict['Channel'] = channel
cdict['Timeout'] = timeout
response = self.send_action(cdict)
return response
def mailbox_count(self, mailbox):
cdict = {'Action':'MailboxCount'}
cdict['Mailbox'] = mailbox
response = self.send_action(cdict)
return response
class ManagerException(Exception): pass
class ManagerSocketException(ManagerException): pass
class ManagerAuthException(ManagerException): pass