Mercurial > personal > weatherlog
view weatherlog/daemon.py @ 17:39c0686e6765
daemon: Change to executing from a config file.
This updates daemon.py to execute based on a config file, rather than
needing manual configuration. It also instructs setup.py to create
an entry point for the new daemon.py main function.
Configuration is specified in a TOML file whose format is documented,
kind of, in the parse_config function.
author | Paul Fisher <paul@pfish.zone> |
---|---|
date | Thu, 17 Oct 2019 22:28:12 -0400 |
parents | c01f9929ae38 |
children | 7117db65715e |
line wrap: on
line source
"""Entry point to set up a temperature logging daemon.""" import argparse import enum import time import typing as t import attr import toml from . import http_writer from . import logger from . import reader DEFAULT_INTERVAL_SECS = 60 MIN_INTERVAL_SECS = 5 def run( rd: reader.Reader, log: logger.BufferedLogger, writer: logger.RemoteWriter, interval: int = DEFAULT_INTERVAL_SECS, ): """Sets up and runs a logger daemon.""" log.start() cycle = 0 start = time.time() try: while True: log.write(rd.read().as_dict()) cycle += 1 target = start + interval * cycle now = time.time() time.sleep(max(target - now, MIN_INTERVAL_SECS)) finally: log.close() class SensorType(enum.Enum): DHT22 = 'DHT22' BME280 = 'BME280' @attr.s(auto_attribs=True, frozen=True, slots=True) class _Config: # The directory to store reading data in. directory: str # The URL to submit readings to. url: str # The type of sensor to read from. sensor: SensorType # The authentication preamble for the URL. auth: t.Dict[str, object] def parse_config(config_file: str): with open(config_file, 'r') as infile: config = toml.load(infile) return _Config( directory=config['directory'], url=config['url'], sensor=SensorType(config['sensor']['type']), auth=config['auth']) def main(args: t.Optional[t.Sequence[str]] = None) -> None: parser = argparse.ArgumentParser() parser.add_argument('config_file', help='TOML file to load config from.') parsed = parser.parse_args(args) config = parse_config(parsed.config_file) writer = http_writer.HTTPWriter(config.url, config.auth) log = logger.BufferedLogger(config.directory, writer) if config.sensor is SensorType.DHT22: rdr = reader.DHT22Reader() elif config.sensor is SensorType.BME280: rdr = reader.BME280Reader() else: raise AssertionError('Unknown sensor type') run(rdr, log, writer) if __name__ == '__main__': main()