mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-06-24 05:35:10 +00:00
104 lines
5.0 KiB
Python
104 lines
5.0 KiB
Python
from datetime import datetime
|
|
|
|
import requests
|
|
|
|
from core.constants import HTTP_HEADERS, SSB_SUB_MODES, DV_SUB_MODES
|
|
from data.sig_ref import SIGRef
|
|
from data.spot import Spot
|
|
from spotproviders.http_spot_provider import HTTPSpotProvider
|
|
|
|
|
|
class SOTA(HTTPSpotProvider):
|
|
"""Spot provider for Summits on the Air"""
|
|
|
|
POLL_INTERVAL_SEC = 120
|
|
# SOTA wants us to check for an "epoch" from the API and see if it's actually changed before querying the main data
|
|
# APIs. So it's actually the EPOCH_URL that we pass into the constructor and get the superclass to call on a timer.
|
|
# The actual data lookup all happens after parsing and checking the epoch.
|
|
EPOCH_URL = "https://api-db2.sota.org.uk/api/spots/epoch"
|
|
SPOTS_URL = "https://api-db2.sota.org.uk/api/spots/60/all/all"
|
|
# SOTA spots don't contain lat/lon, we need a separate lookup for that
|
|
SUMMIT_URL_ROOT = "https://api-db2.sota.org.uk/api/summits/"
|
|
|
|
SUBMIT_URL = "https://api-db2.sota.org.uk/api/spots"
|
|
|
|
VALID_MODES = ["AM", "CW", "Data", "DV", "FM", "SSB"]
|
|
|
|
def __init__(self, provider_config):
|
|
super().__init__(provider_config, self.EPOCH_URL, self.POLL_INTERVAL_SEC)
|
|
self._api_epoch = ""
|
|
|
|
def _http_response_to_spots(self, http_response):
|
|
# OK, source data is actually just the epoch at this point. We'll then go on to fetch real data if we know this
|
|
# has changed.
|
|
epoch_changed = http_response.text != self._api_epoch
|
|
self._api_epoch = http_response.text
|
|
|
|
new_spots = []
|
|
# OK, if the epoch actually changed, now we make the real request for data.
|
|
if epoch_changed:
|
|
source_data = requests.get(self.SPOTS_URL, headers=HTTP_HEADERS, timeout=(5, 30)).json()
|
|
# Iterate through source data
|
|
for source_spot in source_data:
|
|
# Convert to our spot format
|
|
spot = Spot(source=self.name,
|
|
source_id=source_spot["id"],
|
|
dx_call=source_spot["activatorCallsign"].upper(),
|
|
dx_name=source_spot["activatorName"],
|
|
de_call=source_spot["callsign"].upper(),
|
|
freq=(float(source_spot["frequency"]) * 1000000) if (
|
|
source_spot["frequency"] is not None) else None,
|
|
# Seen SOTA spots with no frequency!
|
|
mode=source_spot["mode"].upper(),
|
|
comment=source_spot["comments"],
|
|
sig="SOTA",
|
|
sig_refs=[SIGRef(id=source_spot["summitCode"], sig="SOTA", name=source_spot["summitName"],
|
|
activation_score=source_spot["points"])],
|
|
time=datetime.fromisoformat(source_spot["timeStamp"].replace("Z", "+00:00")).timestamp())
|
|
|
|
# Add to our list. Don't worry about de-duping, removing old spots etc. at this point; other code will do
|
|
# that for us.
|
|
new_spots.append(spot)
|
|
return new_spots
|
|
|
|
def can_submit_spot(self, sig):
|
|
return sig == "SOTA"
|
|
|
|
def submit_spot(self, spot, credentials):
|
|
# TODO test this method works
|
|
access_token = credentials.get("access_token", "")
|
|
id_token = credentials.get("id_token", "")
|
|
if not access_token or not id_token:
|
|
raise ValueError("SOTA API tokens are required. Please log into SOTA in order to spot to it.")
|
|
sig_ref = spot.sig_refs[0].id if spot.sig_refs else ""
|
|
if sig_ref:
|
|
# Split reference into association and summit codes
|
|
ref_split = sig_ref.split("/")
|
|
|
|
# Figure out a valid mode. Borrowed this from PoLo :)
|
|
# https://github.com/ham2k/app-polo/blob/main/src/extensions/activities/sota/SOTAPostSelfSpot.js
|
|
if spot.mode and spot.mode not in self.VALID_MODES:
|
|
if spot.mode in SSB_SUB_MODES:
|
|
spot.mode = "SSB"
|
|
elif spot.mode in DV_SUB_MODES:
|
|
spot.mode = "DV"
|
|
else:
|
|
spot.mode = "Data"
|
|
|
|
body = {
|
|
"activatorCallsign": spot.dx_call,
|
|
"associationCode": ref_split[0],
|
|
"summitCode": ref_split[1],
|
|
"frequency": str(spot.freq / 1000000.0),
|
|
"mode": spot.mode or "",
|
|
"posterCallsign": spot.de_call,
|
|
"comments": spot.comment or "",
|
|
"type": "TEST" # todo remove once testing complete
|
|
}
|
|
headers = {**HTTP_HEADERS, "Authorization": "bearer " + access_token, "id_token": id_token, "Content-Type": "application/json"}
|
|
response = requests.post(self.SUBMIT_URL, json=body, headers=headers, timeout=(5, 30))
|
|
if not response.ok:
|
|
raise RuntimeError("SOTA API returned " + str(response.status_code) + ": " + response.text)
|
|
else:
|
|
raise RuntimeError("Summit reference is required for submitting SOTA spots.")
|