diff --git a/README.md b/README.md index 3620f56..920e02e 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,7 @@ options: -c, --check run update checks and exit -u, --updatecheck check for updates and exit -w, --wait wait for transactions to complete and exit + --config CONFIG use the specified config file --system only run system updates (requires root) ``` diff --git a/src/ublue_update/cli.py b/src/ublue_update/cli.py index 0bc20bd..27667b7 100644 --- a/src/ublue_update/cli.py +++ b/src/ublue_update/cli.py @@ -9,13 +9,13 @@ ) from ublue_update.update_checks.wait import transaction_wait from ublue_update.update_inhibitors.hardware import check_hardware_inhibitors -from ublue_update.config import load_value +from ublue_update.config import cfg from ublue_update.session import get_xdg_runtime_dir, get_active_sessions from ublue_update.filelock import acquire_lock, release_lock def notify(title: str, body: str, actions: list = [], urgency: str = "normal"): - if not dbus_notify: + if not cfg.dbus_notify: return process_uid = os.getuid() args = [ @@ -58,7 +58,7 @@ def notify(title: str, body: str, actions: list = [], urgency: str = "normal"): def ask_for_updates(system): - if not dbus_notify: + if not cfg.dbus_notify: return out = notify( "System Updater", @@ -159,7 +159,7 @@ def run_updates(system, system_update_available): ) log.debug(out.stdout.decode("utf-8")) log.info("System update complete") - if pending_deployment_check() and system_update_available and dbus_notify: + if pending_deployment_check() and system_update_available and cfg.dbus_notify: out = notify( "System Updater", "System update complete, pending changes will take effect after reboot. Reboot now?", @@ -177,8 +177,6 @@ def run_updates(system, system_update_available): os._exit(0) -dbus_notify: bool = load_value("notify", "dbus_notify") - # setup logging logging.basicConfig( format="[%(asctime)s] %(name)s:%(levelname)s | %(message)s", @@ -212,17 +210,24 @@ def main(): action="store_true", help="wait for transactions to complete and exit", ) + parser.add_argument( + "--config", + help="use the specified config file" + ) parser.add_argument( "--system", action="store_true", help="only run system updates (requires root)", ) cli_args = parser.parse_args() - hardware_checks_failed = False + + # Load the configuration file + cfg.load_config(cli_args.config) if cli_args.wait: transaction_wait() os._exit(0) + system_update_available: bool = system_update_check() if not cli_args.force and not cli_args.updatecheck: hardware_checks_failed, failures = check_hardware_inhibitors() diff --git a/src/ublue_update/config.py b/src/ublue_update/config.py index 07a17ab..8d7826d 100644 --- a/src/ublue_update/config.py +++ b/src/ublue_update/config.py @@ -1,8 +1,13 @@ import tomllib import os +from typing import List, Optional +from logging import getLogger +"""Setup logging""" +log = getLogger(__name__) -def load_config(): + +def find_default_config_file(): # load config values config_paths = [ "/etc/ublue-update/ublue-update.toml", @@ -13,12 +18,12 @@ def load_config(): # first config file that is found wins for path in config_paths: if os.path.isfile(path): - return tomllib.load(open(path, "rb")) + return path - return config + return None -def safe_get_nested(dct, *keys): +def load_value(dct, *keys): for key in keys: try: dct = dct[key] @@ -27,8 +32,25 @@ def safe_get_nested(dct, *keys): return dct -def load_value(*keys): - return safe_get_nested(config, *keys) +class Config: + dbus_notify: bool + network_not_metered: Optional[bool] + min_battery_percent: Optional[float] + max_cpu_load_percent: Optional[float] + max_mem_percent: Optional[float] + + def load_config(self, path=None): + config_path = path or find_default_config_file() + config = tomllib.load(open(config_path, "rb")) + log.debug(f"Configuration loaded from {os.path.abspath(config_path)}") + self.load_values(config) + + def load_values(self, config): + self.dbus_notify = load_value(config, "notify", "dbus_notify") or False + self.network_not_metered = load_value(config, "checks", "network_not_metered") + self.min_battery_percent = load_value(config, "checks", "min_battery_percent") + self.max_cpu_load_percent = load_value(config, "checks", "max_cpu_load_percent") + self.max_mem_percent = load_value(config, "checks", "max_mem_percent") -config = load_config() +cfg = Config() diff --git a/src/ublue_update/update_inhibitors/hardware.py b/src/ublue_update/update_inhibitors/hardware.py index 6820998..c3f77e1 100644 --- a/src/ublue_update/update_inhibitors/hardware.py +++ b/src/ublue_update/update_inhibitors/hardware.py @@ -2,16 +2,11 @@ import subprocess from typing import Optional from logging import getLogger -from ublue_update.config import load_value +from ublue_update.config import cfg """Setup logging""" log = getLogger(__name__) -network_not_metered: Optional[bool] = load_value("checks", "network_not_metered") -min_battery_percent: Optional[float] = load_value("checks", "min_battery_percent") -max_cpu_load_percent: Optional[float] = load_value("checks", "max_cpu_load_percent") -max_mem_percent: Optional[float] = load_value("checks", "max_mem_percent") - def check_network_status() -> dict: network_status = psutil.net_if_stats() @@ -26,7 +21,7 @@ def check_network_status() -> dict: def check_network_not_metered() -> dict: - if not network_not_metered: + if not cfg.network_not_metered: return { "passed": True, "message": "Network metering status is ignored", @@ -63,18 +58,18 @@ def check_network_not_metered() -> dict: def check_battery_status() -> dict: - if min_battery_percent: + if cfg.min_battery_percent: battery_status = psutil.sensors_battery() # null safety on the battery variable, it returns "None" # when the system doesn't have a battery battery_pass: bool = True if battery_status is not None: battery_pass = ( - battery_status.percent >= min_battery_percent or battery_status.power_plugged + battery_status.percent >= cfg.min_battery_percent or battery_status.power_plugged ) return { "passed": battery_pass, - "message": f"Battery less than {min_battery_percent}%", + "message": f"Battery less than {cfg.min_battery_percent}%", } else: return { @@ -84,13 +79,13 @@ def check_battery_status() -> dict: def check_cpu_load() -> dict: - if max_cpu_load_percent: + if cfg.max_cpu_load_percent: # get load average percentage in last 5 minutes: # https://psutil.readthedocs.io/en/latest/index.html?highlight=getloadavg cpu_load_percent = psutil.getloadavg()[1] / psutil.cpu_count() * 100 return { - "passed": cpu_load_percent < max_cpu_load_percent, - "message": f"CPU load is above {max_cpu_load_percent}%", + "passed": cpu_load_percent < cfg.max_cpu_load_percent, + "message": f"CPU load is above {cfg.max_cpu_load_percent}%", } else: return { @@ -100,11 +95,11 @@ def check_cpu_load() -> dict: def check_mem_percentage() -> dict: - if max_mem_percent: + if cfg.max_mem_percent: mem = psutil.virtual_memory() return { - "passed": mem.percent < max_mem_percent, - "message": f"Memory usage is above {max_mem_percent}%", + "passed": mem.percent < cfg.max_mem_percent, + "message": f"Memory usage is above {cfg.max_mem_percent}%", } else: return {