import argparse
import asyncio
import logging
import sys
from pathlib import Path
from typing import List
import toml
from dacite.core import from_dict
from .monitor import Collector, Config
from .version import __version__
__author__ = "Constantine Evans"
__copyright__ = "Constantine Evans"
__license__ = "BSD-3-Clause"
_logger = logging.getLogger(__name__)
def parse_args(args: List[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="QuantStudio qPCR data monitor for InfluxDB")
parser.add_argument(
"--version",
action="version",
version="qs-monitor {ver}".format(ver=__version__),
)
parser.add_argument(dest="config", help="configuration file", type=Path, metavar="CONFIG_TOML")
parser.add_argument(
"-v",
"--verbose",
dest="loglevel",
help="set loglevel to INFO",
action="store_const",
const=logging.INFO,
)
parser.add_argument(
"-vv",
"--very-verbose",
dest="loglevel",
help="set loglevel to DEBUG",
action="store_const",
const=logging.DEBUG,
)
return parser.parse_args(args)
def setup_logging(loglevel):
logformat = "[%(asctime)s] %(levelname)s:%(name)s: %(message)s"
logging.basicConfig(level=loglevel, stream=sys.stdout, format=logformat, datefmt="%Y-%m-%dT%H:%M:%S")
def main(args: List[str]):
parsed_args = parse_args(args)
setup_logging(parsed_args.loglevel)
config = dict(toml.load(parsed_args.config))
collector = Collector(from_dict(Config, config))
asyncio.run(collector.reliable_monitor())
def run():
main(sys.argv[1:])
if __name__ == "__main__":
run()