main_panoramisk.py

import logging
from pprint import pprint
import asyncio
from panoramisk import Manager
import coloredlogs


def get_logger(logger_name: str, level: int) -> logging:
    # create logger for prd_ci
    logger = logging.getLogger(logger_name)
    logger.setLevel(level)
    # create formatter and add it to the handlers
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    # create file handler for logger.
    file_handler = logging.FileHandler('%s.log' % logger_name)
    file_handler.setLevel(level=logging.DEBUG)
    file_handler.setFormatter(formatter)
    # create console handler for logger.
    console_handler = logging.StreamHandler()
    console_handler.setLevel(level=logging.DEBUG)
    console_handler.setFormatter(formatter)
    # add handlers to logger.
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger


log = get_logger('panoramisk', logging.DEBUG)
coloredlogs.install(level=logging.DEBUG, logger=log)


async def extension_status():
    manager = Manager(loop=asyncio.get_event_loop(),
                      host='127.0.0.1', port=5038,
                      username='odoo', secret='odoo')
    await manager.connect()

    sip_show_peer = await manager.send_action({'Action': 'SIPShowpeer', 'Peer': '1001'})
    log.info(sip_show_peer)

    core_show_channels = await manager.send_action({'Action': 'CoreShowChannels'})
    log.info(f'CoreShowChannels - {str(core_show_channels)}')

    agi = await manager.send_action({'Action': 'AGI', 'Command': 'Answer', 'Channel': 'SIP/1001-00000000'})
    log.info(f'AGI - {str(agi)}')

    manager.close()


def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    loop.run_until_complete(extension_status())
    loop.close()


if __name__ == '__main__':
    main()