import re from datetime import timedelta from requests_cache import CachedSession from rss_parser import RSSParser from core.constants import HTTP_HEADERS from data.spot import Spot from spotproviders.http_spot_provider import HTTPSpotProvider # Spot provider for Wainwrights on the Air class WOTA(HTTPSpotProvider): POLL_INTERVAL_SEC = 120 SPOTS_URL = "https://www.wota.org.uk/spots_rss.php" LIST_URL = "https://www.wota.org.uk/mapping/data/summits.json" LIST_CACHE_TIME_DAYS = 30 LIST_CACHE = CachedSession("cache/wota_data_cache", expire_after=timedelta(days=LIST_CACHE_TIME_DAYS)) def __init__(self, provider_config): super().__init__(provider_config, self.SPOTS_URL, self.POLL_INTERVAL_SEC) def http_response_to_spots(self, http_response): new_spots = [] rss = RSSParser.parse(http_response.content.decode()) # Iterate through source data for source_alert in rss.channel.items: break # TODO: Need to see a live spot to know what this feed looks like # Convert to our spot format # spot = Spot(source=self.name, # source_id=source_spot["id"], # dx_call=source_spot["activator"].upper(), # de_call=source_spot["spotter"].upper(), # freq=freq_hz, # mode=source_spot["mode"].upper().strip(), # comment=source_spot["comments"], # sig="WOTA", # sig_refs=[source_spot["reference"]], # icon=get_icon_for_sig("WOTA"), # time=datetime.fromisoformat(source_spot["referenced_time"]).astimezone(pytz.UTC).timestamp()) # WOTA name/lat/lon lookup wota_data = self.LIST_CACHE.get(self.LIST_URL, headers=HTTP_HEADERS).json() for feature in wota_data["features"]: if feature["properties"]["wotaId"] == spot.sig_refs[0]: spot.sig_refs_names = [feature["properties"]["title"]] spot.dx_latitude = feature["geometry"]["coordinates"][1] spot.dx_longitude = feature["geometry"]["coordinates"][0] spot.dx_grid = feature["properties"]["qthLocator"] break return new_spots