asterisk-misc/asterisk_misc_common.py

import asyncio
import logging
import pathlib
from typing import Tuple

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"],
    )
    p.add_argument("--credentials", type=pathlib.Path, required=True)
    return p


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


def read_credentials(fpath) -> Tuple[str, str]:
    with fpath.open() as buf:
        username = buf.readline().rstrip("\n\r")
        password = buf.readline().rstrip("\n\r")
    return username, password


def get_manager(args):
    username, password = read_credentials(args.credentials)
    manager = Manager(
        loop=LOOP,
        host="127.0.0.1",
        port=5038,
        ssl=False,
        encoding="utf8",
        username=username,
        secret=password,
        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()