Mercurial > personal > weather-server
diff weather_server/locations.py @ 21:beb42c835c52
Make weather server handle arbitrary data:
- Make logfile record arbitrary BSONs
- Make server handlers OK with same
- Make location type a normal class rather than attrs;
have it handle its own logger.
- Bump version number.
author | Paul Fisher <paul@pfish.zone> |
---|---|
date | Sat, 19 Oct 2019 18:40:48 -0400 |
parents | f1ea183d28ba |
children | e229afdd447b |
line wrap: on
line diff
--- a/weather_server/locations.py Sun Oct 13 18:44:12 2019 -0400 +++ b/weather_server/locations.py Sat Oct 19 18:40:48 2019 -0400 @@ -1,53 +1,95 @@ """Manages the directory containing the various logs.""" import configparser +import datetime import pathlib import typing as t -import attr import pytz from . import logfile from . import types - CONFIG_FILE = 'config.ini' LOG = 'log.bson' -@attr.s(frozen=True, slots=True) -class LocationInfo: - name = attr.ib(type=str) - tz_name = attr.ib(type=str) - password = attr.ib(type=str) +class ConfigError(Exception): + """Raised when a location can't be loaded.""" + + +class Location: + + def __init__(self, root: pathlib.Path): + parser = configparser.ConfigParser(interpolation=None) + self.root = root + config_file = root / CONFIG_FILE + try: + with open(config_file, 'r') as infile: + parser.read_file(infile) + self.location = parser.get( + 'location', 'name', fallback='Weather station') + self.tz_name = parser.get('location', 'timezone', fallback='UTC') + self.password = parser.get('location', 'password') + self.logger = logfile.Logger.create( + str(root / LOG), sample_field='sample_time') + except (IOError, KeyError, configparser.Error): + raise ConfigError("Couldn't load location info.") - def timezone(self): + def record( + self, + entries: t.Iterable[t.Dict[str, object]], + timestamp: datetime.datetime, + ) -> None: + for e in entries: + e['ingest_time'] = timestamp + self.logger.write_rows(entries) + + @property + def entries(self) -> t.Iterable[t.Dict[str, object]]: + return self.logger.data + + def latest(self) -> t.Optional[types.Reading]: + most_recent = reversed(self.logger.data) + for entry in most_recent: + try: + return types.Reading.from_dict(entry) + except KeyError: + pass # go to the older one. + return None + + def timezone(self) -> datetime.tzinfo: try: return pytz.timezone(self.tz_name) except pytz.UnknownTimeZoneError: return pytz.UTC - @classmethod - def load(cls, config_file: pathlib.Path) -> 'LocationInfo': - parser = configparser.ConfigParser(interpolation=None) - parser.read(config_file) - return LocationInfo( - name=parser.get('location', 'name', fallback='Weather station'), - tz_name=parser.get('location', 'timezone', fallback='UTC'), - password=parser.get('location', 'password')) + def __repr__(self) -> str: + return '<Location in %r>'.format(self.root) -class Locations: - def __init__(self, base: str): - self._path = pathlib.Path(base) +class LocationFolder: - def paths(self) -> t.Tuple[str, ...]: - return tuple(sorted(f.name for f in self._path.iterdir())) + def __init__(self, root: pathlib.Path): + self.root = root + # locations, mtime + self.info: t.Tuple[t.Dict[str, Location], t.Optional[int]] = ({}, None) + self._maybe_reload() + + def get(self, name: str) -> Location: + self._maybe_reload() + locs, _ = self.info + return locs[name] - def get(self, name) -> t.Tuple[LocationInfo, logfile.Logger]: - try: - directory = self._path / name - logger = logfile.Logger.create(str(directory / LOG)) - return (LocationInfo.load(directory / CONFIG_FILE), logger) - except OSError: - raise KeyError(name) + def _maybe_reload(self) -> None: + new_mtime = self.root.stat().st_mtime_ns + _, old_mtime = self.info + if old_mtime == new_mtime: + return + locations = {} + for child in self.root.iterdir(): + try: + locations[child.name] = Location(child) + except ConfigError: + pass # It's OK. Skip this. + self.info = locations, new_mtime