openwebrx-clone/owrx/config.py

143 lines
4.5 KiB
Python
Raw Normal View History

2020-03-25 14:48:27 +00:00
from owrx.property import PropertyManager, PropertyLayer
2019-12-08 16:15:48 +00:00
import importlib.util
import os
import logging
import json
2020-03-29 16:08:26 +00:00
from abc import ABC, abstractmethod
2019-05-10 19:50:58 +00:00
logger = logging.getLogger(__name__)
2019-12-08 16:15:48 +00:00
class ConfigNotFoundException(Exception):
pass
class ConfigError(object):
def __init__(self, key, message):
self.key = key
self.message = message
2019-05-05 17:59:03 +00:00
def __str__(self):
return "Configuration Error (key: {0}): {1}".format(self.key, self.message)
2019-05-09 20:44:29 +00:00
2020-03-29 16:08:26 +00:00
class ConfigMigrator(ABC):
@abstractmethod
def migrate(self, config):
pass
def renameKey(self, config, old, new):
if old in config and not new in config:
config[new] = config[old]
del config[old]
2020-03-29 16:08:26 +00:00
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]}
2020-03-29 16:08:26 +00:00
self.renameKey(config, "wsjt_queue_workers", "decoding_queue_workers")
self.renameKey(config, "wsjt_queue_length", "decoding_queue_length")
2020-03-29 16:08:26 +00:00
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 Config:
sharedConfig = None
currentVersion = 3
2020-03-29 16:08:26 +00:00
migrators = {
1: ConfigMigratorVersion1(),
2: ConfigMigratorVersion2(),
2020-03-29 16:08:26 +00:00
}
2019-05-09 20:44:29 +00:00
@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 _loadConfig():
2020-03-29 16:08:26 +00:00
for file in ["./settings.json", "/etc/openwebrx/config_webrx.py", "./config_webrx.py"]:
2019-12-08 16:15:48 +00:00
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)
2019-12-08 16:15:48 +00:00
except FileNotFoundError:
pass
2019-12-08 16:15:48 +00:00
raise ConfigNotFoundException("no usable config found! please make sure you have a valid configuration file!")
@staticmethod
def get():
if Config.sharedConfig is None:
2020-03-29 16:08:26 +00:00
Config.sharedConfig = Config._migrate(Config._loadConfig())
return Config.sharedConfig
@staticmethod
def store():
with open("settings.json", "w") as file:
json.dump(Config.get().__dict__(), file, indent=4)
@staticmethod
def validateConfig():
pm = Config.get()
2021-01-20 16:01:46 +00:00
errors = [Config.checkTempDirectory(pm)]
return [e for e in errors if e is not None]
@staticmethod
2020-03-25 14:47:15 +00:00
def checkTempDirectory(pm: PropertyManager):
key = "temporary_directory"
if key not 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
2020-03-29 16:08:26 +00:00
@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