mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-25 08:44:33 +00:00
231 lines
11 KiB
Python
231 lines
11 KiB
Python
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_activity_by_name, get_icon_for_activity
|
|
from core.call_lookup_helper import get_call_info
|
|
from core.enums import ActivityName, Continent
|
|
from core.utils import get_flag_for_dxcc
|
|
from data.activity_ref import ActivityRef
|
|
|
|
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
|
|
|
|
# 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[ActivityName] = field(default_factory=list)
|
|
# Activity references. We allow multiple here for e.g. n-fer activations, unlike ADIF SIG_INFO.
|
|
activity_refs: list[ActivityRef] = 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 __post_init__(self):
|
|
"""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."""
|
|
|
|
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"""
|
|
|
|
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
|
|
|
|
# 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:
|
|
self.add_activity(activity_ref.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 first activity that has one, otherwise a radio tower
|
|
self.icon = "fa-tower-cell"
|
|
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: 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. Only canonical ActivityNames are
|
|
accepted otherwise we risk sending unknown stuff to API clients."""
|
|
|
|
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):
|
|
"""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())
|
|
)
|