w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-20 22:37:44 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Giant refactor to rebrand "SIG" as "Activity" anywhere that doesn't touch API or config file (which is to be addressed in a future breaking change). #147

Browse Source
This commit is contained in:
Ian Renton
2026-09-18 14:54:11 +01:00
parent 556ea56378
commit 81cd686a00
96 changed files with 1208 additions and 1170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/spot/gma.py
+14 -14
View File
@@ -4,9 +4,9 @@ from datetime import datetime
import pytz
from core.constants import HTTP_HEADERS
from core.enums import Mode, SIGRefType
from core.enums import ActivityRefType, Mode
from core.url_data_cache import URLDataCache
from data.sig_ref import SIGRef
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -69,7 +69,7 @@ class GMA(HTTPSpotProvider):
mode=Mode.from_name(source_spot["MODE"].upper()) if "<>" not in source_spot["MODE"] else None,
comment=source_spot["TEXT"],
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["REF"],
sig="",
name=source_spot["NAME"],
@@ -83,7 +83,7 @@ class GMA(HTTPSpotProvider):
qrt=source_spot["QRG"] == "QRT",
)
# GMA doesn't give what programme (SIG) the reference is for until we separately look it up.
# GMA doesn't give what programme (activity) the reference is for until we separately look it up.
if "REF" in source_spot:
try:
ref_response = self._url_data_cache.get(
@@ -114,31 +114,31 @@ class GMA(HTTPSpotProvider):
match ref_info["reftype"]:
case "Summit":
spot.sig_refs[0].sig = "GMA"
spot.sig_refs[0].ref_type = SIGRefType.SUMMIT
spot.sig_refs[0].ref_type = ActivityRefType.SUMMIT
spot.sig = "GMA"
case "IOTA Island":
spot.sig_refs[0].sig = "IOTA"
spot.sig_refs[0].ref_type = SIGRefType.ISLAND
spot.sig_refs[0].ref_type = ActivityRefType.ISLAND
spot.sig = "IOTA"
case "GMA Island":
spot.sig_refs[0].sig = "GMA Islands"
spot.sig_refs[0].ref_type = SIGRefType.ISLAND
spot.sig_refs[0].ref_type = ActivityRefType.ISLAND
spot.sig = "GMA Islands"
case "Lighthouse (ILLW)":
spot.sig_refs[0].sig = "ILLW"
spot.sig_refs[0].ref_type = SIGRefType.LIGHTHOUSE
spot.sig_refs[0].ref_type = ActivityRefType.LIGHTHOUSE
spot.sig = "ILLW"
case "Lighthouse (ARLHS)":
spot.sig_refs[0].sig = "ARLHS"
spot.sig_refs[0].ref_type = SIGRefType.LIGHTHOUSE
spot.sig_refs[0].ref_type = ActivityRefType.LIGHTHOUSE
spot.sig = "ARLHS"
case "Castle":
spot.sig_refs[0].sig = "WCA"
spot.sig_refs[0].ref_type = SIGRefType.CASTLE
spot.sig_refs[0].ref_type = ActivityRefType.CASTLE
spot.sig = "WCA"
case "Mill":
spot.sig_refs[0].sig = "MOTA"
spot.sig_refs[0].ref_type = SIGRefType.MILL
spot.sig_refs[0].ref_type = ActivityRefType.MILL
spot.sig = "MOTA"
case _:
logger.warning(
@@ -158,7 +158,7 @@ class GMA(HTTPSpotProvider):
)
except Exception:
logger.exception(
f"Exception when looking up {self.REF_INFO_URL_ROOT}{source_spot['REF']}, SIG data will not be populated for the spot."
f"Exception when looking up {self.REF_INFO_URL_ROOT}{source_spot['REF']}, activity data will not be populated for the spot."
)
# Add to our list. Don't worry about de-duping, removing old spots etc. at this point;
@@ -169,8 +169,8 @@ class GMA(HTTPSpotProvider):
return new_spots
def can_submit_spot(self, sig):
return sig == "GMA"
def can_submit_spot(self, activity):
return activity == "GMA"
def submit_spot(self, spot, credentials):
# TODO: Implement.
providers/spot/hema.py
+6 -6
View File
@@ -7,8 +7,8 @@ import requests
from requests.exceptions import ConnectionError, ConnectTimeout, ReadTimeout
from core.constants import HTTP_HEADERS
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -64,13 +64,13 @@ class HEMA(HTTPSpotProvider):
comment=spotter_comment_match.group(2),
sig="HEMA",
sig_refs=[
SIGRef(
ActivityRef(
id=spot_items[3].upper(),
sig="HEMA",
name=spot_items[4],
latitude=float(spot_items[7]),
longitude=float(spot_items[8]),
ref_type=SIGRefType.SUMMIT,
ref_type=ActivityRefType.SUMMIT,
)
],
time=datetime.strptime(spot_items[0], "%d/%m/%Y %H:%M")
@@ -89,8 +89,8 @@ class HEMA(HTTPSpotProvider):
logger.warning("Connection error when accessing HEMA spots API.")
return new_spots
def can_submit_spot(self, sig):
return sig == "HEMA"
def can_submit_spot(self, activity):
return activity == "HEMA"
def submit_spot(self, spot, credentials):
# TODO: Implement. Currently blocked awaiting their API team to make a change to allow us to spot with a
providers/spot/llota.py
+4 -4
View File
@@ -1,7 +1,7 @@
from datetime import datetime
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -36,11 +36,11 @@ class LLOTA(HTTPSpotProvider):
comment=comment,
sig="LLOTA",
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["reference"],
sig="LLOTA",
name=source_spot["reference_name"],
ref_type=SIGRefType.LAKE,
ref_type=ActivityRefType.LAKE,
)
],
time=datetime.fromisoformat(source_spot["updated_at"].replace("Z", "+00:00")).timestamp(),
providers/spot/parksnpeaks.py
+18 -18
View File
@@ -7,7 +7,7 @@ import requests
from core.constants import HTTP_HEADERS
from core.enums import Mode
from data.sig_ref import SIGRef
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -20,7 +20,7 @@ class ParksNPeaks(HTTPSpotProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://www.parksnpeaks.org/api/ALL"
SUBMIT_URL = "https://www.parksnpeaks.org/api/SPOT/"
SUBMITTABLE_SIGS = [
SUBMITTABLE_ACTIVITIES = [
"POTA",
"SOTA",
"WWFF",
@@ -64,26 +64,26 @@ class ParksNPeaks(HTTPSpotProvider):
if not spot.de_call and m:
spot.de_call = str(m.group(1))
# Record SIG information. Sometimes we get a "SIG" of "QRP", which we ignore as it's not a programme with a
# defined set of references
sig = source_spot["actClass"].upper()
sig_ref = source_spot["actSiteID"]
if sig and sig != "" and sig != "QRP" and sig_ref and sig_ref != "":
spot.sig = sig
sig_refs = [
SIGRef(
# Record activity information. Sometimes we get an activity of "QRP", which we ignore as it's not a
# programme with a defined set of references
activity = source_spot["actClass"].upper()
ref_id = source_spot["actSiteID"]
if activity and activity != "" and activity != "QRP" and ref_id and ref_id != "":
spot.sig = activity
activity_refs = [
ActivityRef(
id=source_spot["actSiteID"],
sig=source_spot["actClass"].upper(),
)
]
spot.sig_refs = sig_refs
spot.sig_refs = activity_refs
# Free text location is not present in all spots, so only add it if it's set
if "actLocation" in source_spot and source_spot["actLocation"] != "":
sig_refs[0].name = source_spot["actLocation"]
activity_refs[0].name = source_spot["actLocation"]
# Log a warning for the developer if PnP gives us an unknown programme we've never seen before
if sig not in [
if activity not in [
"POTA",
"SOTA",
"WWFF",
@@ -94,14 +94,14 @@ class ParksNPeaks(HTTPSpotProvider):
"SANPCPA",
"LLOTA",
]:
logger.warning(f"PNP spot found with sig {sig}, developer needs to add support for this!")
logger.warning(f"PNP spot found with activity {activity}, developer needs to add support for this!")
# Add new spot to the list
new_spots.append(spot)
return new_spots
def can_submit_spot(self, sig):
return sig in self.SUBMITTABLE_SIGS
def can_submit_spot(self, activity):
return activity in self.SUBMITTABLE_ACTIVITIES
def submit_spot(self, spot, credentials):
# TODO test this works
@@ -111,11 +111,11 @@ class ParksNPeaks(HTTPSpotProvider):
raise ValueError(
"Parks N Peaks user ID and API key are required. Get yours from your Parks N Peaks account."
)
sig_ref = spot.sig_refs[0].id if spot.sig_refs else ""
ref_id = spot.sig_refs[0].id if spot.sig_refs else ""
body = {
"actClass": spot.sig or "",
"actCallsign": spot.dx_call,
"actSite": sig_ref,
"actSite": ref_id,
"mode": spot.mode or "",
"freq": str(spot.freq / 1000000.0),
"comments": spot.comment or "",
providers/spot/pota.py
+6 -6
View File
@@ -4,8 +4,8 @@ import pytz
import requests
from core.constants import HTTP_HEADERS
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -35,13 +35,13 @@ class POTA(HTTPSpotProvider):
comment=source_spot["comments"],
sig="POTA",
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["reference"],
sig="POTA",
name=source_spot["name"],
latitude=source_spot["latitude"],
longitude=source_spot["longitude"],
ref_type=SIGRefType.PARK
ref_type=ActivityRefType.PARK
)
],
time=datetime.strptime(source_spot["spotTime"], "%Y-%m-%dT%H:%M:%S")
@@ -57,8 +57,8 @@ class POTA(HTTPSpotProvider):
new_spots.append(spot)
return new_spots
def can_submit_spot(self, sig):
return sig == "POTA"
def can_submit_spot(self, activity):
return activity == "POTA"
def submit_spot(self, spot, credentials):
sig_ref = spot.sig_refs[0].id if spot.sig_refs else None
providers/spot/sota.py
+6 -6
View File
@@ -5,8 +5,8 @@ import requests
from requests.exceptions import ConnectionError, ConnectTimeout, ReadTimeout
from core.constants import HTTP_HEADERS
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -58,14 +58,14 @@ class SOTA(HTTPSpotProvider):
comment=source_spot["comments"],
sig="SOTA",
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["summitCode"],
sig="SOTA",
name=source_spot["summitName"],
latitude=source_spot["latitude"],
longitude=source_spot["longitude"],
activation_score=source_spot["points"],
ref_type=SIGRefType.SUMMIT,
ref_type=ActivityRefType.SUMMIT,
)
],
dx_latitude=source_spot["latitude"],
@@ -82,8 +82,8 @@ class SOTA(HTTPSpotProvider):
logger.warning("Timeout when accessing SOTA spots API.")
return new_spots
def can_submit_spot(self, sig):
return sig == "SOTA"
def can_submit_spot(self, activity):
return activity == "SOTA"
def submit_spot(self, spot, credentials):
# TODO test this method works
providers/spot/spot_provider.py
+2 -2
View File
@@ -60,8 +60,8 @@ class SpotProvider:
raise NotImplementedError("Subclasses must implement this method")
def can_submit_spot(self, sig):
"""Return True if this provider supports submitting spots upstream for the given SIG."""
def can_submit_spot(self, activity):
"""Return True if this provider supports submitting spots upstream for the given activity."""
return False
providers/spot/tiles.py
+7 -7
View File
@@ -4,8 +4,8 @@ from datetime import datetime
import requests
from core.constants import HTTP_HEADERS
from core.enums import LocationSourceForSpot, Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, LocationSourceForSpot, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -60,15 +60,15 @@ class Tiles(HTTPSpotProvider):
comment=source_spot["notes"],
sig="Tiles",
# Tiles spots can include POTA & SOTA references, but ignore those on the basis that we will get them separately from the POTA/SOTA providers anyway.
# Just take the grid reference itself as the single Tiles SIG reference.
# Just take the grid reference itself as the single Tiles activity reference.
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["maidenhead_grid"],
sig="Tiles",
name=source_spot["maidenhead_grid"],
latitude=source_spot["latitude"],
longitude=source_spot["longitude"],
ref_type=SIGRefType.GRID
ref_type=ActivityRefType.GRID
)
],
time=datetime.fromisoformat(source_spot["created_at"].replace("Z", "+00:00")).timestamp(),
@@ -83,8 +83,8 @@ class Tiles(HTTPSpotProvider):
new_spots.append(spot)
return new_spots
def can_submit_spot(self, sig):
return sig == "Tiles"
def can_submit_spot(self, activity):
return activity == "Tiles"
def submit_spot(self, spot, credentials):
# Tiles on the air currently only supports *self* spots
providers/spot/towers.py
+3 -3
View File
@@ -3,8 +3,8 @@ from datetime import datetime
import pytz
from core.enums import SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -35,7 +35,7 @@ class Towers(HTTPSpotProvider):
freq=likely_freq,
comment=source_spot["comment"],
sig="Towers",
sig_refs=[SIGRef(id=source_spot["ref"], sig="Towers", ref_type=SIGRefType.TOWER)],
sig_refs=[ActivityRef(id=source_spot["ref"], sig="Towers", ref_type=ActivityRefType.TOWER)],
time=datetime.strptime(response_json["updated"][:10] + source_spot["time"], "%Y-%m-%d%H:%M")
.replace(tzinfo=pytz.utc)
.timestamp(),
providers/spot/wota.py
+5 -5
View File
@@ -8,8 +8,8 @@ import pytz
from rss_parser import Parser
from rss_parser.models.rss import RSS
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -91,7 +91,7 @@ class WOTA(HTTPSpotProvider):
mode=Mode.from_name(mode),
comment=comment,
sig="WOTA",
sig_refs=[SIGRef(id=ref, sig="WOTA", name=ref_name, ref_type=SIGRefType.SUMMIT)] if ref else [],
sig_refs=[ActivityRef(id=ref, sig="WOTA", name=ref_name, ref_type=ActivityRefType.SUMMIT)] if ref else [],
time=time.timestamp(),
)
@@ -104,8 +104,8 @@ class WOTA(HTTPSpotProvider):
return new_spots
def can_submit_spot(self, sig):
return sig == "WOTA"
def can_submit_spot(self, activity):
return activity == "WOTA"
def submit_spot(self, spot, credentials):
# TODO Ask M5TEA if he's happy to share how this is done from his app
providers/spot/wwbota.py
+7 -7
View File
@@ -1,8 +1,8 @@
import json
from datetime import datetime
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.sse_spot_provider import SSESpotProvider
@@ -21,15 +21,15 @@ class WWBOTA(SSESpotProvider):
# n-fer activations.
refs = []
for ref in source_spot["references"]:
sigref = SIGRef(
activity_ref = ActivityRef(
id=ref["reference"],
sig="WWBOTA",
name=ref["name"],
latitude=ref["lat"],
longitude=ref["long"],
ref_type=SIGRefType.BUNKER,
ref_type=ActivityRefType.BUNKER,
)
refs.append(sigref)
refs.append(activity_ref)
spot = Spot(
source=self.name,
@@ -52,8 +52,8 @@ class WWBOTA(SSESpotProvider):
# WWBOTA does support a special "Test" spot type, we need to avoid adding that.
return spot if source_spot["type"] != "Test" else None
def can_submit_spot(self, sig):
return sig == "WWBOTA"
def can_submit_spot(self, activity):
return activity == "WWBOTA"
def submit_spot(self, spot, credentials):
# TODO: Implement. WWBOTA API docs cover this: https://api.wwbota.org/#tag/Spots/operation/create_spot_spots__post
providers/spot/wwff.py
+6 -6
View File
@@ -2,8 +2,8 @@ from datetime import datetime
import pytz
from core.enums import Mode, SIGRefType
from data.sig_ref import SIGRef
from core.enums import ActivityRefType, Mode
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -32,13 +32,13 @@ class WWFF(HTTPSpotProvider):
comment=source_spot["remarks"],
sig="WWFF",
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["reference"],
sig="WWFF",
name=source_spot["reference_name"],
latitude=source_spot["latitude"],
longitude=source_spot["longitude"],
ref_type=SIGRefType.PARK
ref_type=ActivityRefType.PARK
)
],
time=datetime.fromtimestamp(source_spot["spot_time"], tz=pytz.UTC).timestamp(),
@@ -51,8 +51,8 @@ class WWFF(HTTPSpotProvider):
new_spots.append(spot)
return new_spots
def can_submit_spot(self, sig):
return sig == "WWFF"
def can_submit_spot(self, activity):
return activity == "WWFF"
def submit_spot(self, spot, credentials):
# TODO: Implement. Spotting to WWFF should be possible, need to look up the Spotline docs or copy approach from
providers/spot/xota.py
+13 -13
View File
@@ -4,43 +4,43 @@ from datetime import datetime
import pytz
from core.enums import Mode
from data.sig_ref import SIGRef
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.websocket_spot_provider import WebsocketSpotProvider
class XOTA(WebsocketSpotProvider):
"""Spot provider for servers based on the "xOTA" software at https://github.com/nischu/xOTA/
The provider typically doesn't give us a lat/lon or SIG explicitly, so our own config provides a SIG which we can
then use for lookups. This functionality is implemented for Toilets on the Air events, of which there are
several - so a plain lookup of a "TOTA reference" doesn't make sense, it depends on which TOTA, which is why we also
provide a sig_ref_prefix in our config. This is applied to the reference ID, so e.g. "T-01" at C3 might become
"C3 T-01". This allows us to provide location lookups for TOTA at several conferences."""
The provider typically doesn't give us a lat/lon or activity explicitly, so our own config provides an activity
which we can then use for lookups. This functionality is implemented for Toilets on the Air events, of which
there are several - so a plain lookup of a "TOTA reference" doesn't make sense, it depends on which TOTA, which
is why we also provide a sig_ref_prefix in our config. This is applied to the reference ID, so e.g. "T-01" at C3
might become "C3 T-01". This allows us to provide location lookups for TOTA at several conferences."""
LOCATION_DATA = {}
SIG = None
ACTIVITY = None
def __init__(self, provider_config):
name = provider_config.get("name", "xOTA")
super().__init__(name, provider_config, provider_config["url"])
self.SIG = str(provider_config["sig"]) if "sig" in provider_config else None
self._sig_ref_prefix = str(provider_config["sig_ref_prefix"]) if "sig_ref_prefix" in provider_config else ""
self.ACTIVITY = str(provider_config["sig"]) if "sig" in provider_config else None
self._activity_ref_prefix = str(provider_config["sig_ref_prefix"]) if "sig_ref_prefix" in provider_config else ""
def _ws_message_to_spot(self, b):
string = b.decode("utf-8")
source_spot = json.loads(string)
ref_id = f"{self._sig_ref_prefix} {source_spot['reference']['title']}"
ref_id = f"{self._activity_ref_prefix} {source_spot['reference']['title']}"
spot = Spot(
source=self.name,
source_id=source_spot["id"],
dx_call=source_spot["stationCallSign"].upper(),
freq=float(source_spot["freq"]) * 1000,
mode=Mode.from_name(source_spot["mode"].upper()),
sig=self.SIG,
sig=self.ACTIVITY,
sig_refs=[
SIGRef(
ActivityRef(
id=ref_id,
sig=self.SIG or "",
sig=self.ACTIVITY or "",
url=source_spot["reference"]["website"],
)
],
providers/spot/zlota.py
+4 -4
View File
@@ -3,7 +3,7 @@ from datetime import datetime
import pytz
from core.enums import Mode
from data.sig_ref import SIGRef
from data.activity_ref import ActivityRef
from data.spot import Spot
from providers.spot.http_spot_provider import HTTPSpotProvider
@@ -37,7 +37,7 @@ class ZLOTA(HTTPSpotProvider):
comment=source_spot["comments"],
sig="ZLOTA",
sig_refs=[
SIGRef(
ActivityRef(
id=source_spot["reference"],
sig="ZLOTA",
name=source_spot["name"],
@@ -51,8 +51,8 @@ class ZLOTA(HTTPSpotProvider):
new_spots.append(spot)
return new_spots
def can_submit_spot(self, sig):
return sig == "ZLOTA"
def can_submit_spot(self, activity):
return activity == "ZLOTA"
def submit_spot(self, spot, credentials):
# TODO: Implement. Spotting to ZLOTA is supported via POST, see https://ontheair.nz/api