spothole/server/webserver.py

import json
import logging
from datetime import datetime
from threading import Thread

import bottle
import pytz
from bottle import run, response

from core.utils import serialize_everything


# Provides the public-facing web server.
class WebServer:

    # Constructor
    def __init__(self, spot_list, status_data, port):
        self.last_page_access_time = None
        self.last_api_access_time = None
        self.spot_list = spot_list
        self.status_data = status_data
        self.port = port
        self.thread = Thread(target=self.run)
        self.thread.daemon = True
        self.status = "Starting"

        # Set up routing
        bottle.get("/api/spots")(self.serve_api_spots)
        bottle.get("/api/status")(self.serve_api_status)
        bottle.get("/")(self.serve_index)
        bottle.get("/apidocs")(self.serve_apidocs)
        bottle.get("/<filepath:path>")(self.serve_static_file)

    # Start the web server
    def start(self):
        self.thread.start()

    # Run the web server itself. This blocks until the server is shut down, so it runs in a separate thread.
    def run(self):
        logging.info("Starting web server on port " + str(self.port) + "...")
        self.status = "Waiting"
        run(host='localhost', port=self.port)

    # Main spots API
    def serve_api_spots(self):
        self.last_api_access_time = datetime.now(pytz.UTC)
        self.status = "OK"
        spots_json = json.dumps(self.get_spot_list_with_filters(bottle.request.query), default=serialize_everything)
        response.content_type = 'application/json'
        return spots_json

    # Server status API
    def serve_api_status(self):
        self.last_api_access_time = datetime.now(pytz.UTC)
        self.status = "OK"
        status_json = json.dumps(self.status_data, default=serialize_everything)
        response.content_type = 'application/json'
        return status_json

    # Serve the home page. This would be accessible as /index.html but we need this workaround to make it available as /
    def serve_index(self):
        self.last_page_access_time = datetime.now(pytz.UTC)
        self.status = "OK"
        return bottle.static_file("index.html", root="webassets")

    # Serve the API docs page. This would be accessible as /apidocs/index.html but we need this workaround to make it
    # available as /apidocs
    def serve_apidocs(self):
        self.last_page_access_time = datetime.now(pytz.UTC)
        self.status = "OK"
        return bottle.static_file("index.html", root="webassets/apidocs")

    # Serve general static files from "webassets" directory
    def serve_static_file(self, filepath):
        return bottle.static_file(filepath, root="webassets")

    # Utility method to apply filters to the overall spot list and return only a subset. Enables query parameters in
    # the main "spots" GET call. The "query" parameter should be the result of bottle's request.query, and is a MultiDict
    def get_spot_list_with_filters(self, query):
        # Create a shallow copy of the spot list, ordered by spot time. We'll then filter it accordingly.
        # We can filter by spot time and received time with "since" and "received_since", which take a UNIX timestamp
        # in seconds UTC.
        # We can also filter by source, sig, band, mode, dx_continent and de_continent. Each of these accepts a single
        # value or a comma-separated list.
        # We can provide a "limit" number as well. Spots are always returned newest-first; "limit" limits to only the
        # most recent X spots.
        spots = sorted(self.spot_list, key=lambda spot: spot.time, reverse=True)
        for k in query.keys():
            match k:
                case "since":
                    since = datetime.fromtimestamp(int(query.get(k)), pytz.UTC)
                    spots = [s for s in spots if s.time > since]
                case "received_since":
                    since = datetime.fromtimestamp(int(query.get(k)), pytz.UTC)
                    spots = [s for s in spots if s.received_time > since]
                case "source":
                    sources = query.get(k).split(",")
                    spots = [s for s in spots if s.source in sources]
                case "sig":
                    sigs = query.get(k).split(",")
                    spots = [s for s in spots if s.sig in sigs]
                case "band":
                    bands = query.get(k).split(",")
                    spots = [s for s in spots if s.band in bands]
                case "mode":
                    modes = query.get(k).split(",")
                    spots = [s for s in spots if s.mode in modes]
                case "mode_family":
                    mode_families = query.get(k).split(",")
                    spots = [s for s in spots if s.mode_family in mode_families]
                case "dx_continent":
                    dxconts = query.get(k).split(",")
                    spots = [s for s in spots if s.dx_continent in dxconts]
                case "de_continent":
                    deconts = query.get(k).split(",")
                    spots = [s for s in spots if s.de_continent in deconts]
        # If we have a "limit" parameter, we apply that last, regardless of where it appeared in the list of keys.
        if "limit" in query.keys():
            spots = spots[:int(query.get("limit"))]
        return spots