view weather_server/server.py @ 14:dd77a7ee02c1

Create tiny AMD loader and set up tsc for it.
author Paul Fisher <paul@pfish.zone>
date Wed, 09 Oct 2019 23:14:16 -0400
parents 52ef21607b31
children 9a609bcf0809
line wrap: on
line source

import datetime
import hmac
import sys

import bson
import flask

from . import common
from . import locations
from . import types


def build_app(root_directory: str) -> flask.Flask:
    locs = locations.Locations(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, logger = locs.get(loc_name)
            if not hmac.compare_digest(password, loc.password):
                flask.abort(400)
            entries = [
                types.Reading.from_now(
                    sample_time=item['sample_time'],
                    temp_c=item['temp_c'],
                    rh_pct=item['rh_pct'],
                )
                for item in reader
            ]
        except (KeyError, bson.InvalidBSON):
            flask.abort(400)
        logger.write_rows(entries)
        return flask.jsonify({'status': 'OK'})

    @app.route('/<location>')
    def show(location: str):
        try:
            loc, logger = locs.get(location)
        except KeyError:
            flask.abort(404)
        data = logger.data
        if data:
            last_reading = data[-1]
            tz = loc.timezone()
            date = tz.normalize(last_reading.sample_time.astimezone(tz))
        else:
            last_reading = None
            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, logger = 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.as_dict() for r in logger.data
            if start < r.sample_time
        ]
        resp = flask.Response()
        resp.content_type = 'application/json'
        resp.data = common.json_dumps(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:])