w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2025-10-27 16:59:25 +00:00
Code Issues 11 Packages Projects Releases Wiki Activity

Instantiate but disable providers. Closes #16

Browse Source
This commit is contained in:
Ian Renton
2025-10-02 09:38:05 +01:00
parent 9e495a3fae
commit 9640c0e0c1
18 changed files with 134 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
providers/aprsis.py
View File

@@ -13,15 +13,12 @@ from providers.provider import Provider
# Provider for the APRS-IS.
class APRSIS(Provider):
def __init__(self):
super().__init__()
def __init__(self, provider_config):
super().__init__(provider_config)
self.thread = Thread(target=self.connect)
self.thread.daemon = True
self.aprsis = None
def name(self):
return "APRS-IS"
def start(self):
self.thread.start()

13
providers/dxcluster.py
View File

@@ -21,18 +21,15 @@ class DXCluster(Provider):
re.IGNORECASE)
# Constructor requires hostname and port
def __init__(self, hostname, port):
super().__init__()
self.hostname = hostname
self.port = port
def __init__(self, provider_config):
super().__init__(provider_config)
self.hostname = provider_config["host"]
self.port = provider_config["port"]
self.telnet = None
self.thread = Thread(target=self.handle)
self.thread.daemon = True
self.run = True
def name(self):
return "DX Cluster " + self.hostname
def start(self):
self.thread.start()
@@ -67,7 +64,7 @@ class DXCluster(Provider):
if match:
spot_time = datetime.strptime(match.group(5), "%H%MZ")
spot_datetime = datetime.combine(datetime.today(), spot_time.time()).replace(tzinfo=pytz.UTC)
spot = Spot(source="Cluster",
spot = Spot(source=self.name,
dx_call=match.group(3),
de_call=match.group(1),
freq=float(match.group(2)),

9
providers/gma.py
View File

@@ -16,18 +16,15 @@ class GMA(HTTPProvider):
REF_INFO_CACHE_TIME_DAYS = 30
REF_INFO_CACHE = CachedSession("gma_ref_info_cache", expire_after=timedelta(days=REF_INFO_CACHE_TIME_DAYS))
def __init__(self):
super().__init__(self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def name(self):
return "GMA"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def http_response_to_spots(self, http_response):
new_spots = []
# Iterate through source data
for source_spot in http_response.json()["RCD"]:
# Convert to our spot format
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
dx_call=source_spot["ACTIVATOR"].upper(),
de_call=source_spot["SPOTTER"].upper(),
freq=float(source_spot["QRG"]) if (source_spot["QRG"] != "") else None, # Seen GMA spots with no frequency

9
providers/hema.py
View File

@@ -20,13 +20,10 @@ class HEMA(HTTPProvider):
FREQ_MODE_PATTERN = re.compile("^([\\d.]*) \\((.*)\\)$")
SPOTTER_COMMENT_PATTERN = re.compile("^\\((.*)\\) (.*)$")
def __init__(self):
super().__init__(self.SPOT_SEED_URL, self.POLL_INTERVAL_SEC)
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOT_SEED_URL, self.POLL_INTERVAL_SEC)
self.spot_seed = ""
def name(self):
return "HEMA"
def http_response_to_spots(self, http_response):
# OK, source data is actually just the spot seed at this point. We'll then go on to fetch real data if we know
# this has changed.
@@ -48,7 +45,7 @@ class HEMA(HTTPProvider):
spotter_comment_match = re.search(self.SPOTTER_COMMENT_PATTERN, spot_items[6])
# Convert to our spot format
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
dx_call=spot_items[2].upper(),
de_call=spotter_comment_match.group(1).upper(),
freq=float(freq_mode_match.group(1)) * 1000,

15
providers/http_provider.py
View File

@@ -13,20 +13,17 @@ from providers.provider import Provider
# duplication. Subclasses of this query the individual APIs for data.
class HTTPProvider(Provider):
def __init__(self, url, poll_interval):
super().__init__()
def __init__(self, provider_config, url, poll_interval):
super().__init__(provider_config)
self.url = url
self.poll_interval = poll_interval
self.poll_timer = None
def name(self):
raise NotImplementedError("Subclasses must implement this method")
def start(self):
# Fire off a one-shot thread to run poll() for the first time, just to ensure start() returns immediately and
# the application can continue starting. The thread itself will then die, and the timer will kick in on its own
# thread.
logging.info("Set up query of " + self.name() + " API every " + str(self.poll_interval) + " seconds.")
logging.info("Set up query of " + self.name + " API every " + str(self.poll_interval) + " seconds.")
thread = Thread(target=self.poll)
thread.daemon = True
thread.start()
@@ -37,7 +34,7 @@ class HTTPProvider(Provider):
def poll(self):
try:
# Request data from API
logging.debug("Polling " + self.name() + " API...")
logging.debug("Polling " + self.name + " API...")
http_response = requests.get(self.url, headers=self.HTTP_HEADERS)
# Pass off to the subclass for processing
new_spots = self.http_response_to_spots(http_response)
@@ -47,11 +44,11 @@ class HTTPProvider(Provider):
self.status = "OK"
self.last_update_time = datetime.now(pytz.UTC)
logging.debug("Received data from " + self.name() + " API.")
logging.debug("Received data from " + self.name + " API.")
except Exception as e:
self.status = "Error"
logging.exception("Exception in HTTP JSON Provider (" + self.name() + ")")
logging.exception("Exception in HTTP JSON Provider (" + self.name + ")")
sleep(1)
self.poll_timer = Timer(self.poll_interval, self.poll)

9
providers/parksnpeaks.py
View File

@@ -12,18 +12,15 @@ class ParksNPeaks(HTTPProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://www.parksnpeaks.org/api/ALL"
def __init__(self):
super().__init__(self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def name(self):
return "ParksNPeaks"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
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
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
source_id=source_spot["actID"],
dx_call=source_spot["actCallsign"].upper(),
de_call=source_spot["actSpoter"].upper(), # typo exists in API

9
providers/pota.py
View File

@@ -11,18 +11,15 @@ class POTA(HTTPProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://api.pota.app/spot/activator"
def __init__(self):
super().__init__(self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def name(self):
return "POTA"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
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
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
source_id=source_spot["spotId"],
dx_call=source_spot["activator"].upper(),
de_call=source_spot["spotter"].upper(),

10
providers/provider.py
View File

@@ -13,16 +13,14 @@ class Provider:
HTTP_HEADERS = { "User-Agent": SOFTWARE_NAME + " " + SOFTWARE_VERSION + " (operated by " + config["server-owner-callsign"] + ")" }
# Constructor
def __init__(self):
def __init__(self, provider_config):
self.name = provider_config["name"]
self.enabled = provider_config["enabled"]
self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
self.last_spot_time = datetime.min.replace(tzinfo=pytz.UTC)
self.status = "Not Started"
self.status = "Not Started" if self.enabled else "Disabled"
self.spot_list = None
# Return the name of the provider
def name(self):
raise NotImplementedError("Subclasses must implement this method")
# Set up the provider, e.g. giving it the spot list to work from
def setup(self, spot_list):
self.spot_list = spot_list

11
providers/rbn.py
View File

@@ -22,17 +22,14 @@ class RBN(Provider):
re.IGNORECASE)
# Constructor requires port number.
def __init__(self, port):
super().__init__()
self.port = port
def __init__(self, provider_config):
super().__init__(provider_config)
self.port = provider_config["port"]
self.telnet = None
self.thread = Thread(target=self.handle)
self.thread.daemon = True
self.run = True
def name(self):
return "RBN port " + str(self.port)
def start(self):
self.thread.start()
@@ -68,7 +65,7 @@ class RBN(Provider):
if match:
spot_time = datetime.strptime(match.group(5), "%H%MZ")
spot_datetime = datetime.combine(datetime.today(), spot_time.time()).replace(tzinfo=pytz.UTC)
spot = Spot(source="RBN",
spot = Spot(source=self.name,
dx_call=match.group(3),
de_call=match.group(1),
freq=float(match.group(2)),

9
providers/sota.py
View File

@@ -20,13 +20,10 @@ class SOTA(HTTPProvider):
SUMMIT_DATA_CACHE_TIME_DAYS = 30
SUMMIT_DATA_CACHE = CachedSession("sota_summit_data_cache", expire_after=timedelta(days=SUMMIT_DATA_CACHE_TIME_DAYS))
def __init__(self):
super().__init__(self.EPOCH_URL, self.POLL_INTERVAL_SEC)
def __init__(self, provider_config):
super().__init__(provider_config, self.EPOCH_URL, self.POLL_INTERVAL_SEC)
self.api_epoch = ""
def name(self):
return "SOTA"
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.
@@ -40,7 +37,7 @@ class SOTA(HTTPProvider):
# Iterate through source data
for source_spot in source_data:
# Convert to our spot format
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
source_id=source_spot["id"],
dx_call=source_spot["activatorCallsign"].upper(),
dx_name=source_spot["activatorName"],

9
providers/wwbota.py
View File

@@ -9,11 +9,8 @@ class WWBOTA(HTTPProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://api.wwbota.org/spots/"
def __init__(self):
super().__init__(self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def name(self):
return "WWBOTA"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def http_response_to_spots(self, http_response):
new_spots = []
@@ -26,7 +23,7 @@ class WWBOTA(HTTPProvider):
for ref in source_spot["references"]:
refs.append(ref["reference"])
ref_names.append(ref["name"])
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
dx_call=source_spot["call"].upper(),
de_call=source_spot["spotter"].upper(),
freq=float(source_spot["freq"]) * 1000, # MHz to kHz

9
providers/wwff.py
View File

@@ -11,18 +11,15 @@ class WWFF(HTTPProvider):
POLL_INTERVAL_SEC = 120
SPOTS_URL = "https://spots.wwff.co/static/spots.json"
def __init__(self):
super().__init__(self.SPOTS_URL, self.POLL_INTERVAL_SEC)
def name(self):
return "WWFF"
def __init__(self, provider_config):
super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC)
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
spot = Spot(source=self.name(),
spot = Spot(source=self.name,
source_id=source_spot["id"],
dx_call=source_spot["activator"].upper(),
de_call=source_spot["spotter"].upper(),