mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-08-05 18:11:41 +00:00
81 lines
3.6 KiB
Python
81 lines
3.6 KiB
Python
import logging
|
|
from datetime import datetime
|
|
from threading import Thread, Event
|
|
|
|
import pytz
|
|
from requests import ReadTimeout
|
|
from requests.exceptions import ConnectionError, ConnectTimeout
|
|
|
|
from core.constants import HTTP_HEADERS
|
|
from core.url_data_cache import URLDataCache
|
|
from sigrefdataproviders.sig_ref_data_provider import SIGRefDataProvider
|
|
|
|
|
|
class FileDownloadSIGRefDataProvider(SIGRefDataProvider):
|
|
"""Generic SIG ref data provider class for providers that fetch their data from the web by downloading a file."""
|
|
|
|
def __init__(self, sig_name, provider_config, url, poll_interval):
|
|
""" Set up the provider, note poll_interval is in *days*."""
|
|
super().__init__(sig_name, provider_config)
|
|
self._url = url
|
|
self._poll_interval = poll_interval
|
|
self._thread = None
|
|
self._stop_event = Event()
|
|
self._url_data_cache = URLDataCache("sigrefdata_" + sig_name)
|
|
|
|
def start(self):
|
|
# Fire off the polling thread. It will poll immediately on startup, then sleep for poll_interval between
|
|
# subsequent polls, so start() returns immediately and the application can continue starting.
|
|
logging.info(
|
|
"Set up query of " + self.sig_name + " SIG ref data every " + str(self._poll_interval) + " days.")
|
|
self._thread = Thread(target=self._run, daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
self._stop_event.set()
|
|
|
|
def _run(self):
|
|
while True:
|
|
self._poll()
|
|
if self._stop_event.wait(timeout=self._poll_interval * 60 * 60 * 24):
|
|
break
|
|
|
|
def _poll(self):
|
|
try:
|
|
# Request data from API. Use the data cache (with a TTL of 1 day) here, not as the main mechanism for
|
|
# caching, but just so continual restarts of the software during testing don't hammer the servers.
|
|
logging.debug("Downloading " + self.sig_name + " SIG ref data...")
|
|
http_response = self._url_data_cache.get(self._url, headers=HTTP_HEADERS)
|
|
# Check response code was good
|
|
if http_response.ok:
|
|
# Pass off to the subclass for processing
|
|
new_data = self._http_response_to_data(http_response)
|
|
# Add the new data to the SIG Ref data store
|
|
if new_data:
|
|
self._add_data(new_data)
|
|
|
|
self.status = "OK"
|
|
self.last_update_time = datetime.now(pytz.UTC)
|
|
logging.debug("Received SIG ref data for " + self.sig_name)
|
|
else:
|
|
self.status = "Error"
|
|
logging.warning(f"HTTP {http_response.status_code} when downloading SIG ref data for {self.sig_name}.")
|
|
|
|
except ConnectionError:
|
|
self.status = "Error"
|
|
logging.warning(f"Connection error when downloading SIG ref data for {self.sig_name}.")
|
|
except (ConnectTimeout, ReadTimeout):
|
|
self.status = "Error"
|
|
logging.warning(f"Timeout when downloading SIG ref data for {self.sig_name}.")
|
|
except Exception:
|
|
self.status = "Error"
|
|
logging.exception("Exception in HTTP SIG Ref Data Provider (" + self.sig_name + ")")
|
|
self._stop_event.wait(timeout=1)
|
|
|
|
def _http_response_to_data(self, http_response):
|
|
"""Convert an HTTP response returned by the server into SIG Ref data. The whole response is provided here so the
|
|
subclass implementations can check for HTTP status codes if necessary, and handle the response as JSON, CSV,
|
|
whatever the remote file actually is."""
|
|
|
|
raise NotImplementedError("Subclasses must implement this method")
|