mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 14:27:42 +00:00
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
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()
|