import hashlib import json import logging import re from dataclasses import dataclass, field from datetime import datetime, timedelta from math import isnan 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 ( ANY_ACTIVITY_REGEX, get_activity_name_from_comment_name, get_icon_for_activity, get_ref_regex_for_activity, ) from core.call_lookup_helper import get_call_info from core.config import MAX_SPOT_AGE from core.constants import ACTIVITIES, PROPAGATION_MODES from core.data_store import DATA_STORE from core.enums import Continent, LocationSourceForSpot, Mode, ModeSource, ModeType from core.geo_utils import lat_lon_to_cq_zone, lat_lon_to_itu_zone from core.utils import ( get_flag_for_dxcc, infer_band_from_freq, infer_mode_from_comment, infer_mode_from_frequency, infer_mode_type_from_mode, ) from data.activity_ref import ActivityRef logger = logging.getLogger(__name__) @dataclass class Spot: """Data class that defines a spot.""" # Unique identifier for the spot id: str | None = None # DX (spotted) operator info # Callsign of the operator that has been spotted dx_call: str | None = None # Name of the operator that has been spotted dx_name: str | None = None # QTH of the operator that has been spotted. This could be from any activity refs or could be from online lookup of # their home QTH. dx_qth: str | None = None # Country of the DX operator dx_country: str | None = None # Country flag of the DX operator dx_flag: str | None = None # Continent of the DX operator dx_continent: Continent | None = None # DXCC ID of the DX operator dx_dxcc_id: int | None = None # CQ zone of the DX operator dx_cq_zone: int | None = None # ITU zone of the DX operator dx_itu_zone: int | None = None # If this is an APRS/Packet/etc spot, what SSID was the DX operator using? dx_ssid: str | None = None # Maidenhead grid locator for the DX. This could be from a geographical reference e.g. POTA, or just from the # country dx_grid: str | None = None # Latitude & longitude of the DX, in degrees. This could be from a geographical reference e.g. POTA, or from a QRZ # lookup dx_latitude: float | None = None dx_longitude: float | None = None # 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.) dx_location_good: bool = False # DE (Spotter) info # Callsign of the spotter de_call: str | None = None # Country of the spotter de_country: str | None = None # Country flag of the spotter de_flag: str | None = None # Continent of the spotter de_continent: Continent | None = None # DXCC ID of the spotter de_dxcc_id: int | None = None # If this is an APRS/Packet/etc spot, what SSID was the spotter/receiver using? de_ssid: str | None = None # Maidenhead grid locator for the spotter. This is not going to be from a xOTA reference so it will likely just be # a QRZ or DXCC lookup. If the spotter is also portable, this is probably wrong, but it's good enough for some # simple mapping. de_grid: str | None = None # Latitude & longitude of the DX, in degrees. This is not going to be from a xOTA reference so it will likely just # be a QRZ or DXCC lookup. If the spotter is also portable, this is probably wrong, but it's good enough for some # simple mapping. de_latitude: float | None = None de_longitude: float | None = None # General QSO info # Reported mode, such as SSB, PHONE, CW, FT8... mode: Mode | None = None # Inferred mode "family". mode_type: ModeType | None = None # Source of the mode information. mode_source: ModeSource | None = None # Frequency, in Hz freq: float | None = None # Band, defined by the frequency, e.g. "40m" or "70cm" band: str | None = None # Propagation mode, if known propagation_mode: str | None = None # Comment left by the spotter, if any comment: str | None = None # QRT state. Some APIs return spots marked as QRT. Otherwise we can check the comments. qrt: bool = False # 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) # Timing info # Time of the spot, UTC seconds since UNIX epoch time: float | None = None # Time of the spot, ISO 8601 time_iso: str | None = None # Time that this software received the spot, UTC seconds since UNIX epoch. This is used with the "since_received" # call to our API to receive all data that is new to us, even if by a quirk of the API it might be older than the # list time the client polled the API. received_time: float | None = None # Time that this software received the spot, ISO 8601 received_time_iso: str | None = None # Source info # Where we got the spot from, e.g. "POTA", "Cluster"... source: str | None = None # The ID the source gave it, if any. source_id: str | None = None # Display info # Icon to use when displaying this spot in the web UI. Chosen from the Font Awesome set. icon: str | None = None 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..""" if self.sig_refs: self.sig_refs = [ activity_ref if isinstance(activity_ref, ActivityRef) else ActivityRef(**activity_ref) for activity_ref in self.sig_refs ] def infer_missing(self, credentials=None): """Infer missing parameters where possible""" try: # If we somehow don't have a spot time, set it to zero so it sorts off the bottom of any list but # clients can still reliably parse it as a number. if not self.time: self.time = 0 # If we don't have a received time, this has just been received so set that to "now" if not self.received_time: self.received_time = datetime.now(pytz.UTC).timestamp() # Fill in ISO versions of times, in case the client prefers that if self.time and not self.time_iso: self.time_iso = datetime.fromtimestamp(self.time, pytz.UTC).isoformat() if self.received_time and not self.received_time_iso: self.received_time_iso = datetime.fromtimestamp(self.received_time, pytz.UTC).isoformat() # Clean up DX call if it has an SSID or -# from RBN if self.dx_call and "-" in self.dx_call: split = self.dx_call.split("-") self.dx_call = split[0] if len(split) > 1 and split[1] != "#": self.dx_ssid = split[1] # DX country, continent etc. from callsign dx_call_info = get_call_info(self.dx_call, credentials) if self.dx_call and not self.dx_country: self.dx_country = dx_call_info.country if self.dx_call and dx_call_info.continent and not self.dx_continent: self.dx_continent = Continent(dx_call_info.continent) if self.dx_call and not self.dx_dxcc_id: self.dx_dxcc_id = dx_call_info.dxcc_id if self.dx_dxcc_id and not self.dx_flag: self.dx_flag = get_flag_for_dxcc(self.dx_dxcc_id) # Clean up spotter call if it has an SSID or -# from RBN if self.de_call and "-" in self.de_call: split = self.de_call.split("-") self.de_call = split[0] if len(split) > 1 and split[1] != "#": self.de_ssid = split[1] # If we have a spotter of "RBNHOLE", we should have the actual spotter callsign in the comment, so extract it. # RBNHole posts come from a number of providers, so it's dealt with here in the generic spot handling code. if self.de_call == "RBNHOLE" and self.comment: rbnhole_call_match = re.search(r"\Wat ([a-z0-9/]+)\W", self.comment, re.IGNORECASE) if rbnhole_call_match: self.de_call = rbnhole_call_match.group(1).upper() # If we have a spotter of "SOTAMAT", we might have the actual spotter callsign in the comment, if so extract it. # SOTAMAT can do POTA as well as SOTA, so it's dealt with here in the generic spot handling code. if self.de_call == "SOTAMAT" and self.comment: sotamat_call_match = re.search(r"\Wfrom ([a-z0-9/]+)]", self.comment, re.IGNORECASE) if sotamat_call_match: self.de_call = sotamat_call_match.group(1).upper() # Spotter country, continent, zones etc. from callsign. # DE call with no digits, or APRS servers starting "T2" are not things we can look up location for de_call_info = get_call_info(self.de_call, credentials) if ( self.de_call and any(char.isdigit() for char in self.de_call) and not (self.de_call.startswith("T2") and self.source == "APRS-IS") ): if not self.de_country: self.de_country = de_call_info.country if de_call_info.continent and not self.de_continent: self.de_continent = Continent(de_call_info.continent) if not self.de_dxcc_id: self.de_dxcc_id = de_call_info.dxcc_id if self.de_dxcc_id and not self.de_flag: self.de_flag = get_flag_for_dxcc(self.de_dxcc_id) # Remove NaNs in frequency if self.freq and isnan(self.freq): self.freq = None # Band from frequency if self.freq and not self.band: band = infer_band_from_freq(self.freq) self.band = band.name # Mode from comments or bandplan if self.mode: self.mode_source = ModeSource.SPOT if self.comment and not self.mode: self.mode = infer_mode_from_comment(self.comment) self.mode_source = ModeSource.COMMENT if self.freq and not self.mode: self.mode = infer_mode_from_frequency(self.freq) self.mode_source = ModeSource.BANDPLAN # Mode type from mode if self.mode and not self.mode_type: self.mode_type = infer_mode_type_from_mode(self.mode) # If we have a latitude or grid at this point, it can only have been provided by the spot itself 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() # 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() 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)) # 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. # POTA comments like "also WWFF GFF-0001". 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 # 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 # 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. ref_regex = get_ref_regex_for_activity(found_activity) if ref_regex: ref_matches = re.finditer( r"(^|\W)" + found_activity + r"([ -])(" + ref_regex + r")($|\W)", self.comment, re.IGNORECASE, ) for ref_match in ref_matches: self._append_activity_ref_if_missing( ActivityRef(id=ref_match.group(3).upper(), sig=found_activity) ) # See if the comment looks like it contains any activity references *without* the corresponding activity # name, but where the activity reference is unique-looking enough that we can't confuse it with any other # activity. if self.comment: for activity in ACTIVITIES: if activity.refs_globally_unique and activity.ref_regex: ref_matches = re.finditer( 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 self._append_activity_ref_if_missing( ActivityRef(id=ref_match.group(2).upper(), sig=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: 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: self.dx_grid = activity_ref.grid if ( activity_ref.latitude and not self.dx_latitude and activity_ref.longitude and not self.dx_longitude ): self.dx_latitude = activity_ref.latitude self.dx_longitude = activity_ref.longitude if self.sig == "WAB" or self.sig == "WAI" or self.sig == "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 # Parse "de_griddx_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 # being the only source we have for propagation mode. Brace for nightmare regex from hell. if self.comment: grid_mode_grid_match = re.search( r"\b([A-Ra-r]{2}\d{2}(?:[A-Xa-x]{2}(?:\d{2})?)?)(?:<([^>]*)>|\(([^)]*)\))([A-Ra-r]{2}\d{2}(?:[A-Xa-x]{2}(?:\d{2})?)?)\b", self.comment, ) if grid_mode_grid_match: # regex matches, so extract grids: if not self.de_grid: self.de_grid = grid_mode_grid_match.group(1).upper() if not self.dx_grid: self.dx_grid = grid_mode_grid_match.group(4).upper() self.dx_location_source = LocationSourceForSpot.GRID # And extract propagation mode (group 2 for <...>, group 3 for (...)): mode_tag = (grid_mode_grid_match.group(2) or grid_mode_grid_match.group(3) or "").upper() if mode_tag and not self.propagation_mode: if mode_tag in PROPAGATION_MODES: self.propagation_mode = PROPAGATION_MODES[mode_tag] else: self.propagation_mode = mode_tag logger.info(f"Seen a new propagation mode tag not yet in the system: {mode_tag}") # Set activities based on propagation mode if self.propagation_mode == "Satellite" and not self.sig: self.sig = "Satellite" if self.propagation_mode == "Earth-Moon-Earth" and not self.sig: self.sig = "EME" # Parse "de_grid -> dx_grid" structures from the comment if self.comment: grid_mode_grid_match = re.search( r"\b([A-Ra-r]{2}\d{2}(?:[A-Xa-x]{2}(?:\d{2})?)?)\s*->\s*([A-Ra-r]{2}\d{2}(?:[A-Xa-x]{2}(?:\d{2})?)?)\b", self.comment, ) if grid_mode_grid_match: # regex matches, so extract grids: if not self.dx_grid: self.dx_grid = grid_mode_grid_match.group(1).upper() self.dx_location_source = LocationSourceForSpot.GRID if not self.de_grid: self.de_grid = grid_mode_grid_match.group(2).upper() # DX Grid to lat/lon and vice versa in case one is missing if self.dx_grid and (not self.dx_latitude or not self.dx_longitude): try: ll = locator_to_latlong(self.dx_grid) self.dx_latitude = ll[0] self.dx_longitude = ll[1] except Exception: logger.debug("Invalid grid received for spot") if self.dx_latitude and self.dx_longitude and not self.dx_grid: try: self.dx_grid = latlong_to_locator(self.dx_latitude, self.dx_longitude, 8) except Exception: logger.debug("Invalid lat/lon received for spot") # QRT comment detection if self.comment and not self.qrt: self.qrt = "QRT" in self.comment.upper() # Create an ID based on the source and source ID if possible, as these guaranee uniqueness. If there is no # source ID, use a combination of callsign and spot time. Spot time is down to the second or even the # millisecond, so we can be reasonably sure two spots that match are the same spot. if not self.id: if self.source and self.source_id: self.id = hashlib.sha256(str({"s": self.source, "sid": self.source_id}).encode("utf-8")).hexdigest() else: self.id = hashlib.sha256( str({"s": self.source, "c": self.dx_call, "t": self.time}).encode("utf-8") ).hexdigest() # DX operator details lookup. This should be the last resort compared to taking the data from the actual # spotting service, e.g. we don't want to accidentally use a user's QRZ.com home lat/lon or DXCC lat/lon # instead of the one from the park reference they're at. if self.dx_call and not self.dx_name: self.dx_name = dx_call_info.name if self.dx_call and (not self.dx_latitude or not self.dx_longitude): self.dx_latitude = dx_call_info.latitude self.dx_longitude = dx_call_info.longitude self.dx_grid = dx_call_info.grid self.dx_location_source = dx_call_info.location_source # 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}" self.dx_qth = qth else: self.dx_qth = dx_call_info.qth # CQ and ITU zone lookup, preferably from location but failing that, from callsign if not self.dx_cq_zone: if self.dx_latitude and self.dx_longitude: self.dx_cq_zone = lat_lon_to_cq_zone(self.dx_latitude, self.dx_longitude) elif self.dx_call: self.dx_cq_zone = dx_call_info.cq_zone if not self.dx_itu_zone: if self.dx_latitude and self.dx_longitude: self.dx_itu_zone = lat_lon_to_itu_zone(self.dx_latitude, self.dx_longitude) elif self.dx_call: self.dx_itu_zone = dx_call_info.itu_zone # DXCC lookup from callsign if nothing else has provided it if self.dx_call and not self.dx_dxcc_id: for regex, entity_code in DATA_STORE.dxcc_lookup_by_call_regex: if regex.pattern and regex.match(self.dx_call): self.dx_dxcc_id = entity_code break if self.dx_dxcc_id and not self.dx_flag: self.dx_flag = get_flag_for_dxcc(self.dx_dxcc_id) # DX Location is "good" if it is from a spot, or from QRZ if the callsign doesn't contain a slash, so the operator # is likely at home. self.dx_location_good = bool( self.dx_latitude 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.GRID or (self.dx_location_source == LocationSourceForSpot.HOME_QTH and "/" not in (self.dx_call or "")) ) ) # DE with no digits and APRS servers starting "T2" are not things we can look up location for if ( self.de_call and any(char.isdigit() for char in str(self.de_call)) and not (self.de_call.startswith("T2") and self.source == "APRS-IS") and (not self.de_latitude or not self.de_longitude) ): # DE operator location lookup self.de_latitude = de_call_info.latitude 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 self.icon = "fa-tower-cell" if self.sig and (activity_icon := get_icon_for_activity(self.sig)): self.icon = activity_icon except Exception: logger.exception("Exception while inferring missing data from spot") def to_json(self): """JSON serialise""" return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True) def _append_activity_ref_if_missing(self, new_activity_ref): """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() 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: return self.sig_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 returned by the web server if later requested, and removed by the cleanup functions). "Expired" is defined as either having a time further ago than the server's MAX_SPOT_AGE. If it somehow doesn't have a time either, it is considered to be expired.""" return not self.time or self.time < (datetime.now(pytz.UTC) - timedelta(seconds=MAX_SPOT_AGE)).timestamp()