mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-24 16:24:32 +00:00
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
from datetime import datetime
|
|
|
|
import pytz
|
|
|
|
from core.enums import ActivityName
|
|
from data.activity_ref import ActivityRef
|
|
from data.alert import Alert
|
|
from providers.alert.http_alert_provider import HTTPAlertProvider
|
|
|
|
|
|
class Hamsat(HTTPAlertProvider):
|
|
"""Alert provider for Hamsat (hams.at)"""
|
|
|
|
POLL_INTERVAL_SEC = 1800
|
|
ALERTS_URL = "https://hams.at/api/alerts"
|
|
|
|
def __init__(self, provider_config):
|
|
super().__init__("Hamsat", provider_config, self.ALERTS_URL, self.POLL_INTERVAL_SEC)
|
|
|
|
def _http_response_to_alerts(self, http_response):
|
|
new_alerts = []
|
|
# Iterate through source data
|
|
for source_alert in http_response.json()["data"]:
|
|
# Convert to our alert format
|
|
freqs_modes = source_alert.get("mode") or ""
|
|
mhz = source_alert.get("mhz")
|
|
if mhz is not None:
|
|
mhz_direction = source_alert.get("mhz_direction")
|
|
if mhz_direction is not None:
|
|
freqs_modes = f"{mhz!s} {mhz_direction}, {freqs_modes}"
|
|
else:
|
|
freqs_modes = f"{mhz!s}, {freqs_modes}"
|
|
|
|
alert = Alert(
|
|
source=self.name,
|
|
source_id=source_alert["id"],
|
|
dx_calls=[source_alert["callsign"].upper()],
|
|
dx_grid=source_alert["grids"][0],
|
|
freqs_modes=freqs_modes,
|
|
comment=source_alert["comment"],
|
|
activity=ActivityName.SATELLITE,
|
|
# Fudge an activity ref to provide the remaining bits of data we need: the satellite and the operator's grid
|
|
activity_refs=[
|
|
ActivityRef(
|
|
activity=ActivityName.SATELLITE,
|
|
id=source_alert["satellite"]["name"],
|
|
)
|
|
],
|
|
start_time=datetime.strptime(source_alert["aos_at"], "%Y-%m-%dT%H:%M:%SZ")
|
|
.replace(tzinfo=pytz.UTC)
|
|
.timestamp(),
|
|
end_time=datetime.strptime(source_alert["los_at"], "%Y-%m-%dT%H:%M:%SZ")
|
|
.replace(tzinfo=pytz.UTC)
|
|
.timestamp(),
|
|
)
|
|
|
|
# Add to our list
|
|
new_alerts.append(alert)
|
|
return new_alerts
|