diff --git a/core/activity_utils.py b/core/activity_utils.py index 4626005..6cc3ebd 100644 --- a/core/activity_utils.py +++ b/core/activity_utils.py @@ -1,7 +1,12 @@ +import logging + from data.activities import ACTIVITIES +from data.activity import Activity + +logger = logging.getLogger(__name__) -def get_activity_by_name(name): +def get_activity_by_name(name: str | None) -> Activity | None: """Utility function to resolve an arbitrary, case-insensitive activity name string (e.g. from a spot comment, a provider, or an API request) to the matching known Activity. Returns None if no match is found.""" @@ -10,10 +15,11 @@ def get_activity_by_name(name): for activity_name, activity in ACTIVITIES.items(): if activity_name.upper() == name.upper(): return activity + logger.warning(f"Unknown activity name '{name}', developer may need to add support for this!") return None -def get_ref_regex_for_activity(activity): +def get_ref_regex_for_activity(activity: str | None) -> str | None: """Utility function to get the regex string for an activity reference for a named activity. If no match is found, None will be returned.""" @@ -21,7 +27,7 @@ def get_ref_regex_for_activity(activity): return found.ref_regex if found else None -def get_icon_for_activity(activity): +def get_icon_for_activity(activity: str | None) -> str | None: """Utility function to get the icon for a named activity. If no match is found, None will be returned.""" found = get_activity_by_name(activity) diff --git a/data/activity_ref.py b/data/activity_ref.py index e0e7178..af48f4e 100644 --- a/data/activity_ref.py +++ b/data/activity_ref.py @@ -1,6 +1,6 @@ from dataclasses import dataclass -from core.enums import ActivityRefType +from core.enums import ActivityName, ActivityRefType @dataclass @@ -9,7 +9,7 @@ class ActivityRef: name and a lookup URL.""" # Activity that this reference is in, e.g. "POTA". - activity: str + activity: ActivityName # Reference ID, e.g. "GB-0001". id: str | None = None # Name of the reference, e.g. "Null Country Park", if known. diff --git a/data/alert.py b/data/alert.py index fc10843..255ef42 100644 --- a/data/alert.py +++ b/data/alert.py @@ -8,10 +8,11 @@ import pytz from pyhamtools.locator import latlong_to_locator, locator_to_latlong from core.activity_lookup_helper import populate_missing_activity_ref_info -from core.activity_utils import get_icon_for_activity +from core.activity_utils import get_activity_by_name, get_icon_for_activity from core.call_lookup_helper import get_call_info -from core.enums import Continent +from core.enums import ActivityName, Continent from core.utils import get_flag_for_dxcc +from data.activity_ref import ActivityRef logger = logging.getLogger(__name__) @@ -70,9 +71,9 @@ class Alert: # e.g. a POTA and WWFF dual activation. This is a list so we can maintain the order items were added, but needs to # be set-like to avoid dupes, and there's no Python class that handles that properly. So we use a list, but handle # the uniqueness logic manually, so you must use add_activity() to add to it instead of adding directly. - activities: list = field(default_factory=list) + activities: list[ActivityName] = field(default_factory=list) # Activity references. We allow multiple here for e.g. n-fer activations, unlike ADIF SIG_INFO. - activity_refs: list = field(default_factory=list) + activity_refs: list[ActivityRef] = field(default_factory=list) # Timing info @@ -96,9 +97,11 @@ class Alert: icon: str | None = None def __post_init__(self): - """Normalise the activities list, removing any duplicates while keeping the order.""" + """Normalise the activities list, converting any activity names provided as strings to their canonical + ActivityName (dropping any we don't know about) and removing any duplicates while keeping the order.""" - self.activities = list(dict.fromkeys(self.activities)) if self.activities else [] + found_activities = [get_activity_by_name(activity) for activity in self.activities or []] + self.activities = list(dict.fromkeys(found.name for found in found_activities if found)) def infer_missing(self, credentials=None): """Infer missing parameters where possible""" @@ -155,7 +158,7 @@ class Alert: # Add the activities of any activity refs we have to the alert's list of activities. for activity_ref in self.activity_refs: - if activity_ref and activity_ref.activity: + if activity_ref: self.add_activity(activity_ref.activity) # DX Grid to lat/lon and vice versa in case one is missing @@ -197,11 +200,16 @@ class Alert: except Exception: logger.exception("Exception while inferring missing data from spot") - def add_activity(self, activity): + def add_activity(self, activity: ActivityName | None): """Add an activity to the activities list, so long as it's not blank and not already there. The list is kept in - insertion order, so the first activity added is treated as the "primary" one.""" + insertion order, so the first activity added is treated as the "primary" one. Only canonical ActivityNames are + accepted otherwise we risk sending unknown stuff to API clients.""" - if activity and activity not in self.activities: + if not activity: + return + if not isinstance(activity, ActivityName): + raise TypeError(f"add_activity() requires an ActivityName, got {type(activity).__name__} {activity!r}") + if activity not in self.activities: self.activities.append(activity) def to_json(self): diff --git a/data/spot.py b/data/spot.py index 1c12b40..387e886 100644 --- a/data/spot.py +++ b/data/spot.py @@ -5,6 +5,7 @@ import re from dataclasses import dataclass, field from datetime import datetime, timedelta from math import isnan +from typing import cast import pytz from pyhamtools.locator import latlong_to_locator, locator_to_latlong @@ -129,9 +130,9 @@ class Spot: # e.g. a POTA and WWFF dual activation. This is a list so we can maintain the order items were added, but needs to # be set-like to avoid dupes, and there's no Python class that handles that properly. So we use a list, but handle # the uniqueness logic manually, so you must use add_activity() to add to it instead of adding directly. - activities: list = field(default_factory=list) + activities: list[ActivityName] = field(default_factory=list) # Activity references. We allow multiple here for e.g. n-fer activations, unlike ADIF SIG_INFO. - activity_refs: list = field(default_factory=list) + activity_refs: list[ActivityRef] = field(default_factory=list) # Timing info @@ -161,14 +162,22 @@ class Spot: def __post_init__(self): """Normalise fields that don't survive a plain dict to Spot conversion. This is used in the "add spot" API endpoint where the client is submitting JSON, and we want to recreate a full Spot object, including nested - objects such as the activity_refs list, and de-duplicating the activities list.""" + objects such as the activity_refs list, and de-duplicating the activities list. Activity names provided as + strings are converted to their canonical ActivityName. Any activities, or activity refs, for an activity we + don't know about are dropped.""" - self.activities = list(dict.fromkeys(self.activities)) if self.activities else [] - if self.activity_refs: - self.activity_refs = [ - activity_ref if isinstance(activity_ref, ActivityRef) else ActivityRef(**activity_ref) - for activity_ref in self.activity_refs - ] + found_activities = [get_activity_by_name(activity) for activity in self.activities or []] + self.activities = list(dict.fromkeys(found.name for found in found_activities if found)) + # When created from JSON, activity refs arrive as dicts rather than ActivityRef objects. + activity_refs = [] + for activity_ref in cast("list[ActivityRef | dict]", self.activity_refs or []): + if isinstance(activity_ref, ActivityRef): + activity_refs.append(activity_ref) + elif found_activity := get_activity_by_name(activity_ref.get("activity")): + activity_ref_data = dict(activity_ref) + activity_ref_data["activity"] = found_activity.name + activity_refs.append(ActivityRef(**activity_ref_data)) + self.activity_refs = activity_refs def infer_missing(self, credentials=None): """Infer missing parameters where possible""" @@ -274,13 +283,12 @@ class Spot: # Add the activities of any activity refs we have to the top-level activities list. for activity_ref in self.activity_refs: - if activity_ref.activity: - self.add_activity(activity_ref.activity.upper()) + self.add_activity(activity_ref.activity) # See if we already have an activity reference, but the comment looks like it contains more for the same # activity. This should catch e.g. POTA comments like "2-fer: GB-0001 GB-0002". - if self.comment and self.activity_refs and self.activity_refs[0].activity: - activity = self.activity_refs[0].activity.upper() + if self.comment and self.activity_refs: + activity = self.activity_refs[0].activity regex = get_ref_regex_for_activity(activity) if regex: all_comment_ref_matches = re.finditer(r"(?" not in source_spot["MODE"] else None, comment=source_spot["TEXT"], - activity_refs=[ - ActivityRef( - id=source_spot["REF"], - activity="", - name=source_spot["NAME"], - latitude=lat, - longitude=lon, - ) - ], time=time, dx_latitude=lat, dx_longitude=lon, @@ -98,57 +89,52 @@ class GMA(HTTPSpotProvider): and ref_response.text != "\n" ): ref_info = ref_response.json() - if spot.activity_refs and ref_info and "reftype" in ref_info: + if ref_info and "reftype" in ref_info: + activity: ActivityName | None = None + ref_type: ActivityRefType | None = None match ref_info["reftype"]: case "Summit": # Summits are a bit complicated, they can be SOTA or GMA depending on the # separate "sota" field: if "sota" in ref_info and ref_info["sota"] != "": - spot.activity_refs[0].activity = ActivityName.SOTA - spot.activity_refs[0].ref_type = ActivityRefType.SUMMIT - spot.add_activity(ActivityName.SOTA) + activity, ref_type = ActivityName.SOTA, ActivityRefType.SUMMIT else: - spot.activity_refs[0].activity = ActivityName.GMA - spot.activity_refs[0].ref_type = ActivityRefType.SUMMIT - spot.add_activity(ActivityName.GMA) + activity, ref_type = ActivityName.GMA, ActivityRefType.SUMMIT case "POTA": - spot.activity_refs[0].activity = ActivityName.POTA - spot.activity_refs[0].ref_type = ActivityRefType.PARK - spot.add_activity(ActivityName.POTA) + activity, ref_type = ActivityName.POTA, ActivityRefType.PARK case "WWFF": - spot.activity_refs[0].activity = ActivityName.WWFF - spot.activity_refs[0].ref_type = ActivityRefType.PARK - spot.add_activity(ActivityName.WWFF) + activity, ref_type = ActivityName.WWFF, ActivityRefType.PARK case "IOTA Island": - spot.activity_refs[0].activity = ActivityName.IOTA - spot.activity_refs[0].ref_type = ActivityRefType.ISLAND - spot.add_activity(ActivityName.IOTA) + activity, ref_type = ActivityName.IOTA, ActivityRefType.ISLAND case "GMA Island": - spot.activity_refs[0].activity = ActivityName.GMA_ISLANDS - spot.activity_refs[0].ref_type = ActivityRefType.ISLAND - spot.add_activity(ActivityName.GMA_ISLANDS) + activity, ref_type = ActivityName.GMA_ISLANDS, ActivityRefType.ISLAND case "Lighthouse (ILLW)": - spot.activity_refs[0].activity = ActivityName.ILLW - spot.activity_refs[0].ref_type = ActivityRefType.LIGHTHOUSE - spot.add_activity(ActivityName.ILLW) + activity, ref_type = ActivityName.ILLW, ActivityRefType.LIGHTHOUSE case "Lighthouse (ARLHS)": - spot.activity_refs[0].activity = ActivityName.ARLHS - spot.activity_refs[0].ref_type = ActivityRefType.LIGHTHOUSE - spot.add_activity(ActivityName.ARLHS) + activity, ref_type = ActivityName.ARLHS, ActivityRefType.LIGHTHOUSE case "Castle": - spot.activity_refs[0].activity = ActivityName.WCA - spot.activity_refs[0].ref_type = ActivityRefType.CASTLE - spot.add_activity(ActivityName.WCA) + activity, ref_type = ActivityName.WCA, ActivityRefType.CASTLE case "Mill": - spot.activity_refs[0].activity = ActivityName.MOTA - spot.activity_refs[0].ref_type = ActivityRefType.MILL - spot.add_activity(ActivityName.MOTA) + activity, ref_type = ActivityName.MOTA, ActivityRefType.MILL case _: logger.warning( f"GMA spot found with ref type {ref_info['reftype']}, developer needs to add support for this!" ) - spot.activity_refs[0].activity = ref_info["reftype"] - spot.add_activity(ref_info["reftype"]) + + # Now we know the activity, add the reference to the spot. If it's an activity we + # don't know, there's no ActivityName for it, so we can't add a reference. + if activity is not None: + spot.activity_refs = [ + ActivityRef( + id=source_spot["REF"], + activity=activity, + ref_type=ref_type, + name=source_spot["NAME"], + latitude=lat, + longitude=lon, + ) + ] + spot.add_activity(activity) elif not ref_response.from_cache: if not ref_response.ok: diff --git a/providers/spot/parksnpeaks.py b/providers/spot/parksnpeaks.py index 7c8d58f..4e90397 100644 --- a/providers/spot/parksnpeaks.py +++ b/providers/spot/parksnpeaks.py @@ -6,6 +6,7 @@ from typing import ClassVar import pytz import requests +from core.activity_utils import get_activity_by_name from core.constants import HTTP_HEADERS from core.enums import ActivityName, Mode from data.activity_ref import ActivityRef @@ -67,16 +68,19 @@ class ParksNPeaks(HTTPSpotProvider): # Record activity information activity = source_spot["actClass"].upper() + found_activity = get_activity_by_name(activity) ref_id = source_spot["actSiteID"] if activity: - spot.add_activity(activity) + if found_activity is not None: + spot.add_activity(found_activity.name) - if ref_id: + # We can only add a reference if we know the activity it's for + if ref_id and found_activity is not None: activity_refs = [ ActivityRef( id=ref_id, - activity=activity, + activity=found_activity.name, # Free text location is not present in all spots, so only add it if it's set name=source_spot["actLocation"] if "actLocation" in source_spot and source_spot["actLocation"] != "" @@ -96,23 +100,6 @@ class ParksNPeaks(HTTPSpotProvider): ): spot.comment = source_spot["actLocation"] - # Log a warning for the developer if PnP gives us an unknown programme we've never seen before - if activity not in [ - ActivityName.POTA, - ActivityName.SOTA, - ActivityName.WWFF, - ActivityName.HEMA, - ActivityName.SIOTA, - ActivityName.ZLOTA, - ActivityName.KRMNPA, - ActivityName.SANPCPA, - ActivityName.LLOTA, - ActivityName.QRP, - ]: - 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 diff --git a/providers/spot/xota.py b/providers/spot/xota.py index 57fb8b1..74ced34 100644 --- a/providers/spot/xota.py +++ b/providers/spot/xota.py @@ -1,13 +1,17 @@ import json +import logging from datetime import datetime import pytz -from core.enums import Mode +from core.activity_utils import get_activity_by_name +from core.enums import ActivityName, Mode from data.activity_ref import ActivityRef from data.spot import Spot from providers.spot.websocket_spot_provider import WebsocketSpotProvider +logger = logging.getLogger(__name__) + class XOTA(WebsocketSpotProvider): """Spot provider for servers based on the "xOTA" software at https://github.com/nischu/xOTA/ @@ -17,12 +21,17 @@ class XOTA(WebsocketSpotProvider): is why we also provide an activity_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.""" - ACTIVITY = None + ACTIVITY: ActivityName | None = None def __init__(self, provider_config): name = provider_config.get("name", "xOTA") super().__init__(name, provider_config, provider_config["url"]) - self.ACTIVITY = str(provider_config["activity"]) if "activity" in provider_config else None + found_activity = get_activity_by_name(provider_config.get("activity")) + self.ACTIVITY = found_activity.name if found_activity else None + if not self.ACTIVITY: + logger.error( + "XOTA provider has no activity reference, this is a config problem - your config needs to specify a known activity type!" + ) self._activity_ref_prefix = ( str(provider_config["activity_ref_prefix"]) if "activity_ref_prefix" in provider_config else "" ) @@ -31,20 +40,20 @@ class XOTA(WebsocketSpotProvider): string = b.decode("utf-8") source_spot = json.loads(string) ref_id = f"{self._activity_ref_prefix} {source_spot['reference']['title']}" + activity = self.ACTIVITY + activities = [] + activity_refs = [] + if activity is not None: + activities = [activity] + activity_refs = [ActivityRef(id=ref_id, activity=activity, url=source_spot["reference"]["website"])] 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()), - activities=[self.ACTIVITY] if self.ACTIVITY else [], - activity_refs=[ - ActivityRef( - id=ref_id, - activity=self.ACTIVITY or "", - url=source_spot["reference"]["website"], - ) - ], + activities=activities, + activity_refs=activity_refs, time=datetime.now(pytz.UTC).timestamp(), qrt=source_spot["state"] != "active", ) diff --git a/templates/add-spot.html b/templates/add-spot.html index a7f90a9..7d8534f 100644 --- a/templates/add-spot.html +++ b/templates/add-spot.html @@ -77,7 +77,7 @@ - + diff --git a/templates/alerts.html b/templates/alerts.html index b4fbf11..2a1d88e 100644 --- a/templates/alerts.html +++ b/templates/alerts.html @@ -85,7 +85,7 @@ - + diff --git a/templates/bands.html b/templates/bands.html index 617e2bf..1d106fa 100644 --- a/templates/bands.html +++ b/templates/bands.html @@ -76,8 +76,8 @@ - - + + diff --git a/templates/base.html b/templates/base.html index 6cbe0e2..318967f 100644 --- a/templates/base.html +++ b/templates/base.html @@ -1,6 +1,6 @@ {% extends "skeleton.html" %} {% block head_extra %} - + @@ -16,10 +16,10 @@ window.fetchEventSource = fetchEventSource; - - - - + + + + {% end %} {% block body %}