Skip to content

Commit

Permalink
feat: Add --config argument for specifying a configuration file
Browse files Browse the repository at this point in the history
  • Loading branch information
cr7pt0gr4ph7 committed May 22, 2024
1 parent cea7eb8 commit 6e69105
Show file tree
Hide file tree
Showing 4 changed files with 53 additions and 30 deletions.
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
```

Expand Down
19 changes: 12 additions & 7 deletions src/ublue_update/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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?",
Expand All @@ -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",
Expand Down Expand Up @@ -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()
Expand Down
36 changes: 29 additions & 7 deletions src/ublue_update/config.py
Original file line number Diff line number Diff line change
@@ -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",
Expand All @@ -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]
Expand All @@ -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()
27 changes: 11 additions & 16 deletions src/ublue_update/update_inhibitors/hardware.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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",
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand All @@ -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 {
Expand Down

0 comments on commit 6e69105

Please sign in to comment.