Reference client implementation

Article sections

    API example of user presence monitoring

    This is a reference client implementation written in Python. Here, event polling is used to monitor the phone and user status events (to display presence information).

    Sample python script showing how to monitor PBX user presence using the PBX
    import sys
    import urllib.request, urllib.parse, urllib.error
    import urllib.request, urllib.error, urllib.parse
    import logging
    import argparse
    import ssl
    from getpass import getpass
    from base64 import b64encode
    import json
    except ImportError:
    import simplejson as json
    SSL_CONTEXT = None
    def disable_certificate_verification():
    global SSL_CONTEXT
    if hasattr(ssl, "create_default_context"):
    SSL_CONTEXT = ssl.create_default_context()
    SSL_CONTEXT.check_hostname = False
    SSL_CONTEXT.verify_mode = ssl.CERT_NONE
    def api_call(call_url, auth, 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.parse.urlencode(options)
    if auth:
    auth = [a.encode("utf-8") for a in auth]
    auth_hdr = "Basic " + b64encode(b":".join(auth)).decode("us-ascii")
    logging.debug("calling: {0!r}".format(call_url))
    attempt = 0
    while True:
    attempt += 1
    req = urllib.request.Request(call_url)
    req.add_header('Accept', 'application/json')
    req.add_header('Authorization', auth_hdr)
    response = urllib.request.urlopen(req, context=SSL_CONTEXT)
    result = json.loads("utf-8"))
    except IOError as err:
    print("I/O error for %s: %s" % (call_url, err))
    if attempt == MAX_RETRIES:
    except ValueError as err:
    print("Invalid response from %s: %s" % (call_url, err))
    if attempt == MAX_RETRIES:
    logging.debug("got: {0!r}".format(result))
    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.parse.urlencode(options)
    return json.load(urllib.request.urlopen(call_url, context=SSL_CONTEXT))
    def load_users(api_url, auth):
    users = api_call(api_url + "users", auth, {"status": "1"})
    return dict([(user['uuid'], user) for user in users])
    def load_phones(api_url, auth):
    phones = api_call(api_url + "phones", auth, {"status": "1"})
    return dict([(phone['uuid'], phone) for phone in phones])
    def load_events(api_url, auth, 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", auth, options)
    return events
    def get_user_name(user):
    username = user["username"]
    full_name = user["full_name"]
    if full_name:
    name = "%s (%s)" % (username, full_name)
    name = username
    return name
    def dump_user(user, phones):
    name = get_user_name(user)
    phone_uuid = user['phone']
    if not phone_uuid or phone_uuid not in phones:
    print("%s: no phone" % (name,))
    phone = phones[phone_uuid]
    print("%s: %s" % (name, phone['status']))
    def dump_users(users, phones):
    users = list(users.values())
    users.sort(key = lambda u: u["username"])
    for user in users:
    dump_user(user, phones)
    def dump_user_verbose(user, phones):
    name = get_user_name(user)
    print("%s (%s):" % (name, user.get("uuid")))
    print("    username: ", user.get("username"))
    print("    full name:", user.get("full_name"))
    if "phones" not in user:
    # PBX 3.x
    p_phone = user.get("primary_phone")
    if p_phone:
    u_phones = [{
    "type": "phone",
    "phone": p_phone
    u_phones = []
    # PBX 4.x
    u_phones = user["phones"]
    for u_phone in u_phones:
    u_phones = user.get("phones")
    p_type = u_phone["type"]
    phone_uuid = u_phone.get("phone")
    if not phone_uuid or phone_uuid not in phones:
    phone = None
    phone = phones[phone_uuid]
    if p_type == "phone":
    if not phone_uuid or phone_uuid not in phones:
    print("        phone: %s: %s" % (phone['phone_username'],
    elif p_type == "hot-desk":
    print("        hot-desk")
    elif p_type == "number":
    print("        number: %s" % (u_phone.get("number"),))
    elif p_type == "trunk number":
    print("        number: %s on trunk: %s"
    % (u_phone.get("number"), u_phone.get("trunk")))
    print("        unrecognized")
    if not u_phones:
    print("        none")
    if "hot_desk" in user:
    hot_desk = user["hot_desk"]
    hot_desk = user.get("current_phone")
    if hot_desk == user.get("primary_phone"):
    hot_desk = None
    if hot_desk:
    phone = phones.get(hot_desk)
    if phone:
    print("    logged in on phone: %s (%s)"
    % (phone["phone_username"], phone["status"]))
    def dump_users_verbose(users, phones):
    users = list(users.values())
    users.sort(key = lambda u: u["username"])
    for user in users:
    dump_user_verbose(user, phones)
    def dump_target(target, users, phones):
    user_uuid = target.get("user_uuid")
    phone_uuid = target.get("phone_uuid")
    trunk_uuid = target.get("trunk_uuid")
    number = target.get("number")
    if user_uuid:
    user = users.get(user_uuid)
    user_phone_type = target.get("user_phone_type")
    if user:
    print("    user:", get_user_name(user))
    print("    unknown user:", user_uuid)
    if user_phone_type == "phone":
    phone = phones.get(phone_uuid)
    print("        phone:", phone["phone_username"]
    if phone else phone_uuid)
    elif user_phone_type == "hot-desk":
    phone = phones.get(phone_uuid)
    print("        hot-desk phone:", phone["phone_username"]
    if phone else phone_uuid)
    elif user_phone_type == "number":
    print("        number:", number)
    elif user_phone_type == "trunk number":
    print("        number:", number, "on trunk:", trunk_uuid)
    elif user_phone_type == "forward":
    print("        forwarded to:", number)
    elif phone_uuid:
    phone = phones.get(phone_uuid)
    if phone:
    print("    phone:", phone["phone_username"])
    print("    unknown phone:", phone_uuid)
    elif number:
    print("    number:", target["number"])
    print("    unknown:", json.dumps(target))
    def dump_call(event, users, phones):
    print(" call", event["unique_id"])
    if "event_tag" in event:
    print("  event tag:", event["event_tag"])
    caller = event["caller"]
    if "user_uuid" in caller:
    user_uuid = caller["user_uuid"]
    user = users.get(user_uuid)
    if user:
    print("  from user:", get_user_name(user))
    print("  from unknown user:", user_uuid)
    if "phone_uuid" in caller:
    phone_uuid = caller["phone_uuid"]
    phone = phones.get(phone_uuid)
    if phone:
    if phone.get("name"):
    print("  from phone:", phone["name"])
    print("  from unnamed phone:", phone_uuid)
    print("  from unknown phone:", phone_uuid)
    if "normalized_number" in caller:
    print("  from number (normalized):", caller["normalized_number"])
    elif "number" in caller:
    print("  from number:", caller["number"])
    if "name" in caller:
    print("  from name:", caller["name"])
    if "called_did" in event:
    print("  to DID:", event["called_did"])
    if "called_extension" in event:
    print("  to extension:", event["called_extension"])
    if "called_number" in event:
    print("  to number:", event["called_number"])
    targets = event.get("targets")
    if targets is None:
    target = event.get("target")
    if target:
    print("  Target:")
    dump_target(target, users, phones)
    for i, target in enumerate(targets, 1):
    print("  Target #%i:" % (i,))
    dump_target(target, users, phones)
    def main():
    parser = argparse.ArgumentParser("Axeos PBX API example")
    parser.add_argument("--tls", "--ssl", action="store_true",
    help="use TLS")
    parser.add_argument("--disable-certificate-verification", action="store_true",
    help="disable certificate verification")
    parser.add_argument("--debug", action="store_true",
    help="Display debug output")
    help="PBX user name")
    help="PBX user GUI password")
    help="PBX API key")
    help="IP address or host name of a PBX")
    options = parser.parse_args()
    if options.disable_certificate_verification:
    if options.api_key:
    if options.user or options.password:
    parser.error("--api-key cannot be used with --user or --password")
    auth = ('apiKey', options.api_key)
    if not options.user:
    options.user = input("Username? > ")
    if not options.password:
    options.password = getpass("Password? > ")
    auth = (options.user, options.password)
    if options.tls:
    api_url = 'https://{0}/apis/pbx/'.format(options.pbx_address)
    api_url = 'http://{0}/apis/pbx/'.format(options.pbx_address)
    if options.debug:
    # flush old events
    events = load_events(api_url, auth, limit=1, timeout=0)
    if events:
    last_event = events[0]['seq']
    last_event = None
    # load user and phone database with current state
    users = load_users(api_url, auth)
    phones = load_phones(api_url, auth)
    dump_users_verbose(users, phones)
    while True:
    events = load_events(api_url, auth, 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, auth)
    dump_users(users, phones)
    elif event_type == 'phone list changed':
    print('Phone list changed')
    phones = load_phones(api_url, auth)
    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)
    phone = phones[phone_uuid]
    phone['status'] = event['status']
    phone['status_code'] = event['status_code']
    for user in list(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)
    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)
    user = users[user_uuid]
    user['phone'] = user['primary_phone']
    dump_user(user, phones)
    elif event_type == 'incoming call':
    print('incoming call:')
    dump_call(event, users, phones)
    elif event_type == 'api call':
    print('api call:')
    dump_call(event, users, phones)
    elif event_type == 'internal dial':
    print('internal dial:')
    dump_call(event, users, phones)
    elif event_type == 'call answered':
    print('call answered:')
    dump_call(event, users, phones)
    elif event_type == 'call finished':
    print('call finished (%s):' % (event["reason"],))
    dump_call(event, users, phones)
    print('unknown event:', event_type)
    if __name__ == "__main__":
    except KeyboardInterrupt:
    except IOError:

    in API
    Did this article answer your question?

    No Comments – be the first.

    Leave a Reply