mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2025-10-27 16:59:25 +00:00
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
import logging
|
|
from datetime import datetime, timezone
|
|
from threading import Timer, Thread
|
|
from time import sleep
|
|
|
|
import requests
|
|
|
|
from providers.provider import Provider
|
|
|
|
|
|
# Generic data provider class for providers that request data via HTTP(S). Just for convenience to avoid code
|
|
# duplication. Subclasses of this query the individual APIs for data.
|
|
class HTTPProvider(Provider):
|
|
|
|
def __init__(self, url, poll_interval):
|
|
super().__init__()
|
|
self.url = url
|
|
self.poll_interval = poll_interval
|
|
self.poll_timer = None
|
|
|
|
def name(self):
|
|
raise NotImplementedError("Subclasses must implement this method")
|
|
|
|
def start(self):
|
|
# Fire off a one-shot thread to run poll() for the first time, just to ensure start() returns immediately and
|
|
# the application can continue starting. The thread itself will then die, and the timer will kick in on its own
|
|
# thread.
|
|
thread = Thread(target=self.poll)
|
|
thread.start()
|
|
|
|
def stop(self):
|
|
self.poll_timer.cancel()
|
|
|
|
def poll(self):
|
|
try:
|
|
# Request data from API
|
|
http_response = requests.get(self.url, headers=self.HTTP_HEADERS)
|
|
# Pass off to the subclass for processing
|
|
new_spots = self.http_response_to_spots(http_response)
|
|
# Submit the new spots for processing. There might not be any spots for the less popular programs.
|
|
if new_spots:
|
|
self.submit(new_spots)
|
|
|
|
self.status = "OK"
|
|
self.last_update_time = datetime.now(timezone.utc)
|
|
|
|
except Exception as e:
|
|
self.status = "Error"
|
|
logging.exception("Exception in HTTP JSON Provider (" + self.name() + ")")
|
|
sleep(1)
|
|
|
|
self.poll_timer = Timer(self.poll_interval, self.poll)
|
|
self.poll_timer.start()
|
|
|
|
# Convert an HTTP response returned by the API into spot 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, XML, text, whatever
|
|
# the API actually provides.
|
|
def http_response_to_spots(self, http_response):
|
|
raise NotImplementedError("Subclasses must implement this method") |