openwebrx-clone/owrx/config.py
2021-02-06 22:08:27 +01:00

181 lines
5.8 KiB
Python

from owrx.property import PropertyLayer
import importlib.util
import os
import logging
import json
from glob import glob
from abc import ABC, abstractmethod
from configparser import ConfigParser
logger = logging.getLogger(__name__)
class ConfigNotFoundException(Exception):
pass
class ConfigError(Exception):
def __init__(self, key, message):
super().__init__("Configuration Error (key: {0}): {1}".format(key, message))
class ConfigMigrator(ABC):
@abstractmethod
def migrate(self, config):
pass
def renameKey(self, config, old, new):
if old in config and new not in config:
config[new] = config[old]
del config[old]
class ConfigMigratorVersion1(ConfigMigrator):
def migrate(self, config):
if "receiver_gps" in config:
gps = config["receiver_gps"]
config["receiver_gps"] = {"lat": gps[0], "lon": gps[1]}
if "waterfall_auto_level_margin" in config:
levels = config["waterfall_auto_level_margin"]
config["waterfall_auto_level_margin"] = {"min": levels[0], "max": levels[1]}
self.renameKey(config, "wsjt_queue_workers", "decoding_queue_workers")
self.renameKey(config, "wsjt_queue_length", "decoding_queue_length")
config["version"] = 2
return config
class ConfigMigratorVersion2(ConfigMigrator):
def migrate(self, config):
if "waterfall_colors" in config and any(v > 0xFFFFFF for v in config["waterfall_colors"]):
config["waterfall_colors"] = [v >> 8 for v in config["waterfall_colors"]]
return config
class CoreConfig(object):
defaults = {
"core": {
"data_directory": "/var/lib/openwebrx",
"temporary_directory": "/tmp",
},
"web": {
"port": 8073,
},
}
def __init__(self):
config = ConfigParser()
overrides_dir = "/etc/openwebrx/openwebrx.conf.d"
if os.path.exists(overrides_dir) and os.path.isdir(overrides_dir):
overrides = glob(overrides_dir + "/*.conf")
else:
overrides = []
# sequence things together
config.read(["./openwebrx.conf", "/etc/openwebrx/openwebrx.conf"] + overrides)
self.data_directory = config.get(
"core", "data_directory", fallback=CoreConfig.defaults["core"]["data_directory"]
)
CoreConfig.checkDirectory(self.data_directory, "data_directory")
self.temporary_directory = config.get(
"core", "temporary_directory", fallback=CoreConfig.defaults["core"]["temporary_directory"]
)
CoreConfig.checkDirectory(self.temporary_directory, "temporary_directory")
self.web_port = config.getint("web", "port", fallback=CoreConfig.defaults["web"]["port"])
@staticmethod
def checkDirectory(dir, key):
if not os.path.exists(dir):
raise ConfigError(key, "{dir} doesn't exist".format(dir=dir))
if not os.path.isdir(dir):
raise ConfigError(key, "{dir} is not a directory".format(dir=dir))
if not os.access(dir, os.W_OK):
raise ConfigError(key, "{dir} is not writable".format(dir=dir))
def get_web_port(self):
return self.web_port
def get_data_directory(self):
return self.data_directory
def get_temporary_directory(self):
return self.temporary_directory
class Config:
sharedConfig = None
currentVersion = 3
migrators = {
1: ConfigMigratorVersion1(),
2: ConfigMigratorVersion2(),
}
@staticmethod
def _loadPythonFile(file):
spec = importlib.util.spec_from_file_location("config_webrx", file)
cfg = importlib.util.module_from_spec(spec)
spec.loader.exec_module(cfg)
pm = PropertyLayer()
for name, value in cfg.__dict__.items():
if name.startswith("__"):
continue
pm[name] = value
return pm
@staticmethod
def _loadJsonFile(file):
with open(file, "r") as f:
pm = PropertyLayer()
for k, v in json.load(f).items():
pm[k] = v
return pm
@staticmethod
def _getSettingsFile():
coreConfig = CoreConfig()
return "{data_directory}/settings.json".format(data_directory=coreConfig.get_data_directory())
@staticmethod
def _loadConfig():
for file in [Config._getSettingsFile(), "/etc/openwebrx/config_webrx.py", "./config_webrx.py"]:
try:
if file.endswith(".py"):
return Config._loadPythonFile(file)
elif file.endswith(".json"):
return Config._loadJsonFile(file)
else:
logger.warning("unsupported file type: %s", file)
except FileNotFoundError:
pass
raise ConfigNotFoundException("no usable config found! please make sure you have a valid configuration file!")
@staticmethod
def get():
if Config.sharedConfig is None:
Config.sharedConfig = Config._migrate(Config._loadConfig())
return Config.sharedConfig
@staticmethod
def store():
with open(Config._getSettingsFile(), "w") as file:
json.dump(Config.get().__dict__(), file, indent=4)
@staticmethod
def validateConfig():
# no config checks atm
# just basic loading verification
Config.get()
@staticmethod
def _migrate(config):
version = config["version"] if "version" in config else 1
if version == Config.currentVersion:
return config
logger.debug("migrating config from version %i", version)
migrators = [Config.migrators[i] for i in range(version, Config.currentVersion)]
for migrator in migrators:
config = migrator.migrate(config)
return config