Support RSGB contests

providers/alert/rsgb_ical_alert_provider.py

@@ -0,0 +1,101 @@
+import re
+from datetime import datetime, time
+from typing import cast
+
+import pytz
+from icalendar import Calendar, Event
+
+from core.enums import AlertType, Continent
+from data.alert import Alert
+from providers.alert.http_alert_provider import HTTPAlertProvider
+
+
+class RSGBICALAlertProvider(HTTPAlertProvider):
+    """Generic alert provider for RSGB contest ICAL calendars."""
+
+    def __init__(self, name, provider_config, url, poll_interval):
+        super().__init__(name, provider_config, url, poll_interval)
+
+    FREQ_PATTERN = re.compile(r"([\d.]+(?:MHz|GHz))|SHF")
+
+    def _http_response_to_alerts(self, http_response):
+        new_alerts = []
+        cal = Calendar.from_ical(http_response.content)
+
+        # Iterate through events
+        for component in cal.walk():
+            if component.name != "VEVENT":
+                continue
+
+            event = cast(Event, component)
+            summary = str(event.get("summary", "")).strip()
+
+            # Ensure summaries start with "RSGB" to avoid any confusion
+            if not summary.startswith("RSGB "):
+                summary = "RSGB " + summary
+
+            # Extract freq from summary
+            freqs_modes = ""
+
+            # HF contests don't give frequencies, all VHF ones do
+            match = self.FREQ_PATTERN.search(summary)
+            if match:
+                freqs_modes = match.group()
+            else:
+                freqs_modes = "HF bands"
+            freqs_modes = freqs_modes + ", "
+
+            # Extract mode from summary
+            if "FMAC" in summary:
+                freqs_modes = freqs_modes + "FM"
+            elif "CW" in summary:
+                freqs_modes = freqs_modes + "CW"
+            elif "SSB" in summary:
+                freqs_modes = freqs_modes + "SSB"
+            elif "FT8" in summary:
+                freqs_modes = freqs_modes + "FT8"
+            elif "FT4" in summary:
+                freqs_modes = freqs_modes + "FT4"
+            elif "DATA" in summary:
+                freqs_modes = freqs_modes + "Data modes"
+            else:
+                freqs_modes = freqs_modes + "All modes"
+
+            dtstart = event.get("dtstart")
+            dtend = event.get("dtend")
+            start_timestamp = self._to_utc_timestamp(dtstart.dt)
+            end_timestamp = self._to_utc_timestamp(dtend.dt) - 1 if dtend is not None else start_timestamp
+
+            # Convert to our alert format
+            alert = Alert(
+                source=self.name,
+                dx_calls=[],
+                dx_country="United Kingdom",
+                dx_dxcc_id=235,
+                dx_continent=Continent.EU,
+                dx_cq_zone=14,
+                dx_itu_zone=27,
+                dx_flag="🇬🇧",
+                freqs_modes=freqs_modes,
+                comment=summary,
+                start_time=start_timestamp,
+                end_time=end_timestamp,
+                alert_type=AlertType.CONTEST,
+            )
+
+            # Add to our list.
+            new_alerts.append(alert)
+        return new_alerts
+
+    @staticmethod
+    def _to_utc_timestamp(value):
+        """Convert a date or datetime value from an iCal field into a UTC UNIX timestamp."""
+
+        # Datetime object so we can treat it as-is, check if it has a non-UTC tz and convert it if necessary
+        if isinstance(value, datetime):
+            if value.tzinfo is None:
+                value = pytz.UTC.localize(value)
+            return value.astimezone(pytz.UTC).timestamp()
+
+        # Date object so this is an all day event
+        return pytz.UTC.localize(datetime.combine(value, time.min)).timestamp()