sig->activity and multiple activity changes for API v3. #143

This commit is contained in:
ian
2026-09-25 07:01:13 +01:00
committed by Ian Renton
parent 02d08c17cd
commit ecdcbe17e9
92 changed files with 1113 additions and 796 deletions
+67 -59
View File
@@ -76,8 +76,8 @@ class Spot:
# DX Location source. Indicates how accurate the location might be.
dx_location_source: LocationSourceForSpot | None = None
# DX Location good. Indicates that the software thinks the location data is good enough to plot on a map. This is
# true if the location source is "SPOT", "SIG REF LOOKUP" or "GRID", or if the location source is "HOME QTH" and
# the DX callsign doesn't have a suffix like /P. (Location source retains "SIG" wording for API compatibility.)
# true if the location source is "SPOT", "ACTIVITY REF LOOKUP" or "GRID", or if the location source is "HOME QTH"
# and the DX callsign doesn't have a suffix like /P.
dx_location_good: bool = False
# DE (Spotter) info
@@ -125,11 +125,13 @@ class Spot:
# Activity info
# Activity (e.g. outdoor activity programme such as POTA). Still named "sig" for API backwards compatibility.
sig: str | None = None
# Activity references. We allow multiple here for e.g. n-fer activations, unlike ADIF SIG_INFO. Still named
# "sig_refs" for API backwards compatibility.
sig_refs: list = field(default_factory=list)
# Activities (e.g. outdoor activity programmes such as POTA). An alert can be for several activities at once,
# 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)
# Activity references. We allow multiple here for e.g. n-fer activations, unlike ADIF SIG_INFO.
activity_refs: list = field(default_factory=list)
# Timing info
@@ -159,12 +161,13 @@ 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 sig_refs list.."""
objects such as the activity_refs list, and de-duplicating the activities list."""
if self.sig_refs:
self.sig_refs = [
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.sig_refs
for activity_ref in self.activity_refs
]
def infer_missing(self, credentials=None):
@@ -269,19 +272,22 @@ class Spot:
if self.dx_latitude or self.dx_grid:
self.dx_location_source = LocationSourceForSpot.SPOT
# Set the top-level activity if it is missing but we have at least one activity ref.
if not self.sig and self.sig_refs:
self.sig = self.sig_refs[0].sig.upper()
# 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())
# 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.sig_refs and self.sig_refs[0].sig:
activity = self.sig_refs[0].sig.upper()
if self.comment and self.activity_refs and self.activity_refs[0].activity:
activity = self.activity_refs[0].activity.upper()
regex = get_ref_regex_for_activity(activity)
if regex:
all_comment_ref_matches = re.finditer(r"(^|\W)(" + regex + r")($|\W)", self.comment, re.IGNORECASE)
for ref_match in all_comment_ref_matches:
self._append_activity_ref_if_missing(ActivityRef(id=ref_match.group(2).upper(), sig=activity))
self._append_activity_ref_if_missing(
ActivityRef(id=ref_match.group(2).upper(), activity=activity)
)
# See if the comment looks like it contains any activities (and optionally activity references) that we
# can add to the spot. This should catch cluster spot comments like "POTA GB-0001 WWFF GFF-0001" and e.g.
@@ -289,14 +295,13 @@ class Spot:
if self.comment:
activity_matches = re.finditer(r"(^|\W)" + ANY_ACTIVITY_REGEX + r"($|\W)", self.comment, re.IGNORECASE)
for activity_match in activity_matches:
# First of all, if we haven't got an activity for this spot set yet, now we have. This covers
# First of all, add the activity to this spot's list of activities. This covers
# things like cluster spots where the comment is just "POTA".
found_activity = get_activity_name_from_comment_name(activity_match.group(2))
if not self.sig:
self.sig = found_activity
self.add_activity(found_activity)
# Now look to see if that activity name was followed by something that looks like a reference ID
# for that activity. If so, add that to the sig_refs list for this spot.
# for that activity. If so, add that to the activity_refs list for this spot.
found_activity_info = get_activity_by_name(found_activity)
if found_activity_info and found_activity_info.has_refs and found_activity_info.ref_regex:
ref_matches = re.finditer(
@@ -306,7 +311,7 @@ class Spot:
)
for ref_match in ref_matches:
self._append_activity_ref_if_missing(
ActivityRef(id=ref_match.group(3).upper(), sig=found_activity)
ActivityRef(id=ref_match.group(3).upper(), activity=found_activity)
)
# See if the comment looks like it contains any activity references *without* the corresponding activity
@@ -319,20 +324,18 @@ class Spot:
r"(^|\W)(" + activity.ref_regex + r")($|\W)", self.comment, re.IGNORECASE
)
for ref_match in ref_matches:
# First of all, if we haven't got an activity for this spot set yet, now we have. This
# covers things like cluster spots where the comment is just "OHFF-1234", now we know
# it's WWFF.
if not self.sig:
self.sig = activity.name
# First of all, add the activity to this spot's list of activities. This covers things
# like cluster spots where the comment is just "OHFF-1234", now we know it's WWFF.
self.add_activity(activity.name)
self._append_activity_ref_if_missing(
ActivityRef(id=ref_match.group(2).upper(), sig=activity.name)
ActivityRef(id=ref_match.group(2).upper(), activity=activity.name)
)
# Fetch activity data. In case a particular API doesn't provide a full set of name, lat, lon & grid for a
# reference in its initial call, we use this code to populate the rest of the data. This includes working
# out grid refs from WAB and WAI, which count as an activity even though there's no real lookup, just maths
if self.sig_refs:
for activity_ref in self.sig_refs:
if self.activity_refs:
for activity_ref in self.activity_refs:
activity_ref = populate_missing_activity_ref_info(activity_ref)
# If the spot itself doesn't have location yet, but the activity ref does, extract it
if activity_ref.grid and not self.dx_grid:
@@ -345,15 +348,10 @@ class Spot:
):
self.dx_latitude = activity_ref.latitude
self.dx_longitude = activity_ref.longitude
if self.sig in (ActivityName.WAB, ActivityName.WAI, ActivityName.TILES):
if activity_ref.activity in (ActivityName.WAB, ActivityName.WAI, ActivityName.TILES):
self.dx_location_source = LocationSourceForSpot.GRID
else:
self.dx_location_source = LocationSourceForSpot.SIG_REF_LOOKUP
# If the spot itself doesn't have an activity yet, but we have at least one activity reference, take that
# reference's activity and apply it to the whole spot.
if self.sig_refs and not self.sig:
self.sig = self.sig_refs[0].sig
self.dx_location_source = LocationSourceForSpot.ACTIVITY_REF_LOOKUP
# Parse "de_grid<prop_mode>dx_grid" structures from the comment, e.g. "JN61ES(ES)JM56XT" or "JO02GQ<>KN17LG".
# These are common on cluster spots and can provide grid references in preference to e.g. QRZ lookup, as well as
@@ -406,32 +404,33 @@ class Spot:
self.dx_location_source = LocationSourceForSpot.GRID
# Set activities based on propagation mode
if self.propagation_mode == "Satellite" and not self.sig:
self.sig = ActivityName.SATELLITE
if self.propagation_mode == "Earth-Moon-Earth" and not self.sig:
self.sig = ActivityName.EME
if self.propagation_mode == "Satellite":
self.add_activity(ActivityName.SATELLITE)
if self.propagation_mode == "Earth-Moon-Earth":
self.add_activity(ActivityName.EME)
# Set activities based on the DX callsign suffix
if self.dx_call and not self.sig:
if self.dx_call:
if self.dx_call.upper().endswith(ActivityName.AERONAUTICAL_MOBILE):
self.sig = ActivityName.AERONAUTICAL_MOBILE
self.add_activity(ActivityName.AERONAUTICAL_MOBILE)
elif self.dx_call.upper().endswith(ActivityName.MARITIME_MOBILE):
self.sig = ActivityName.MARITIME_MOBILE
self.add_activity(ActivityName.MARITIME_MOBILE)
# Alright, now let's get really fancy. Check if the DX callsign matches one taking part in a currently
# running DXpedition which we know from the alerts list.
if self.dx_call and not self.sig:
if self.dx_call and not self.activities:
now = datetime.now(pytz.UTC).timestamp()
for alert in DATA_STORE.alerts.values():
if (
alert.sig == ActivityName.DXPEDITION
alert.activities
and ActivityName.DXPEDITION in alert.activities
and alert.dx_calls
and alert.start_time
and alert.end_time
and alert.start_time < now < alert.end_time
and self.dx_call.upper() in [c.upper() for c in alert.dx_calls if c]
):
self.sig = ActivityName.DXPEDITION
self.add_activity(ActivityName.DXPEDITION)
break
# DX Grid to lat/lon and vice versa in case one is missing
@@ -476,10 +475,10 @@ class Spot:
# Determine a "QTH" string. If we have an activity ref, pick the first one and turn it into a suitable
# string, otherwise see what they have set on an online lookup service.
if self.sig_refs:
qth = self.sig_refs[0].id
if self.sig_refs[0].name:
qth += f" {self.sig_refs[0].name}"
if self.activity_refs:
qth = self.activity_refs[0].id
if self.activity_refs[0].name:
qth += f" {self.activity_refs[0].name}"
self.dx_qth = qth
else:
self.dx_qth = dx_call_info.qth
@@ -512,7 +511,7 @@ class Spot:
and self.dx_longitude
and (
self.dx_location_source == LocationSourceForSpot.SPOT
or self.dx_location_source == LocationSourceForSpot.SIG_REF_LOOKUP
or self.dx_location_source == LocationSourceForSpot.ACTIVITY_REF_LOOKUP
or self.dx_location_source == LocationSourceForSpot.GRID
or (self.dx_location_source == LocationSourceForSpot.HOME_QTH and "/" not in (self.dx_call or ""))
)
@@ -530,14 +529,23 @@ class Spot:
self.de_longitude = de_call_info.longitude
self.de_grid = de_call_info.grid
# Icon for the spot should be the icon of its activity if known, otherwise a radio tower
# Icon for the spot should be the icon of its first activity that has one, otherwise a radio tower
self.icon = "fa-tower-cell"
if self.sig and (activity_icon := get_icon_for_activity(self.sig)):
self.icon = activity_icon
for activity in self.activities:
if activity_icon := get_icon_for_activity(activity):
self.icon = activity_icon
break
except Exception:
logger.exception("Exception while inferring missing data from spot")
def add_activity(self, activity):
"""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."""
if activity and activity not in self.activities:
self.activities.append(activity)
def to_json(self):
"""JSON serialise"""
@@ -547,13 +555,13 @@ class Spot:
"""Append an activity ref to the list, so long as it's not already there."""
new_activity_ref.id = new_activity_ref.id.strip().upper()
new_activity_ref.sig = new_activity_ref.sig.strip().upper()
new_activity_ref.activity = new_activity_ref.activity.strip().upper()
if new_activity_ref.id == "":
return
for activity_ref in self.sig_refs:
if activity_ref.id == new_activity_ref.id and activity_ref.sig == new_activity_ref.sig:
for activity_ref in self.activity_refs:
if activity_ref.id == new_activity_ref.id and activity_ref.activity == new_activity_ref.activity:
return
self.sig_refs.append(new_activity_ref)
self.activity_refs.append(new_activity_ref)
def expired(self):
"""Decide if this spot has expired (in which case it should not be added to the system in the first place, and not