mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-08-10 09:21:41 +00:00
84 lines
3.8 KiB
Python
84 lines
3.8 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import pytz
|
|
import tornado
|
|
from tornado import httputil
|
|
from tornado.web import Application
|
|
|
|
from core.config import MAX_SPOT_AGE, ALLOW_SPOTTING
|
|
from core.constants import BANDS, ALL_MODES, MODE_TYPES, SIGS, CONTINENTS, PROPAGATION_MODES
|
|
from core.prometheus_metrics_handler import api_requests_counter
|
|
from core.utils import safe_json_dumps
|
|
|
|
|
|
class APIOptionsHandler(tornado.web.RequestHandler):
|
|
"""API request handler for /api/v2/options"""
|
|
|
|
def __init__(self, application: "Application", request: httputil.HTTPServerRequest, **kwargs: Any):
|
|
self._status_data = None
|
|
self._web_server_metrics = None
|
|
self._spot_providers = None
|
|
super().__init__(application, request, **kwargs)
|
|
|
|
def initialize(self, status_data, web_server_metrics, spot_providers=None):
|
|
self._status_data = status_data
|
|
self._web_server_metrics = web_server_metrics
|
|
self._spot_providers = spot_providers or []
|
|
|
|
def get(self):
|
|
try:
|
|
# Metrics
|
|
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
|
|
self._web_server_metrics["api_access_counter"] += 1
|
|
self._web_server_metrics["status"] = "OK"
|
|
api_requests_counter.inc()
|
|
|
|
# Build a map of SIG name -> list of provider names that can submit spots for that SIG
|
|
spot_submit_providers = {}
|
|
|
|
# Spothole v2.0 - disable this for now, API changes are in but this functionality is not ready yet. TODO
|
|
# for provider in self._spot_providers:
|
|
# if not provider.enabled:
|
|
# continue
|
|
# for sig in SIGS:
|
|
# if provider.can_submit_spot(sig.name):
|
|
# spot_submit_providers.setdefault(sig.name, []).append(provider.name)
|
|
|
|
# Spot/alert sources are filtered for only ones that are enabled in config, no point letting the user toggle
|
|
# things that aren't even available.
|
|
spot_providers: list = list(
|
|
map(lambda p: p["name"], filter(lambda p: p["enabled"], self._status_data["spot_providers"])))
|
|
alert_providers = list(
|
|
map(lambda p: p["name"], filter(lambda p: p["enabled"], self._status_data["alert_providers"])))
|
|
callsign_data_providers = list(
|
|
map(lambda p: p["name"], filter(lambda p: p["enabled"], self._status_data["callsign_data_providers"])))
|
|
# If spotting to this server is enabled, "API" is another valid spot source even though it does not come from
|
|
# one of our providers.
|
|
if ALLOW_SPOTTING:
|
|
spot_providers.append("API")
|
|
|
|
options = {"bands": BANDS,
|
|
"modes": ALL_MODES,
|
|
"mode_types": MODE_TYPES,
|
|
"sigs": SIGS,
|
|
"spot_providers": spot_providers,
|
|
"alert_providers": alert_providers,
|
|
"callsign_data_providers": callsign_data_providers,
|
|
"continents": CONTINENTS,
|
|
"propagation_modes": list(PROPAGATION_MODES.values()),
|
|
"max_spot_age": MAX_SPOT_AGE,
|
|
"spot_allowed": ALLOW_SPOTTING,
|
|
"spot_submit_providers": spot_submit_providers}
|
|
|
|
self.write(safe_json_dumps(options))
|
|
self.set_status(200)
|
|
self.set_header("Cache-Control", "no-store")
|
|
self.set_header("Content-Type", "application/json")
|
|
|
|
except Exception as e:
|
|
logging.error("Exception when handling client request to options API: %s", e, exc_info=True)
|
|
self.write(safe_json_dumps("Error - an internal server error occurred."))
|
|
self.set_status(500)
|