import hashlib import json import logging from dataclasses import dataclass, field from datetime import datetime, timedelta 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.call_lookup_helper import get_call_info from core.enums import Continent from core.utils import get_flag_for_dxcc logger = logging.getLogger(__name__) @dataclass class Alert: """Data class that defines an alert.""" # Unique identifier for the alert id: str | None = None # DX (alerting) operator info # Callsigns of the operators that has been alerted dx_calls: list | None = None # Names of the operators that has been alerted dx_names: list | 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 # Maidenhead grid locator for the DX. This could be from a geographical reference e.g. POTA or grid. dx_grid: str | None = None # Latitude & longitude of the DX, in degrees. This could be from a geographical reference e.g. POTA or grid. dx_latitude: float | None = None dx_longitude: float | None = None # General alert info # Intended frequencies & modes of operation. Essentially just a different kind of comment field. freqs_modes: str | None = None # Start time of the activation, UTC seconds since UNIX epoch start_time: float | None = None # Start time of the activation of the alert, ISO 8601 start_time_iso: str | None = None # End time of the activation, UTC seconds since UNIX epoch. Optional end_time: float | None = None # End time of the activation of the alert, ISO 8601 end_time_iso: str | None = None # Comment made by the alerter, if any comment: str | None = None # A URL link to more information, if any url: str | None = None # Activity info # Activity (e.g. outdoor activity programme such as POTA). activity: str | None = None # 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 # Time that this software received the alert, 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 alert, ISO 8601 received_time_iso: str | None = None # Source info # Where we got the alert from, e.g. "POTA", "SOTA"... 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 alert in the web UI. Chosen from the Font Awesome set. icon: str | None = None def infer_missing(self, credentials=None): """Infer missing parameters where possible""" try: # If we somehow don't have a start 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.start_time: self.start_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.start_time and not self.start_time_iso: self.start_time_iso = datetime.fromtimestamp(self.start_time, pytz.UTC).isoformat() if self.end_time and not self.end_time_iso: self.end_time_iso = datetime.fromtimestamp(self.end_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() # DX country, continent, zones etc. from callsign. if self.dx_calls and self.dx_calls[0]: call_info = get_call_info(self.dx_calls[0], credentials) if self.dx_calls and self.dx_calls[0] and not self.dx_country: self.dx_country = call_info.country if self.dx_calls and self.dx_calls[0] and call_info.continent and not self.dx_continent: self.dx_continent = Continent(call_info.continent) if self.dx_calls and self.dx_calls[0] and not self.dx_cq_zone: self.dx_cq_zone = call_info.cq_zone if self.dx_calls and self.dx_calls[0] and not self.dx_itu_zone: self.dx_itu_zone = call_info.itu_zone if self.dx_calls and self.dx_calls[0] and not self.dx_dxcc_id: self.dx_dxcc_id = 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) # Fetch activity data, and set a real position if we can get one. if self.activity_refs: for activity_ref in self.activity_refs: activity_ref = populate_missing_activity_ref_info(activity_ref) # If the alert 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 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.activity_refs and self.activity_refs[0] and not self.activity: self.activity = self.activity_refs[0].activity # 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", exc_info=True) 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", exc_info=True) # 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 start time. Excluding things like the comment here allows for # user updates of their alert comments without duplicating in the system. 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_calls, "t": self.start_time}).encode("utf-8") ).hexdigest() # DX operator name lookup, using QRZ.com/HamQTH. if self.dx_calls and not self.dx_names: self.dx_names = [get_call_info(c, credentials).name for c in self.dx_calls] # Icon for the alert should be the icon of its activity if known, otherwise a radio tower self.icon = "fa-tower-cell" if self.activity and (activity_icon := get_icon_for_activity(self.activity)): 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 expired(self): """Decide if this alert 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 an end_time in the past, or if it only has a start_time, then that start time was more than 3 hours ago. If it somehow doesn't have a start_time either, it is considered to be expired.""" return ( not self.start_time or (self.end_time and self.end_time < datetime.now(pytz.UTC).timestamp()) or (not self.end_time and self.start_time < (datetime.now(pytz.UTC) - timedelta(hours=3)).timestamp()) )