Reference client implementation

Article sections

    Example of user presence monitoring

    This is a reference client implementation written in Python.

    user_monitor.py

    #!/usr/bin/python
    """
    Sample python script showing how to monitor PBX user presence using the PBX
    API.
    """
    import sys
    import urllib
    import urllib2
    from base64 import b64encode
    try:
    import json
    except ImportError:
    import simplejson as json
    USAGE = """Usage:
    %(name)s [--ssl] <pbx_address> <api_key>
    arguments:
    --ssl         - use SSL
    <pbx_address> - IP address or host name of a PBX
    <api_key>     - the API Key
    """
    MAX_RETRIES = 3
    def api_call(call_url, api_key, options=None):
    """Make an API call, return decoded JSON response.
    Uses HTTP authentication and retries on error.
    See api_call_simple() for simplified version.
    """
    if options:
    call_url += "?" + urllib.urlencode(options)
    # standard urlib2 authentication handler is inefficient
    auth = "Basic " + b64encode('apiKey:' + api_key)
    attempt = 0
    while True:
    attempt += 1
    try:
    req = urllib2.Request(call_url)
    req.add_header('Accept', 'application/json')
    req.add_header('Authorization', auth)   
    response = urllib2.urlopen(req)
    result = json.load(response)
    break
    except IOError, err:
    print "I/O error for %s: %s" % (call_url, err)
    if attempt == MAX_RETRIES:
    raise
    else:
    continue
    except ValueError, err:
    print "Invalid response from %s: %s" % (call_url, err)
    if attempt == MAX_RETRIES:
    raise
    else:
    continue
    return result
    def api_call_simple(call_url, api_key, options=None):
    """The minimal implementation of an API call."""
    if not options:
    options = {}
    options['key'] = api_key
    call_url += "?" + urllib.urlencode(options)
    return json.load(urllib.urlopen(call_url))
    def load_users(api_url, api_key):
    users = api_call(api_url + "users", api_key, {"status": "1"})
    return dict([(user['uuid'], user) for user in users])
    def load_phones(api_url, api_key):
    phones = api_call(api_url + "phones", api_key, {"status": "1"})
    return dict([(phone['uuid'], phone) for phone in phones])
    def load_events(api_url, api_key, last=None, limit=None, timeout=None):
    options = {}
    if last is not None:
    options["last"] = last
    if limit is not None:
    options["limit"] = limit
    if timeout is not None:
    options["timeout"] = timeout
    events = api_call(api_url + "events", api_key, options)
    return events
    def dump_user(user, phones):
    username = user["username"]
    full_name = user["full_name"]
    if full_name:
    name = "%s (%s)" % (username, full_name)
    else:
    name = username
    phone_uuid = user['phone']
    if not phone_uuid or phone_uuid not in phones:
    print "%s: no phone" % (name,)
    return
    phone = phones[phone_uuid]
    print "%s: %s" % (name, phone['status'])
    def dump_users(users, phones):
    users = users.values()
    users.sort(key = lambda u: u["username"])
    for user in users:
    dump_user(user, phones)
    def main():
    if len(sys.argv) == 4 and sys.argv[1] == '--ssl':
    address, api_key = sys.argv[2:]
    api_url = 'https://%s/apis/pbx/' % (address,)
    elif len(sys.argv) == 3:
    address, api_key = sys.argv[1:]
    api_url = 'http://%s/apis/pbx/' % (address,)
    else:
    print >> sys.stderr, USAGE % {"name": sys.argv[0]}
    sys.exit(2)
    # flush old events
    events = load_events(api_url, api_key, limit=1, timeout=0)
    if events:
    last_event = events[0]['seq']
    else:
    last_event = None
    # load user and phone database with current state
    users = load_users(api_url, api_key)
    phones = load_phones(api_url, api_key)
    dump_users(users, phones)
    print
    while True:
    events = load_events(api_url, api_key, last=last_event)
    if events:
    last_event = events[-1]['seq']
    for event in events:
    event_type = event['event_type']
    if event_type == 'user list changed':
    print 'User list changed'
    users = load_users(api_url, api_key)
    dump_users(users, phones)
    elif event_type == 'phone list changed':
    print 'Phone list changed'
    phones = load_phones(api_url, api_key)
    dump_users(users, phones)
    elif event_type == 'phone status update':
    phone_uuid = event['phone_uuid']
    if phone_uuid not in phones:
    print 'status change for unknown phone:', phone_uuid
    else:
    phone = phones[phone_uuid]
    phone['status'] = event['status']
    phone['status_code'] = event['status_code']
    for user in users.values():
    if user['phone'] == phone_uuid:
    dump_user(user, phones)
    elif event_type == 'user logged in':
    user_uuid = event['user_uuid']
    if user_uuid not in users:
    print 'unknown user logged in:', user_uuid
    else:
    user = users[user_uuid]
    user['phone'] = event['phone_uuid']
    dump_user(user, phone)
    elif event_type == 'user logged out':
    user_uuid = event['user_uuid']
    if user_uuid not in users:
    print 'unknown user logged out:', user_uuid
    else:
    user = users[user_uuid]
    user['phone'] = user['primary_phone']
    dump_user(user, phones)
    print
    if __name__ == "__main__":
    try:
    main()
    except KeyboardInterrupt:
    print
    except IOError:
    print "Aborting!"
    

    in API
    Did this article answer your question?

    No Comments – be the first.

    Leave a Reply