asterisk-misc/asterisk_misc_common.py

import asyncio
import logging

from panoramisk import Manager

LOOP = asyncio.get_event_loop()


def add_arguments(p):
    p.add_argument(
        "--log-level",
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
    )
    return p


def do_common_options(args):
    logging.basicConfig(level=logging.DEBUG)
    logging.getLogger("panoramisk.manager").setLevel(logging.WARNING)


def get_manager(args):
    manager = Manager(
        loop=LOOP,
        host="127.0.0.1",
        port=5038,
        ssl=False,
        encoding="utf8",
        username="ring",
        secret="campanellino",
        ping_delay=10,  # Delay after start
        ping_interval=10,  # Periodically ping AMI (dead or alive)
        reconnect_timeout=2,  # Timeout reconnect if connection lost
    )
    return manager


def run_manager(manager, args, on_startup=None, on_shutdown=None):
    manager.connect()
    try:
        if on_startup:
            manager.loop.run_until_complete(on_startup(manager))
        manager.loop.run_forever()
    except KeyboardInterrupt:
        if on_shutdown:
            manager.loop.run_until_complete(on_shutdown(manager))
        manager.loop.close()