mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-08-06 02:21:42 +00:00
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
import logging
|
|
from datetime import datetime
|
|
|
|
import pytz
|
|
|
|
from core.data_store import DATA_STORE
|
|
|
|
|
|
class CallsignDataProvider:
|
|
"""Generic callsign reference data provider class. Subclasses of this set up the various mechanisms via which
|
|
Spothole can look up data for callsigns."""
|
|
|
|
def __init__(self, name, provider_config, storage):
|
|
"""Constructor. As well as name and config, provide the storage object from DATA_STORE that will be used to
|
|
store the result of lookups to speed up future access."""
|
|
|
|
self.name = name
|
|
self.enabled = provider_config["enabled"]
|
|
self.priority = int(provider_config["priority"])
|
|
self.last_update_time = datetime.min.replace(tzinfo=pytz.UTC)
|
|
self.status = "Not Started" if self.enabled else "Disabled"
|
|
self._storage = storage
|
|
|
|
def start(self):
|
|
"""Start the provider. This should return immediately after spawning threads to access remote resources, if
|
|
needed."""
|
|
|
|
raise NotImplementedError("Subclasses must implement this method")
|
|
|
|
def stop(self):
|
|
"""Stop any threads and prepare for application shutdown"""
|
|
|
|
raise NotImplementedError("Subclasses must implement this method")
|
|
|
|
def lookup(self, callsign, lookup_credentials):
|
|
"""Looks up data for the provided callsign. Takes a LookupCredentials object, which provides any credentials
|
|
that have been provided by the user for this session (QRZ.com/HamQTH) to allow us to look up using those
|
|
services on the user's behalf. (Clublog is looked up using an API key owned by the server and provided in its
|
|
config file, so users need not provide their own.) Returns a Callsign object with as much data populated as
|
|
possible. Data is cached internally for a set period of 30 days to avoid the need to request data from servers
|
|
each time."""
|
|
|
|
if self.enabled:
|
|
if callsign in self._storage:
|
|
return self._storage[callsign]
|
|
else:
|
|
c = self._perform_new_lookup(callsign, lookup_credentials)
|
|
if c:
|
|
self._storage.set(callsign, c, expire=DATA_STORE.CALLSIGN_DATA_TTL_SEC)
|
|
return c
|
|
else:
|
|
return None
|
|
|
|
|
|
def _perform_new_lookup(self, callsign, lookup_credentials):
|
|
"""Makes a new request to the data source for callsign data."""
|
|
|
|
raise NotImplementedError("Subclasses must implement this method")
|