mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 14:27:42 +00:00
Add support for WA7BNM Contest Calendar. Closes #106
This commit is contained in:
Ian Renton
@@ -1,101 +1,75 @@
|
||||
import re
|
||||
from datetime import datetime, time
|
||||
from typing import cast
|
||||
|
||||
import pytz
|
||||
from icalendar import Calendar, Event
|
||||
from icalendar import Event
|
||||
|
||||
from core.enums import AlertType, Continent
|
||||
from data.alert import Alert
|
||||
from providers.alert.http_alert_provider import HTTPAlertProvider
|
||||
from providers.alert.ical_alert_provider import ICALAlertProvider
|
||||
|
||||
|
||||
class RSGBICALAlertProvider(HTTPAlertProvider):
|
||||
"""Generic alert provider for RSGB contest ICAL calendars."""
|
||||
class RSGBICALAlertProvider(ICALAlertProvider):
|
||||
"""Generic alert provider for RSGB contest iCal calendars. Builds on the generic iCal alert provider by adding
|
||||
handling specific to how RSGB's iCal events are formatted. This is still effectively an abstract class itself;
|
||||
RSGB has two contest calendars (HF & VHF) that each subclass this."""
|
||||
|
||||
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)
|
||||
def event_to_alert(self, event: Event) -> Alert:
|
||||
"""Convert an iCal event in RSGB's format to an Alert object."""
|
||||
|
||||
# Iterate through events
|
||||
for component in cal.walk():
|
||||
if component.name != "VEVENT":
|
||||
continue
|
||||
summary = str(event.get("summary", "")).strip()
|
||||
|
||||
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
|
||||
|
||||
# Ensure summaries start with "RSGB" to avoid any confusion
|
||||
if not summary.startswith("RSGB "):
|
||||
summary = "RSGB " + summary
|
||||
# Extract freq from summary. 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 freq from summary
|
||||
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"
|
||||
|
||||
# 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 + ", "
|
||||
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
|
||||
|
||||
# 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"
|
||||
# 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,
|
||||
)
|
||||
|
||||
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()
|
||||
return alert
|
||||
|
||||
Reference in New Issue
Block a user