Mercurial > personal > weather-server
view weather_server/server.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 | 9a609bcf0809 |
children | 88249e451566 |
line wrap: on
line source
import datetime import hmac import pathlib import sys import bson import flask import pytz from . import common from . import locations def build_app(root_directory: str) -> flask.Flask: locs = locations.LocationFolder(pathlib.Path(root_directory)) app = flask.Flask(__name__) app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0 @app.route('/favicon.ico') def favicon(): return flask.send_file('static/favicon.ico') @app.route('/') def home(): return 'Weather server' @app.route('/_submit', methods=['POST']) def submit(): req = flask.request reader = bson.decode_file_iter( req.stream, codec_options=common.BSON_OPTIONS) try: preamble = next(reader) loc_name = preamble['location'] password = str(preamble['password']) loc = locs.get(loc_name) if not hmac.compare_digest(password, loc.password): flask.abort(400) entries = tuple(reader) except (KeyError, bson.InvalidBSON): flask.abort(400) now = datetime.datetime.now(tz=pytz.UTC) loc.record(entries, now) return flask.jsonify({'status': 'OK'}) @app.route('/<location>') def show(location: str): try: loc = locs.get(location) except KeyError: flask.abort(404) last_reading = loc.latest() if last_reading: tz = loc.timezone() date = tz.normalize(last_reading.sample_time.astimezone(tz)) else: date = None return flask.render_template( 'location.html', location=loc, last_reading=last_reading, date=date) @app.route('/<location>/recent') def recent(location: str): try: loc = locs.get(location) except KeyError: flask.abort(404) req = flask.request try: seconds = int(req.args['seconds']) except (KeyError, ValueError): flask.abort(400) start = common.utc_now() - datetime.timedelta(seconds=seconds) readings = [r for r in loc.entries if start < r['sample_time']] resp = flask.Response() resp.content_type = 'application/json' resp.data = common.json_dumps({ 'timezone': loc.tz_name, 'readings': readings, }) return resp return app def main(argv): """Main function for a simple local demo.""" app = build_app(argv[0]) app.run(host='0.0.0.0') if __name__ == '__main__': main(sys.argv[1:])