spothole/sigrefdataproviders/local_file_sig_ref_data_provider.py

import logging
from datetime import datetime

import pytz

from sigrefdataproviders.sig_ref_data_provider import SIGRefDataProvider


class LocalFileSIGRefDataProvider(SIGRefDataProvider):
    """Generic SIG ref data provider class for providers that fetch their data from a local file on startup."""

    def __init__(self, sig, provider_config, path):
        super().__init__(sig, provider_config)
        self._path = path

    def start(self):
        logging.debug("Loading " + self.sig_name + " SIG ref data from file.")
        try:
            new_data = self._file_to_data(self._path)
            if new_data:
                self._add_data(new_data)
                self.status = "OK"
                self.last_update_time = datetime.now(pytz.UTC)
            else:
                self.status = "Error"
                logging.info("Failed to load SIG ref data for " + self.sig_name)
        except Exception as e:
            self.status = "Error"
            logging.error("Exception in local file SIG Ref Data Provider (" + self.sig_name + ")", e, exc_info=True)

    def stop(self):
        pass

    def _file_to_data(self, path):
        """Load a file on the given path and turn it into SIG Ref data."""

        raise NotImplementedError("Subclasses must implement this method")