1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859 |
- 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()
|