import json 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 serialize_everything 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): # 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 = {} 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_sources: list = list( map(lambda p: p["name"], filter(lambda p: p["enabled"], self._status_data["spot_providers"]))) alert_sources = list( map(lambda p: p["name"], filter(lambda p: p["enabled"], self._status_data["alert_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_sources.append("API") options = {"bands": BANDS, "modes": ALL_MODES, "mode_types": MODE_TYPES, "sigs": SIGS, "spot_sources": spot_sources, "alert_sources": alert_sources, "continents": CONTINENTS, "propagation_modes": PROPAGATION_MODES.values(), "max_spot_age": MAX_SPOT_AGE, "spot_allowed": ALLOW_SPOTTING, "spot_submit_providers": spot_submit_providers} self.write(json.dumps(options, default=serialize_everything)) self.set_status(200) self.set_header("Cache-Control", "no-store") self.set_header("Content-Type", "application/json")