Get WWBOTA data via SSE. Thanks to Steven M1SDH for the patch. Closes #4

This commit is contained in:
Ian Renton
2025-10-05 08:09:06 +01:00
parent 74153a9d94
commit c4aac4973d
5 changed files with 103 additions and 39 deletions

View File

@@ -1,48 +1,44 @@
import json
from datetime import datetime
from data.spot import Spot
from spotproviders.http_spot_provider import HTTPSpotProvider
from spotproviders.sse_spot_provider import SSESpotProvider
# Spot provider for Worldwide Bunkers on the Air
class WWBOTA(HTTPSpotProvider):
POLL_INTERVAL_SEC = 120
class WWBOTA(SSESpotProvider):
SPOTS_URL = "https://api.wwbota.org/spots/"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
super().__init__(provider_config, self.SPOTS_URL)
def http_response_to_spots(self, http_response):
new_spots = []
# Iterate through source data
for source_spot in http_response.json():
# Convert to our spot format. First we unpack references, because WWBOTA spots can have more than one for
# n-fer activations.
refs = []
ref_names = []
for ref in source_spot["references"]:
refs.append(ref["reference"])
ref_names.append(ref["name"])
spot = Spot(source=self.name,
dx_call=source_spot["call"].upper(),
de_call=source_spot["spotter"].upper(),
freq=float(source_spot["freq"]) * 1000000,
mode=source_spot["mode"].upper(),
comment=source_spot["comment"],
sig="WWBOTA",
sig_refs=refs,
sig_refs_names=ref_names,
icon="radiation",
time=datetime.fromisoformat(source_spot["time"]).timestamp(),
# WWBOTA spots can contain multiple references for bunkers being activated simultaneously. For
# now, we will just pick the first one to use as our grid, latitude and longitude.
grid=source_spot["references"][0]["locator"],
latitude=source_spot["references"][0]["lat"],
longitude=source_spot["references"][0]["long"],
qrt=source_spot["type"] == "QRT")
def sse_message_to_spot(self, message):
source_spot = json.loads(message)
# Convert to our spot format. First we unpack references, because WWBOTA spots can have more than one for
# n-fer activations.
refs = []
ref_names = []
for ref in source_spot["references"]:
refs.append(ref["reference"])
ref_names.append(ref["name"])
# Add to our list. Don't worry about de-duping, removing old spots etc. at this point; other code will do
# that for us. But WWBOTA does support a special "Test" spot type, we need to avoid adding that.
if source_spot["type"] != "Test":
new_spots.append(spot)
return new_spots
spot = Spot(source=self.name,
dx_call=source_spot["call"].upper(),
de_call=source_spot["spotter"].upper(),
freq=float(source_spot["freq"]) * 1000000,
mode=source_spot["mode"].upper(),
comment=source_spot["comment"],
sig="WWBOTA",
sig_refs=refs,
sig_refs_names=ref_names,
icon="radiation",
time=datetime.fromisoformat(source_spot["time"]).timestamp(),
# WWBOTA spots can contain multiple references for bunkers being activated simultaneously. For
# now, we will just pick the first one to use as our grid, latitude and longitude.
grid=source_spot["references"][0]["locator"],
latitude=source_spot["references"][0]["lat"],
longitude=source_spot["references"][0]["long"],
qrt=source_spot["type"] == "QRT")
# WWBOTA does support a special "Test" spot type, we need to avoid adding that.
return spot if source_spot["type"] != "Test" else None