59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
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()
|