import csv import logging from datetime import datetime, timedelta import pytz import requests from requests_cache import CachedSession from data.spot import Spot from spotproviders.http_spot_provider import HTTPSpotProvider # Spot provider for Parks n Peaks class ParksNPeaks(HTTPSpotProvider): POLL_INTERVAL_SEC = 120 SPOTS_URL = "https://www.parksnpeaks.org/api/ALL" SIOTA_CSV_URL = "https://www.silosontheair.com/data/silos.csv" SIOTA_CSV_CACHE_TIME_DAYS = 30 SIOTA_CSV_CACHE = CachedSession("siota_data_cache", expire_after=timedelta(days=SIOTA_CSV_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 = [] # Iterate through source data for source_spot in http_response.json(): # Convert to our spot format spot = Spot(source=self.name, source_id=source_spot["actID"], dx_call=source_spot["actCallsign"].upper(), de_call=source_spot["actSpoter"].upper(), # typo exists in API freq=float(source_spot["actFreq"].replace(",", "")) * 1000000 if ( source_spot["actFreq"] != "") else None, # Seen PNP spots with empty frequency, and with comma-separated thousands digits mode=source_spot["actMode"].upper(), comment=source_spot["actComments"], sig=source_spot["actClass"], sig_refs=[source_spot["actSiteID"]], time=datetime.strptime(source_spot["actTime"], "%Y-%m-%d %H:%M:%S").replace( tzinfo=pytz.UTC).timestamp()) # PNP supports a bunch of programs which should have different icons if spot.sig == "SiOTA": spot.icon = "wheat-awn" elif spot.sig in ["POTA", "SOTA", "WWFF"]: # Don't care about an icon as this will be rejected anyway, we have better data from POTA/SOTA/WWFF direct spot.icon = "" else: # Unknown programme we've never seen before logging.warn( "PNP spot found with sig " + spot.sig + ", developer needs to add support for icon and grid/lat/lon lookup!") spot.icon = "question" # SiOTA lat/lon/grid lookup if spot.sig == "SiOTA": siota_csv_data = self.SIOTA_CSV_CACHE.get(self.SIOTA_CSV_URL, headers=self.HTTP_HEADERS) siota_dr = csv.DictReader(siota_csv_data.content.decode().splitlines()) for row in siota_dr: if row["SILO_CODE"] == spot.sig_refs[0]: spot.dx_country = row["COUNTRY"] spot.latitude = float(row["LAT"]) spot.longitude = float(row["LON"]) spot.grid = row["LOCATOR"] break # If this is POTA, SOTA or WWFF data we already have it through other means, so ignore. Otherwise, add to # the spot list. if spot.sig not in ["POTA", "SOTA", "WWFF"]: new_spots.append(spot) return new_spots