view weather_server/server.py @ 34:8d3f32455575

Bump version.
author Paul Fisher <paul@pfish.zone>
date Sat, 12 Jun 2021 20:24:30 +0000
parents 7def5611895b
children
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():
        locations = tuple(locs.locations().values())
        return flask.render_template(
            'index.html',
            locations=locations,
        )

    @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
        now = datetime.datetime.now(tz=pytz.UTC)
        diff = (now - date) if date else None
        is_recent = diff and diff < datetime.timedelta(hours=12)
        return flask.render_template(
            'location.html',
            location=loc,
            last_reading=last_reading,
            date=date,
            date_format=f'%H:%M' if is_recent else '%Y-%m-%d %H:%M')

    @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:])