openwebrx-clone/owrx/config.py

import importlib.util
import logging
import os

logger = logging.getLogger(__name__)


class Subscription(object):
    def __init__(self, subscriptee, subscriber):
        self.subscriptee = subscriptee
        self.subscriber = subscriber

    def call(self, *args, **kwargs):
        self.subscriber(*args, **kwargs)

    def cancel(self):
        self.subscriptee.unwire(self)


class Property(object):
    def __init__(self, value=None):
        self.value = value
        self.subscribers = []

    def getValue(self):
        return self.value

    def setValue(self, value):
        if self.value == value:
            return self
        self.value = value
        for c in self.subscribers:
            try:
                c.call(self.value)
            except Exception as e:
                logger.exception(e)
        return self

    def wire(self, callback):
        sub = Subscription(self, callback)
        self.subscribers.append(sub)
        if not self.value is None:
            sub.call(self.value)
        return sub

    def unwire(self, sub):
        try:
            self.subscribers.remove(sub)
        except ValueError:
            # happens when already removed before
            pass
        return self


class ConfigNotFoundException(Exception):
    pass


class PropertyManager(object):
    sharedInstance = None

    @staticmethod
    def getSharedInstance():
        if PropertyManager.sharedInstance is None:
            PropertyManager.sharedInstance = PropertyManager()
        return PropertyManager.sharedInstance

    def collect(self, *props):
        return PropertyManager(
            {name: self.getProperty(name) if self.hasProperty(name) else Property() for name in props}
        )

    def __init__(self, properties=None):
        self.properties = {}
        self.subscribers = []
        if properties is not None:
            for (name, prop) in properties.items():
                self.add(name, prop)

    def add(self, name, prop):
        self.properties[name] = prop

        def fireCallbacks(value):
            for c in self.subscribers:
                try:
                    c.call(name, value)
                except Exception as e:
                    logger.exception(e)

        prop.wire(fireCallbacks)
        return self

    def __contains__(self, name):
        return self.hasProperty(name)

    def __getitem__(self, name):
        return self.getPropertyValue(name)

    def __setitem__(self, name, value):
        if not self.hasProperty(name):
            self.add(name, Property())
        self.getProperty(name).setValue(value)

    def __dict__(self):
        return {k: v.getValue() for k, v in self.properties.items()}

    def hasProperty(self, name):
        return name in self.properties

    def getProperty(self, name):
        if not self.hasProperty(name):
            self.add(name, Property())
        return self.properties[name]

    def getPropertyValue(self, name):
        return self.getProperty(name).getValue()

    def wire(self, callback):
        sub = Subscription(self, callback)
        self.subscribers.append(sub)
        return sub

    def unwire(self, sub):
        try:
            self.subscribers.remove(sub)
        except ValueError:
            # happens when already removed before
            pass
        return self

    def defaults(self, other_pm):
        for (key, p) in self.properties.items():
            if p.getValue() is None:
                p.setValue(other_pm[key])
        return self

    def loadConfig(self):
        for file in ["/etc/openwebrx/config_webrx.py", "./config_webrx.py"]:
            try:
                spec = importlib.util.spec_from_file_location("config_webrx", file)
                cfg = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(cfg)
                for name, value in cfg.__dict__.items():
                    if name.startswith("__"):
                        continue
                    self[name] = value
                return self
            except FileNotFoundError:
                pass
        raise ConfigNotFoundException("no usable config found! please make sure you have a valid configuration file!")


class ConfigError(object):
    def __init__(self, key, message):
        self.key = key
        self.message = message

    def __str__(self):
        return "Configuration Error (key: {0}): {1}".format(self.key, self.message)


class Config:
    @staticmethod
    def validateConfig():
        pm = PropertyManager.getSharedInstance()
        errors = [
            Config.checkTempDirectory(pm)
        ]

        return [e for e in errors if e is not None]

    @staticmethod
    def checkTempDirectory(pm: PropertyManager):
        key = "temporary_directory"
        if not key in pm or pm[key] is None:
            return ConfigError(key, "temporary directory is not set")
        if not os.path.exists(pm[key]):
            return ConfigError(key, "temporary directory doesn't exist")
        if not os.path.isdir(pm[key]):
            return ConfigError(key, "temporary directory path is not a directory")
        if not os.access(pm[key], os.W_OK):
            return ConfigError(key, "temporary directory is not writable")
        return None