Mercurial > personal > weatherlog
view weatherlog/daemon.py @ 20:92367b644e29
Remove support for DHT22 to remove native dependencies.
This removes the ability to read from the DHT22 sensor, since it
requires native code to run and will not compile on ARM64 (at least
for me, which is enough reason to remove it).
author | Paul Fisher <paul@pfish.zone> |
---|---|
date | Sat, 25 Dec 2021 00:02:03 +0000 |
parents | 7117db65715e |
children |
line wrap: on
line source
"""Entry point to set up a temperature logging daemon.""" import argparse import enum import signal import threading 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, ) -> None: """Sets up and runs a logger daemon.""" evt = threading.Event() signal.signal(signal.SIGTERM, lambda *args: evt.set()) cycle = 0 start = time.time() running = True log.start() try: while running: log.write(rd.read().as_dict()) cycle += 1 target = start + interval * cycle now = time.time() running = not evt.wait(max(target - now, MIN_INTERVAL_SECS)) finally: log.close() class SensorType(enum.Enum): BME280 = 'BME280' _SENSOR_FACTORIES = { SensorType.BME280: reader.BME280Reader, } @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) -> _Config: 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) try: factory = _SENSOR_FACTORIES[config.sensor] except KeyError: print(f'Invalid sensor type {config.sensor}.') rdr = factory() run(rdr, log, writer) if __name__ == '__main__': main()