From b5a869eadee63615644e6e195a3befaee00ac4f7 Mon Sep 17 00:00:00 2001 From: Zen Date: Fri, 10 Jul 2026 21:51:33 -0500 Subject: [PATCH 01/20] generalize the flow for finding module files Signed-off-by: Zen --- src/ugrd/config_helpers.py | 10 ++++++++++ src/ugrd/initramfs_config.py | 29 +++++++++++------------------ 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/src/ugrd/config_helpers.py b/src/ugrd/config_helpers.py index 02e97ecb..d1f5734e 100644 --- a/src/ugrd/config_helpers.py +++ b/src/ugrd/config_helpers.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Any from zenlib.util import parse_toml @@ -23,6 +24,15 @@ def get_module_name(module_path: Path) -> str: return parent_name + module_name +def read_ugrd_module(module_name: str) -> dict[str, Any]: + """Reads a ugrd module given a module name. Returns the config""" + for module in get_module_paths(): + if module_name == get_module_name(module): + return parse_toml(module) + + raise FileNotFoundError(f"Unable to find module: {module_name}") + + def get_parameters() -> dict[str, dict[str, str]]: """Returns a dictionary of modules and their corresponding variables these are defined in the "custom_parameters" section of the config file diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index dfa21c2b..aa9d664a 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -16,6 +16,7 @@ from zenlib.util import handle_plural, pretty_print from .exceptions import ValidationError +from .config_helpers import read_ugrd_module class InitramfsConfig(LoggerMixIn, UserDict): @@ -348,29 +349,21 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: function.__name__.removeprefix("_process_") ) # Re-process any queued values + + + @handle_plural def _process_modules(self, module: str) -> None: - """processes a single module into the config""" + """processes a single module into the config + If that module (by name) has already been loaded, does nothing + + """ if module in self["modules"]: - self.logger.debug("Module '%s' already loaded" % module) + self.logger.debug(f"Module already loaded: {c_(module, 'yellow')}") return - self.logger.info("Processing module: %s" % c_(module, bold=True)) - - module_subpath = module.replace(".", "/") + ".toml" - - module_path = Path(__file__).parent.parent / module_subpath - if not module_path.exists(): - module_path = Path("/var/lib/ugrd") / module_subpath - if not module_path.exists(): - raise FileNotFoundError("Unable to locate module: %s" % module) - self.logger.debug("Module path: %s" % module_path) - - with open(module_path, "rb") as module_file: - try: - module_config = load(module_file) - except TOMLDecodeError as e: - raise TOMLDecodeError("Unable to load module config: %s" % module) from e + module_config = read_ugrd_module(module) + self.logger.info(f"Processing module: {c_(module, bold=True)}") if imports := module_config.get("imports"): self.logger.debug("[%s] Processing imports: %s" % (module, imports)) From f88a0d9120580f716f3d4a5cad0e64fd38bf79f2 Mon Sep 17 00:00:00 2001 From: Zen Date: Fri, 10 Jul 2026 21:52:02 -0500 Subject: [PATCH 02/20] fix formatting Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index aa9d664a..3c0dbff7 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -6,7 +6,6 @@ from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path from queue import Queue -from tomllib import TOMLDecodeError, load from typing import Callable from pycpio import PyCPIO @@ -15,8 +14,8 @@ from zenlib.util import colorize as c_ from zenlib.util import handle_plural, pretty_print -from .exceptions import ValidationError from .config_helpers import read_ugrd_module +from .exceptions import ValidationError class InitramfsConfig(LoggerMixIn, UserDict): @@ -273,7 +272,9 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: # Extra checks are added for logging but mostly to make type checking happy spec = spec_from_file_location(module_name, module_path) if spec is None: - raise ModuleNotFoundError(f"[{c_(module_name, 'yellow')}] Failed to load spec from file: {c_(module_path, 'red')}") + raise ModuleNotFoundError( + f"[{c_(module_name, 'yellow')}] Failed to load spec from file: {c_(module_path, 'red')}" + ) spec_loader = spec.loader if spec_loader is None: @@ -282,7 +283,9 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: module = module_from_spec(spec) if module is None: - raise ModuleNotFoundError(f"[{c_(module_name, 'yellow')}] Failed to load module from spec: {c_(spec, 'red')}") + raise ModuleNotFoundError( + f"[{c_(module_name, 'yellow')}] Failed to load module from spec: {c_(spec, 'red')}" + ) spec_loader.exec_module(module) self.logger.debug(f"Loaded external module: {c_(module_name, 'blue')}") @@ -349,9 +352,6 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: function.__name__.removeprefix("_process_") ) # Re-process any queued values - - - @handle_plural def _process_modules(self, module: str) -> None: """processes a single module into the config From 4915dcd57b561f063a9cd5abd8dc59ca01d29ba2 Mon Sep 17 00:00:00 2001 From: Zen Date: Fri, 10 Jul 2026 22:34:31 -0500 Subject: [PATCH 03/20] simplify config loading Signed-off-by: Zen --- src/ugrd/initramfs_generator.py | 39 ++++++++++++++------------------- 1 file changed, 17 insertions(+), 22 deletions(-) diff --git a/src/ugrd/initramfs_generator.py b/src/ugrd/initramfs_generator.py index e3976cf6..126f717a 100644 --- a/src/ugrd/initramfs_generator.py +++ b/src/ugrd/initramfs_generator.py @@ -1,11 +1,10 @@ from importlib.metadata import version from textwrap import dedent -from tomllib import TOMLDecodeError, load from typing import Any from zenlib.logging import LoggerMixIn from zenlib.util import colorize as c_ -from zenlib.util import pretty_print +from zenlib.util import parse_toml, pretty_print from ugrd import InitramfsConfig @@ -30,18 +29,19 @@ def __init__(self, config="/etc/ugrd/config.toml", *args, **kwargs) -> None: # Passed kwargs must be imported early, so they will be processed against the base configuration self.config_dict.import_args(kwargs) - try: # Attempt to load the config file, if it exists + + # don't attempt to load config if not specified + if not config: + self.logger.info("No config file specified, using the base config.") + return + + try: self.load_config(config) # The user config is loaded over the base config, clobbering kwargs self.config_dict.import_args( kwargs, quiet=True, late=True ) # Re-import kwargs (cmdline params) to apply them over the config except FileNotFoundError: - if config: # If a config file was specified, log an error that it's missing - self.logger.critical("[%s] Config file not found, using the base config." % config) - else: # Otherwise, log info that the base config is being used - self.logger.info("No config file specified, using the base config.") - except TOMLDecodeError as e: - raise ValueError("[%s] Error decoding config file: %s" % (config, e)) + self.logger.critical(f"[{c_(config, 'red')}] Config file not found, using the base config.") def load_config(self, config_filename) -> None: """ @@ -49,22 +49,17 @@ def load_config(self, config_filename) -> None: Populates self.config_dict with the config. Ensures that the required parameters are present. """ - if not config_filename: - raise FileNotFoundError("Config file not specified.") + self.logger.info(f"Loading config file: {c_(config_filename, 'blue', bold=True, bright=True)}") + raw_config = parse_toml(config_filename) - with open(config_filename, "rb") as config_file: - self.logger.info("Loading config file: %s" % c_(config_file.name, "blue", bold=True, bright=True)) - raw_config = load(config_file) - - # Process into the config dict, it should handle parsing + # Process all config into the config dict, which will handle processing/validation for config, value in raw_config.items(): - self.logger.debug("[%s] (%s) Processing config value: %s" % (config_file.name, config, value)) - try: - self[config] = value - except FileNotFoundError as e: - raise ValueError("[%s] Error loading config parameter '%s': %s" % (config_file.name, config, e)) + self.logger.debug( + f"[{c_(config_filename, 'blue')}] ({c_(config, bold=True)}) Processing config value: {c_(value, 'green')}" + ) + self[config] = value - self.logger.debug("Loaded config:\n%s" % self.config_dict) + self.logger.debug(f"Loaded config:\n{self.config_dict}") # If the initramfs generator is used as a dictionary, it will use the config_dict. def __setitem__(self, key, value) -> None: From 4ded80f66135392a2810bee7c16a83ef4addcc76 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 00:07:00 -0500 Subject: [PATCH 04/20] generalize the module loading function Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 3c0dbff7..f8297be3 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -6,7 +6,7 @@ from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path from queue import Queue -from typing import Callable +from typing import Any, Callable from pycpio import PyCPIO from zenlib.logging import LoggerMixIn @@ -354,15 +354,26 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: @handle_plural def _process_modules(self, module: str) -> None: - """processes a single module into the config + """processes a single module (by name) into the config If that module (by name) has already been loaded, does nothing - """ if module in self["modules"]: self.logger.debug(f"Module already loaded: {c_(module, 'yellow')}") return - module_config = read_ugrd_module(module) + self._load_module(read_ugrd_module(module), module) + + def _load_module(self, module_config: dict[str, Any], module: str = "config") -> None: + """Loads a module given a config dict module_config + the module var is used for logging + + 'imports' are registered first as they may have processing functions + Checks needs before processing the module further + + Adds unregistered values to the processing queue if they are not lists/dicts + + Finally, registers custom values so queued values can be processed and registers provided tags if present + """ self.logger.info(f"Processing module: {c_(module, bold=True)}") if imports := module_config.get("imports"): @@ -403,7 +414,9 @@ def _process_modules(self, module: str) -> None: self._process_unprocessed(custom_parameter) # Append the module to the list of loaded modules, avoid recursion - self["modules"].append(module) + # Do not do this for modules called the default name 'config' + if module != "config": + self["modules"].append(module) # Handle provides tags, ensure only a single module provides a tag if provides := module_config.get("provides"): From 36e14ce684e1995400252bd96cf87dfc9413e5a5 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 00:10:49 -0500 Subject: [PATCH 05/20] use fstrings / color for config dict logging Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index f8297be3..c4638149 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -377,36 +377,36 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> self.logger.info(f"Processing module: {c_(module, bold=True)}") if imports := module_config.get("imports"): - self.logger.debug("[%s] Processing imports: %s" % (module, imports)) + self.logger.debug(f"[{c_(module, 'green')}] Processing imports: {imports}") self["imports"] = imports if needs := module_config.get("needs"): if isinstance(needs, str): if needs not in self["provided"]: - raise ValueError("[%s] Required tag not provided: %s" % (module, needs)) + raise ValueError(f"[{c_(module, 'green')}] Required tag not provided: {c_(needs, 'red')}") elif isinstance(needs, list): for need in needs: if need not in self["provided"]: - raise ValueError("[%s] Required tag not provided: %s" % (module, need)) + raise ValueError(f"[{c_(module, 'green')}] Required tag not provided: {c_(need, 'red')}") else: - raise ValueError("[%s] Invalid needs value: %s" % (module, needs)) + raise ValueError(f"[{c_(module, 'green')}] Invalid needs value: {c_(needs, 'red')}") custom_parameters = module_config.get("custom_parameters", {}) # Process other config such as import orders, defined values for name, value in module_config.items(): if name in ["imports", "custom_parameters", "provides", "needs"]: - self.logger.log(5, "[%s] Skipping '%s'" % (module, name)) + self.logger.log(5, f"[{c_(module, 'green')}] Skipping: {c_(name, 'yellow')}") continue if name in self["_processing"] and custom_parameters.get(name) not in ["list", "NoDupFlatList", "dict"]: self.logger.debug(f"Skipping setting defaults for parameter with queued values: {c_(name, 'yellow')}") else: - self.logger.debug("[%s] (%s) Setting value: %s" % (module, name, value)) + self.logger.debug(f"[{c_(module, 'green')}] ({c_(name, bold=True)}) Setting value: {c_(value, 'blue')}") self[name] = value # Add custom parameters after values are added, so they are processed in the correct order if custom_parameters: - self.logger.debug("[%s] Processing custom parameters: %s" % (module, custom_parameters)) + self.logger.debug(f"[{c_(module, 'green')}] Processing custom parameters: {custom_parameters}") self["custom_parameters"] = custom_parameters # If custom parameters were added, process unprocessed values From 6c28e1875d2889274b6a6ea4009ad2b16570c405 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 14:03:13 -0500 Subject: [PATCH 06/20] bup dep ver need higher zenlib version for parse_toml Signed-off-by: Zen --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 1eccc7f0..7f6e9516 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ classifiers = [ ] dependencies = [ - "zenlib >= 3.3.0", + "zenlib >= 3.5.0", "pycpio >= 1.7.0" ] From d71fe55e6f8cc4a65b2c42a8ebb65530e59ab25a Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 00:15:09 -0500 Subject: [PATCH 07/20] move initramfs config file processing to the config dict keeps config processing where it should be Signed-off-by: Zen --- src/ugrd/config_helpers.py | 1 + src/ugrd/initramfs_config.py | 36 ++++++++++++++++++++++++--- src/ugrd/initramfs_generator.py | 44 ++++++--------------------------- 3 files changed, 41 insertions(+), 40 deletions(-) diff --git a/src/ugrd/config_helpers.py b/src/ugrd/config_helpers.py index d1f5734e..679c2812 100644 --- a/src/ugrd/config_helpers.py +++ b/src/ugrd/config_helpers.py @@ -3,6 +3,7 @@ from zenlib.util import parse_toml +DEFAULT_CONFIG_PATH = "/etc/ugrd/config.toml" MODULE_SEARCH_PATHS = [Path(__file__).parent, Path("/var/lib/ugrd")] diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index c4638149..14628147 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -12,9 +12,9 @@ from zenlib.logging import LoggerMixIn from zenlib.types import NoDupFlatList from zenlib.util import colorize as c_ -from zenlib.util import handle_plural, pretty_print +from zenlib.util import handle_plural, parse_toml, pretty_print -from .config_helpers import read_ugrd_module +from .config_helpers import DEFAULT_CONFIG_PATH, read_ugrd_module from .exceptions import ValidationError @@ -45,7 +45,14 @@ class InitramfsConfig(LoggerMixIn, UserDict): "test_copy_config": NoDupFlatList, # A list of config values which are copied into test images, from the parent } - def __init__(self, NO_BASE=False, *args, **kwargs): + def __init__( + self, + startup_args: dict | None = None, + config_file: Path | str | None = None, + NO_BASE: bool = False, + *args, + **kwargs, + ) -> None: self.init_logger(args, kwargs) super().__init__(*args, **kwargs) @@ -56,11 +63,34 @@ def __init__(self, NO_BASE=False, *args, **kwargs): else: self.data[parameter] = default_type() self["import_order"] = {"before": {}, "after": {}} + if not NO_BASE: self["modules"] = "ugrd.base.base" else: self["modules"] = "ugrd.base.core" + # If startup args are defined, process them + if startup_args: + self.import_args(startup_args) + + # If config is defined, load everything but the modules defined in it + if config_file: + try: + self._load_module(parse_toml(config_file)) + except FileNotFoundError as e: + if str(config_file) == DEFAULT_CONFIG_PATH: + self.logger.warning( + f"Using base config because default config not found: {c_(config_file, 'yellow')}" + ) + else: + raise e + else: + self.logger.info("No config file specified, using the base config.") + + # Import args again so they take precedence over other config + if startup_args: + self.import_args(startup_args, quiet=True, late=True) + def import_args(self, args: dict, quiet=False, late=False) -> None: """Imports data from an argument dict.""" log_level = 10 if quiet else 20 diff --git a/src/ugrd/initramfs_generator.py b/src/ugrd/initramfs_generator.py index 126f717a..b20f5977 100644 --- a/src/ugrd/initramfs_generator.py +++ b/src/ugrd/initramfs_generator.py @@ -1,21 +1,25 @@ from importlib.metadata import version +from pathlib import Path from textwrap import dedent from typing import Any from zenlib.logging import LoggerMixIn from zenlib.util import colorize as c_ -from zenlib.util import parse_toml, pretty_print +from zenlib.util import pretty_print from ugrd import InitramfsConfig +from .config_helpers import DEFAULT_CONFIG_PATH from .exceptions import ValidationError from .generator_helpers import GeneratorHelpers class InitramfsGenerator(GeneratorHelpers, LoggerMixIn): - def __init__(self, config="/etc/ugrd/config.toml", *args, **kwargs) -> None: + def __init__(self, config: Path | str | None = DEFAULT_CONFIG_PATH, *args, **kwargs) -> None: self.init_logger(args, kwargs) - self.config_dict = InitramfsConfig(NO_BASE=kwargs.pop("NO_BASE", False), logger=self.logger) + self.config_dict = InitramfsConfig( + NO_BASE=kwargs.pop("NO_BASE", False), logger=self.logger, startup_args=kwargs, config_file=config + ) # Used for functions that are added to the shell profile # The key name is the function name, the value is the content @@ -27,40 +31,6 @@ def __init__(self, config="/etc/ugrd/config.toml", *args, **kwargs) -> None: # init_pre and init_final are run as part of generate_initramfs_main self.init_types = ["init_debug", "init_main", "init_mount"] - # Passed kwargs must be imported early, so they will be processed against the base configuration - self.config_dict.import_args(kwargs) - - # don't attempt to load config if not specified - if not config: - self.logger.info("No config file specified, using the base config.") - return - - try: - self.load_config(config) # The user config is loaded over the base config, clobbering kwargs - self.config_dict.import_args( - kwargs, quiet=True, late=True - ) # Re-import kwargs (cmdline params) to apply them over the config - except FileNotFoundError: - self.logger.critical(f"[{c_(config, 'red')}] Config file not found, using the base config.") - - def load_config(self, config_filename) -> None: - """ - Loads the config from the specified toml file. - Populates self.config_dict with the config. - Ensures that the required parameters are present. - """ - self.logger.info(f"Loading config file: {c_(config_filename, 'blue', bold=True, bright=True)}") - raw_config = parse_toml(config_filename) - - # Process all config into the config dict, which will handle processing/validation - for config, value in raw_config.items(): - self.logger.debug( - f"[{c_(config_filename, 'blue')}] ({c_(config, bold=True)}) Processing config value: {c_(value, 'green')}" - ) - self[config] = value - - self.logger.debug(f"Loaded config:\n{self.config_dict}") - # If the initramfs generator is used as a dictionary, it will use the config_dict. def __setitem__(self, key, value) -> None: self.config_dict[key] = value From 83ec40e6611e25c4bffbf3728a1cf63e464d639a Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 13:12:37 -0500 Subject: [PATCH 08/20] don't try to process queued values after importing a processor doesn't do anything because this happens before custom_paramaters are defined if the custom parameters were already defined, the queue should not have values in it Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 14628147..4fccca43 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -377,10 +377,7 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: if import_type == "config_processing": # Register the functions for processing after all imports are done for function in function_list: self["custom_processing"][function.__name__] = function - self.logger.debug("Registered config processing function: %s" % function.__name__) - self._process_unprocessed( - function.__name__.removeprefix("_process_") - ) # Re-process any queued values + self.logger.debug(f"Registered config processing function: {c_(function.__name__, 'blue')}") @handle_plural def _process_modules(self, module: str) -> None: @@ -395,7 +392,7 @@ def _process_modules(self, module: str) -> None: def _load_module(self, module_config: dict[str, Any], module: str = "config") -> None: """Loads a module given a config dict module_config - the module var is used for logging + the module var is used for logging and tracking loaded modules 'imports' are registered first as they may have processing functions Checks needs before processing the module further From 13bd2fcb592eee58122702c3de0097052a0a1bde Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 13:34:17 -0500 Subject: [PATCH 09/20] remove unused/pointless import function processor Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 36 +++++++++++------------------------- 1 file changed, 11 insertions(+), 25 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 4fccca43..fd24339a 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -6,7 +6,7 @@ from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path from queue import Queue -from typing import Any, Callable +from typing import Any from pycpio import PyCPIO from zenlib.logging import LoggerMixIn @@ -264,28 +264,6 @@ def _process_import_order(self, import_order: dict) -> None: self.logger.debug("Registered import order requirements: %s" % import_order) - def _process_import_functions(self, module, functions: list) -> list[Callable]: - """Processes defined import functions, importing them and adding them to the returned list. - The 'function' key is required if dicts are used, - 'before' and 'after' keys can be used to specify order requirements.""" - function_list = [] - for f in functions: - match type(f).__name__: - case "str": - function_list.append(getattr(module, f)) - case "dict": - if "function" not in f: - raise ValueError("Function key not found in import dict: %s" % functions) - func = f["function"] - function_list.append(getattr(module, func)) - if "before" in f: - self["import_order"] = {"before": {func: f["before"]}} - if "after" in f: - self["import_order"] = {"after": {func: f["after"]}} - case _: - raise ValueError("Invalid type for import function: %s" % type(f)) - return function_list - @handle_plural def _process_imports(self, import_type: str, import_value: dict) -> None: """Processes imports in a module, importing the functions and adding them to the appropriate list.""" @@ -343,9 +321,17 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: self.logger.warning("Skipping custom init function: %s" % mask) continue - function_list = self._process_import_functions(module, function_names) + try: + function_list = [getattr(module, func) for func in function_names] + except AttributeError as e: + raise AttributeError( + f"[{c_(module_name, 'yellow')}] Failed to import function: {c_(e.name, 'red')}" + ) from e + if not function_list: - self.logger.warning("[%s] No functions found for import: %s" % (module_name, import_type)) + self.logger.warning( + f"[{c_(module_name, 'yellow')}] No functions found for import: {c_(import_type, 'red')}" + ) continue if import_type == "custom_init": # Only get the first function for custom init (should be 1) From 406992bc7250dd0b0ac9e5afb0b6875daf6cb55f Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 13:43:06 -0500 Subject: [PATCH 10/20] split module sideloading logic into dedicated method Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 66 ++++++++++++++++++++---------------- 1 file changed, 37 insertions(+), 29 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index fd24339a..8d3d69e1 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -264,6 +264,41 @@ def _process_import_order(self, import_order: dict) -> None: self.logger.debug("Registered import order requirements: %s" % import_order) + def _import_external_module(self, module_name: str): + """Given a module name, attempts to load it from /var/lib/ugrd, returning the module""" + module_path = Path("/var/lib/ugrd/" + module_name.replace(".", "/")).with_suffix(".py") + self.logger.debug(f"Attempting to sideload module from: {c_(module_path, 'green')}") + if not module_path.exists(): + raise ModuleNotFoundError(f"Module not found: {c_(module_name, 'red')}") + + try: # If the module is not built in, try to load it from /var/lib/ugrd + # Extra checks are added for logging but mostly to make type checking happy + spec = spec_from_file_location(module_name, module_path) + if spec is None: + raise ModuleNotFoundError( + f"[{c_(module_name, 'yellow')}] Failed to load spec from file: {c_(module_path, 'red')}" + ) + + spec_loader = spec.loader + if spec_loader is None: + raise RuntimeError(f"Failed to initialize loader for spec: {c_(spec, 'yellow')}") + + module = module_from_spec(spec) + + if module is None: + raise ModuleNotFoundError( + f"[{c_(module_name, 'yellow')}] Failed to load module from spec: {c_(spec, 'red')}" + ) + + spec_loader.exec_module(module) + self.logger.debug(f"Loaded external module: {c_(module_name, 'blue')}") + + except Exception as e: + raise ModuleNotFoundError(f"[{c_(module_name, 'yellow')}] Unable to load module: {e}") from e + + self.logger.info(f"Sideloaded module: {c_(module_name, 'green')}") + return module + @handle_plural def _process_imports(self, import_type: str, import_value: dict) -> None: """Processes imports in a module, importing the functions and adding them to the appropriate list.""" @@ -271,35 +306,8 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: self.logger.debug("[%s]<%s> Importing module functions : %s" % (module_name, import_type, function_names)) try: # First, the module must be imported, so its functions can be accessed module = import_module(module_name) - except ModuleNotFoundError as e: - module_path = Path("/var/lib/ugrd/" + module_name.replace(".", "/")).with_suffix(".py") - self.logger.debug("Attempting to sideload module from: %s" % module_path) - if not module_path.exists(): - raise ModuleNotFoundError("Module not found: %s" % module_name) from e - try: # If the module is not built in, try to load it from /var/lib/ugrd - # Extra checks are added for logging but mostly to make type checking happy - spec = spec_from_file_location(module_name, module_path) - if spec is None: - raise ModuleNotFoundError( - f"[{c_(module_name, 'yellow')}] Failed to load spec from file: {c_(module_path, 'red')}" - ) - - spec_loader = spec.loader - if spec_loader is None: - raise RuntimeError(f"Failed to initialize loader for spec: {c_(spec, 'yellow')}") - - module = module_from_spec(spec) - - if module is None: - raise ModuleNotFoundError( - f"[{c_(module_name, 'yellow')}] Failed to load module from spec: {c_(spec, 'red')}" - ) - - spec_loader.exec_module(module) - self.logger.debug(f"Loaded external module: {c_(module_name, 'blue')}") - - except Exception as e: - raise ModuleNotFoundError(f"[{c_(module_name, 'yellow')}] Unable to load module: {e}") from e + except ModuleNotFoundError: # If it can't be natively imported, try to sideload it + module = self._import_external_module(module_name) self.logger.log(5, "[%s] Imported module contents: %s" % (module_name, dir(module))) if "_module_name" in dir(module) and module._module_name != module_name: From 96e5002b73e48285721b10842062cda9a483003a Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 13:46:10 -0500 Subject: [PATCH 11/20] remove pointless else statement Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 8d3d69e1..b5a2066b 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -118,19 +118,19 @@ def __setitem__(self, key: str, value) -> None: # If the type is registered, use the appropriate update function if any(key in d for d in (self.builtin_parameters, self["custom_parameters"])): return self.handle_parameter(key, value) - else: + + self.logger.debug( + f"[{c_(key, 'yellow')}] Unable to determine expected type, valid builtin types:\n{c_(self.builtin_parameters.keys(), bold=True)}" + ) + self.logger.debug("f[{c_(key, 'blue')}] Custom types: {c_(self['custom_parameters'].keys(), bold=True')}") + # for anything but the logger, add to the processing queue + if key != "logger": self.logger.debug( - f"[{c_(key, 'yellow')}] Unable to determine expected type, valid builtin types:\n{c_(self.builtin_parameters.keys(), bold=True)}" + f"[{c_(key, 'yellow')}] Adding unknown internal parameter to processing queue: {c_(value, bold=True)}" ) - self.logger.debug("f[{c_(key, 'blue')}] Custom types: {c_(self['custom_parameters'].keys(), bold=True')}") - # for anything but the logger, add to the processing queue - if key != "logger": - self.logger.debug( - f"[{c_(key, 'yellow')}] Adding unknown internal parameter to processing queue: {c_(value, bold=True)}" - ) - if key not in self["_processing"]: - self["_processing"][key] = Queue() - self["_processing"][key].put(value) + if key not in self["_processing"]: + self["_processing"][key] = Queue() + self["_processing"][key].put(value) def handle_parameter(self, key: str, value) -> None: """ @@ -181,9 +181,8 @@ def check_mask(import_name: str) -> bool: if key not in self: self.logger.log(5, "Setting dict '%s' to: %s" % (key, value)) return super().__setitem__(key, value) - else: - self.logger.log(5, "Updating dict '%s' with: %s" % (key, value)) - return self[key].update(value) + self.logger.log(5, "Updating dict '%s' with: %s" % (key, value)) + return self[key].update(value) self.logger.debug("Setting custom parameter: %s" % key) self.data[key] = expected_type(value) # For everything else, simply set it @@ -419,6 +418,7 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> if name in ["imports", "custom_parameters", "provides", "needs"]: self.logger.log(5, f"[{c_(module, 'green')}] Skipping: {c_(name, 'yellow')}") continue + if name in self["_processing"] and custom_parameters.get(name) not in ["list", "NoDupFlatList", "dict"]: self.logger.debug(f"Skipping setting defaults for parameter with queued values: {c_(name, 'yellow')}") else: From 39b047826b05512e8951bbe47ce52c396cd8b9db Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 14:27:37 -0500 Subject: [PATCH 12/20] simplify custom parameter processing when custom parameters are defined, immediately process queued values Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index b5a2066b..edcca999 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -220,6 +220,9 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) self.logger.warning("Leaving '%s' as None" % parameter_name) self.data[parameter_name] = None + # Process queued values if they exist + self._process_unprocessed(parameter_name) + def _process_unprocessed(self, parameter_name: str) -> None: """Processes queued values for a parameter.""" if parameter_name not in self["_processing"]: @@ -411,29 +414,20 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> else: raise ValueError(f"[{c_(module, 'green')}] Invalid needs value: {c_(needs, 'red')}") - custom_parameters = module_config.get("custom_parameters", {}) - # Process other config such as import orders, defined values for name, value in module_config.items(): if name in ["imports", "custom_parameters", "provides", "needs"]: self.logger.log(5, f"[{c_(module, 'green')}] Skipping: {c_(name, 'yellow')}") continue - if name in self["_processing"] and custom_parameters.get(name) not in ["list", "NoDupFlatList", "dict"]: - self.logger.debug(f"Skipping setting defaults for parameter with queued values: {c_(name, 'yellow')}") - else: - self.logger.debug(f"[{c_(module, 'green')}] ({c_(name, bold=True)}) Setting value: {c_(value, 'blue')}") - self[name] = value + self.logger.debug(f"[{c_(module, 'green')}] ({c_(name, bold=True)}) Setting value: {c_(value, 'blue')}") + self[name] = value - # Add custom parameters after values are added, so they are processed in the correct order - if custom_parameters: + # If custom parameters are defined, process them and then process any unprocessed values + if custom_parameters := module_config.get("custom_parameters", {}): self.logger.debug(f"[{c_(module, 'green')}] Processing custom parameters: {custom_parameters}") self["custom_parameters"] = custom_parameters - # If custom parameters were added, process unprocessed values - for custom_parameter in custom_parameters: - self._process_unprocessed(custom_parameter) - # Append the module to the list of loaded modules, avoid recursion # Do not do this for modules called the default name 'config' if module != "config": From 03a910309e10abda2bb6868714dbd27651c17840 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 14:33:31 -0500 Subject: [PATCH 13/20] improve log formatting remove check for validated, if it was validated then writing would not be allowed anyways Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 87 +++++++++++++++++++++--------------- 1 file changed, 52 insertions(+), 35 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index edcca999..d53d64e7 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -119,14 +119,15 @@ def __setitem__(self, key: str, value) -> None: if any(key in d for d in (self.builtin_parameters, self["custom_parameters"])): return self.handle_parameter(key, value) - self.logger.debug( - f"[{c_(key, 'yellow')}] Unable to determine expected type, valid builtin types:\n{c_(self.builtin_parameters.keys(), bold=True)}" + self.logger.log( + 5, + f"[{c_(key, 'yellow')}] Unable to determine expected type, valid builtin types:\n{c_(self.builtin_parameters.keys(), 'blue', bold=True)}", ) - self.logger.debug("f[{c_(key, 'blue')}] Custom types: {c_(self['custom_parameters'].keys(), bold=True')}") + self.logger.log(5, f"[{c_(key, 'blue')}] Custom types: {c_(self['custom_parameters'].keys(), bold=True)}") # for anything but the logger, add to the processing queue if key != "logger": self.logger.debug( - f"[{c_(key, 'yellow')}] Adding unknown internal parameter to processing queue: {c_(value, bold=True)}" + f"[{c_(key, 'blue', background=True)}] Adding unknown internal parameter to processing queue: {c_(value, 'yellow')}" ) if key not in self["_processing"]: self["_processing"][key] = Queue() @@ -145,13 +146,13 @@ def handle_parameter(self, key: str, value) -> None: if expected_type: if expected_type.__name__ == "InitramfsGenerator": self.data[key] = value - return self.logger.debug("Setting InitramfsGenerator: %s" % key) + return self.logger.debug(f"Setting InitramfsGenerator: {c_(key, 'magenta', bold=True)}") break # Break and raise an exception if the type is not found else: - raise KeyError("Parameter not registered: %s" % key) + raise KeyError(f"Parameter not registered: {c_(key, 'red')}") if hasattr(self, f"_process_{key}"): # The builtin function is decorated and can handle plural - self.logger.log(5, "[%s] Using builtin setitem: %s" % (key, f"_process_{key}")) + self.logger.log(5, f"[{c_(key, 'blue')}] Using builtin setitem: _process_{key}") return getattr(self, f"_process_{key}")(value) # Don't use masked processing functions for custom values, fall back to standard setters @@ -161,31 +162,39 @@ def check_mask(import_name: str) -> bool: if func := self["custom_processing"].get(f"_process_{key}"): if check_mask(func.__name__): - self.logger.debug("Skipping masked function: %s" % func.__name__) + self.logger.debug(f"Skipping masked function: {c_(func.__name__, 'yellow', background=True)}") else: - self.logger.log(5, "[%s] Using custom setitem: %s" % (key, func.__name__)) + self.logger.log( + 5, f"[{c_(key, 'blue')}] Using custom setitem: {c_(func.__name__, 'blue', underline=True)}" + ) return func(self, value) if func := self["custom_processing"].get(f"_process_{key}_multi"): if check_mask(func.__name__): - self.logger.debug("Skipping masked function: %s" % func.__name__) + self.logger.debug(f"Skipping masked function: {c_(func.__name__, 'yellow', background=True)}") else: - self.logger.log(5, "[%s] Using custom plural setitem: %s" % (key, func.__name__)) + self.logger.log( + 5, + f"[{c_(key, 'blue')}] Using custom plural setitem: {c_(func.__name__, 'blue', underline=True, bold=True)}", + ) return handle_plural(func)(self, value) if expected_type in (list, NoDupFlatList): # Append to lists, don't replace - self.logger.log(5, "Using list setitem for: %s" % key) + self.logger.log(5, f"[{c_(key, 'blue')}] Using list setitem") return self[key].append(value) if expected_type is dict: # Create new keys, update existing if key not in self: - self.logger.log(5, "Setting dict '%s' to: %s" % (key, value)) + self.logger.log(5, f"[{c_(key, 'blue')}] Setting dict to: {value}") return super().__setitem__(key, value) - self.logger.log(5, "Updating dict '%s' with: %s" % (key, value)) + self.logger.log(5, f"[{c_(key, 'blue')}] Updating dict with: {value}") return self[key].update(value) - self.logger.debug("Setting custom parameter: %s" % key) - self.data[key] = expected_type(value) # For everything else, simply set it + casted_value = expected_type(value) + self.logger.debug( + f"[{c_(key, 'blue')}]{c_(expected_type, 'red', dim=True)} Setting value: {c_(casted_value, bold=True)}" + ) + self.data[key] = casted_value # For everything else, simply set it @handle_plural def _process_custom_parameters(self, parameter_name: str, parameter_type: type) -> None: @@ -197,7 +206,9 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) parameter_type = eval(parameter_type) self["custom_parameters"][parameter_name] = parameter_type - self.logger.debug("Registered custom parameter '%s' with type: %s" % (parameter_name, parameter_type)) + self.logger.debug( + f"[{c_(parameter_name, 'blue')}] Registered custom parameter with type: {c_(parameter_type, 'red', dim=True)}" + ) match parameter_type.__name__: case "NoDupFlatList": @@ -217,7 +228,9 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) case "PyCPIO": self.data[parameter_name] = PyCPIO(logger=self.logger, _log_bump=10) case _: # For strings and things, don't init them so they are None - self.logger.warning("Leaving '%s' as None" % parameter_name) + self.logger.warning( + f"[{c_(parameter_name, 'blue')}] Leaving unknown parameter type as None! <{c_(parameter_type.__name__, 'red')}>" + ) self.data[parameter_name] = None # Process queued values if they exist @@ -226,16 +239,15 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) def _process_unprocessed(self, parameter_name: str) -> None: """Processes queued values for a parameter.""" if parameter_name not in self["_processing"]: - self.logger.log(5, "No queued values for: %s" % parameter_name) + self.logger.log(5, f"No queued values for: {c_(parameter_name, 'yellow', dim=True)}") return value_queue = self["_processing"].pop(parameter_name) while not value_queue.empty(): value = value_queue.get() - if self["validated"]: # Log at info level if the config has been validated - self.logger.info("[%s] Processing queued value: %s" % (parameter_name, value)) - else: - self.logger.debug("[%s] Processing queued value: %s" % (parameter_name, value)) + self.logger.debug( + f"[{c_(parameter_name, 'blue')}] Processing queued value: {c_(value, 'yellow', bold=True)}" + ) self[parameter_name] = value def _process_import_order(self, import_order: dict) -> None: @@ -305,18 +317,21 @@ def _import_external_module(self, module_name: str): def _process_imports(self, import_type: str, import_value: dict) -> None: """Processes imports in a module, importing the functions and adding them to the appropriate list.""" for module_name, function_names in import_value.items(): - self.logger.debug("[%s]<%s> Importing module functions : %s" % (module_name, import_type, function_names)) + f_name = f"[{c_(module_name, 'green')}]({c_(import_type, underline=True)})" + self.logger.debug(f"{f_name} Importing module functions: {function_names}") try: # First, the module must be imported, so its functions can be accessed module = import_module(module_name) except ModuleNotFoundError: # If it can't be natively imported, try to sideload it module = self._import_external_module(module_name) - self.logger.log(5, "[%s] Imported module contents: %s" % (module_name, dir(module))) + self.logger.log(5, f"{f_name} Imported module contents:{dir(module)}") if "_module_name" in dir(module) and module._module_name != module_name: - self.logger.warning("Module name mismatch: %s != %s" % (module._module_name, module_name)) + self.logger.warning( + f"Module name mismatch: {c_(module._module_name, 'green')} != {c_(module_name, 'red')}" + ) if import_type not in self["imports"]: # Import types are only actually created when needed - self.logger.log(5, "Creating import type: %s" % import_type) + self.logger.log(5, f"Creating import type: {c_(import_type, underline=True)}") self["imports"][import_type] = NoDupFlatList(_log_bump=10, logger=self.logger) if import_masks := self.get("masks", {}).get(import_type, []): @@ -328,7 +343,7 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: ) function_names.remove(mask) if import_type == "custom_init": - self.logger.warning("Skipping custom init function: %s" % mask) + self.logger.warning(f"Skipping masked custom init function: {c_(mask, 'yellow')}") continue try: @@ -351,24 +366,26 @@ def _process_imports(self, import_type: str, import_value: dict) -> None: custom_init = function_list if self["imports"]["custom_init"]: - self.logger.warning("Custom init function already defined: %s" % self["imports"]["custom_init"]) + self.logger.warning( + f"Custom init function already defined: {c_(self['imports']['custom_init'], 'yellow')}" + ) else: self["imports"]["custom_init"] = custom_init self.logger.info( - "Registered custom init function: %s" % c_(custom_init.__name__, "blue", bold=True) + f"[{c_(module_name, 'green')}] Registered custom init function: {c_(custom_init.__name__, 'blue', bold=True)}" ) continue if import_type == "funcs": # Check for collisions with defined binaries and functions for function in function_list: if function.__name__ in self["imports"]["funcs"]: - raise ValueError("Function '%s' already registered" % function.__name__) + raise ValueError(f"Function already registered: {c_(function.__name__, 'red')}") if function.__name__ in self["binaries"]: - raise ValueError("Function collides with defined binary: %s'" % function.__name__) + raise ValueError(f"Function collides with defined binary: {c_(function.__name__, 'red')}") # Append the functions to the appropriate list self["imports"][import_type] += function_list - self.logger.debug("[%s] Updated import functions: %s" % (import_type, function_list)) + self.logger.log(5, f"[{c_(import_type, underline=True)}] Updated import functions: {function_list}") if import_type == "config_processing": # Register the functions for processing after all imports are done for function in function_list: @@ -397,7 +414,7 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> Finally, registers custom values so queued values can be processed and registers provided tags if present """ - self.logger.info(f"Processing module: {c_(module, bold=True)}") + self.logger.info(f"Processing module: {c_(module, 'green', bold=True)}") if imports := module_config.get("imports"): self.logger.debug(f"[{c_(module, 'green')}] Processing imports: {imports}") @@ -420,7 +437,7 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> self.logger.log(5, f"[{c_(module, 'green')}] Skipping: {c_(name, 'yellow')}") continue - self.logger.debug(f"[{c_(module, 'green')}] ({c_(name, bold=True)}) Setting value: {c_(value, 'blue')}") + self.logger.log(5, f"[{c_(module, 'green')}]({c_(name, 'blue')}) Setting value: {c_(value, bold=True)}") self[name] = value # If custom parameters are defined, process them and then process any unprocessed values From fd385b08ab319fcb3e80017d5bb91effd0c64d3b Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 15:52:49 -0500 Subject: [PATCH 14/20] simplify config ordering process instead of loading multiple times, just load things from the base up this sets the stage for removing late_args (possibly) but mostly for making early arg loading possible Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 33 ++++++++++++++++----------------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index d53d64e7..981ac7b7 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -53,6 +53,17 @@ def __init__( *args, **kwargs, ) -> None: + """Initialize the initramfs config + + First, init the logger and UserDict superclass + Then, define all default parameters + + With those defined, the base config can be loaded. + As config modules are loaded, parameters which are not built in are queued for later processing + The last queued value will take precedence + + The order is: base config -> user config -> arguments + """ self.init_logger(args, kwargs) super().__init__(*args, **kwargs) @@ -69,10 +80,6 @@ def __init__( else: self["modules"] = "ugrd.base.core" - # If startup args are defined, process them - if startup_args: - self.import_args(startup_args) - # If config is defined, load everything but the modules defined in it if config_file: try: @@ -87,28 +94,20 @@ def __init__( else: self.logger.info("No config file specified, using the base config.") - # Import args again so they take precedence over other config + # Import args last so they take precedence over other config if startup_args: - self.import_args(startup_args, quiet=True, late=True) + self.import_args(startup_args) - def import_args(self, args: dict, quiet=False, late=False) -> None: + def import_args(self, args: dict) -> None: """Imports data from an argument dict.""" - log_level = 10 if quiet else 20 for arg, value in args.items(): - self.logger.log(log_level, f"[{c_(arg, 'blue')}] Setting from arguments: {c_(value, 'green')}") + self.logger.info(f"[{c_(arg, 'blue')}] Setting from arguments: {c_(value, 'blue', bold=True)}") if arg == "modules": # allow loading modules by name from the command line for module in value.split(","): self[arg] = module - elif arg in self["_late_args"] and not late: - self.logger.debug( - f"[{c_(arg, 'yellow')}] Deferring late argument processing until after config load: {c_(value, 'green')}", - ) - continue - elif getattr(self, arg, None) != value: # Only set the value if it differs: - self[arg] = value else: - self.logger.debug("Skipping unchanged argument '%s' with value: %s" % (arg, value)) + self[arg] = value def __setitem__(self, key: str, value) -> None: if self["validated"]: From b106532167b4d700048a92a006e1a5f46e3e7a01 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 16:58:26 -0500 Subject: [PATCH 15/20] Add config stage system used to ensure late_args are only processed once the stage is set to the late stage the late stage is started by the generator before the build process starts. At the start of the stage, all enqueued late args are processed this should ensure that args like "kernel_version" are processed after parameters like "no_kmod" are set anywhere Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 88 +++++++++++++++++++++++++++++---- src/ugrd/initramfs_generator.py | 4 +- 2 files changed, 82 insertions(+), 10 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 981ac7b7..5543512f 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -42,6 +42,7 @@ class InitramfsConfig(LoggerMixIn, UserDict): "custom_processing": dict, # Custom processing functions which will be run to validate and process parameters "_processing": dict, # A dict of queues containing parameters which have been set before the type was known "_late_args": NoDupFlatList, # A list of arguments which could be passed as command line args but need to be processed after the config is loaded + "stage": str, # The current processing stage (early, late, final) "test_copy_config": NoDupFlatList, # A list of config values which are copied into test images, from the parent } @@ -73,6 +74,7 @@ def __init__( self.data[parameter] = default_type(no_warn=True, _log_bump=5, logger=self.logger) else: self.data[parameter] = default_type() + self["stage"] = "early" self["import_order"] = {"before": {}, "after": {}} if not NO_BASE: @@ -109,11 +111,38 @@ def import_args(self, args: dict) -> None: else: self[arg] = value + def _enqueue(self, key: str, value: Any) -> None: + """Adds a value to the processing queue""" + if key not in self["_processing"]: + self["_processing"][key] = Queue() + self["_processing"][key].put(value) + self.logger.debug( + f"[{c_(key, 'blue', background=True)}] Adding parameter to processing queue: {c_(value, 'yellow')}" + ) + def __setitem__(self, key: str, value) -> None: + """Custom setitem for the config dict + If the dict is validated, refuse to set config + If it's the final stage and it's not validated, raise a critical warning + + If a _late_arg is being set and it's not in the late stage, also add it to the queue + + For registered config values, use the handle_parameter function + + For everything but the logger, queue values if they are not registered + """ if self["validated"]: return self.logger.error( f"[{c_(key, 'yellow')}] Config is validated, refusing to set value: {c_(value, 'red')}" ) + if self["stage"] == "final" and not self["validated"]: + return self.logger.critical( + f"[{c_(key, 'yellow')}] Config is finalized but invalid, refusing to set value: {c_(value, 'red')}" + ) + + if not self._check_late(key): + return self._enqueue(key, value) + # If the type is registered, use the appropriate update function if any(key in d for d in (self.builtin_parameters, self["custom_parameters"])): return self.handle_parameter(key, value) @@ -125,18 +154,15 @@ def __setitem__(self, key: str, value) -> None: self.logger.log(5, f"[{c_(key, 'blue')}] Custom types: {c_(self['custom_parameters'].keys(), bold=True)}") # for anything but the logger, add to the processing queue if key != "logger": - self.logger.debug( - f"[{c_(key, 'blue', background=True)}] Adding unknown internal parameter to processing queue: {c_(value, 'yellow')}" - ) - if key not in self["_processing"]: - self["_processing"][key] = Queue() - self["_processing"][key].put(value) + self._enqueue(key, value) def handle_parameter(self, key: str, value) -> None: """ Handles a config parameter, setting the value and processing it if the type is known. Raises a KeyError if the parameter is not registered. + If the value is a _late_arg, and it's not in the late stage, skip it + Uses custom processing functions if they are defined, otherwise uses the standard setters. """ # Get the expected type, first searching builtin_parameters, then custom_parameters @@ -200,6 +226,8 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) """ Updates the custom_parameters attribute. Sets the initial value of the parameter based on the type. + + Processes any queued values if they exist (unless they are _late_args and it is not the late stage) """ if isinstance(parameter_type, str): parameter_type = eval(parameter_type) @@ -236,11 +264,19 @@ def _process_custom_parameters(self, parameter_name: str, parameter_type: type) self._process_unprocessed(parameter_name) def _process_unprocessed(self, parameter_name: str) -> None: - """Processes queued values for a parameter.""" + """Processes queued values for a parameter. + Does nothing if there are no queued values + + If the value is a _late_arg and it's not the late stage, skip + this leaves the processing queue untouched + """ if parameter_name not in self["_processing"]: self.logger.log(5, f"No queued values for: {c_(parameter_name, 'yellow', dim=True)}") return + if not self._check_late(parameter_name): + return + value_queue = self["_processing"].pop(parameter_name) while not value_queue.empty(): value = value_queue.get() @@ -460,13 +496,47 @@ def _load_module(self, module_config: dict[str, Any], module: str = "config") -> self["provided"] = tag self.logger.info(f"[{c_(module, bright=True)}] Registered provided tag: {c_(tag, 'green', bold=True)}") - def validate(self) -> None: + def _check_late(self, parameter_name: str) -> bool: + """Checks if it is time to run a late arg""" + if self["stage"] != "late" and parameter_name in self["_late_args"]: + self.logger.debug( + f"[{c_(self['stage'], underline=True)}] Deferring processing for late arg: {c_(parameter_name, 'yellow')}" + ) + return False + return True + + def _process_late_values(self) -> None: + """Processes all late values""" + if self["stage"] != "late": + raise RuntimeError( + f"Cannot process late values in current stage: {c_(self['stage'], 'red', underline=True)}" + ) + + for arg in self["_late_args"]: + self._process_unprocessed(arg) + + def _process_stage(self, stage: str) -> None: + """Sets the stage + When the stage is changed to final, calls validation and does not allow it to be changed again + """ + if self["stage"] == "final": + raise RuntimeError("Cannot change stage after finalized") + + self.data["stage"] = stage + self.logger.debug(f"Entering config stage: {c_(stage, 'blue', underline=True, bold=True)}") + + if stage == "late": + self._process_late_values() + elif stage == "final": + self._validate() + + def _validate(self) -> None: """Validate config, checks that all values are processed, sets validated flag.""" if self["_processing"]: return self.logger.critical( "Unprocessed config values: %s" % c_(", ".join(list(self["_processing"].keys())), "red", bold=True) ) - self["validated"] = True + self.data["validated"] = True def __str__(self) -> str: return pretty_print(self.data) diff --git a/src/ugrd/initramfs_generator.py b/src/ugrd/initramfs_generator.py index b20f5977..2af49476 100644 --- a/src/ugrd/initramfs_generator.py +++ b/src/ugrd/initramfs_generator.py @@ -52,10 +52,12 @@ def __getattr__(self, item) -> Any: def build(self) -> None: """Builds the initramfs image.""" + self.config_dict["stage"] = "late" # Set the config stage to late, loading deferred config self._log_run(f"Running ugrd v{version('ugrd')}") self.run_build() - self.config_dict.validate() # Validate the config after the build tasks have been run + self.config_dict["stage"] = "final" # Finalize the config, triggering validation + # If validation fails, raise an error if validation mode is enabled if self.validate and not self.validated: raise ValidationError( f"Failed to validate config. Unprocessed values: {', '.join(list(self['_processing'].keys()))}" From 60bc44065861bfece51d629699c3b53291078c2c Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 17:17:51 -0500 Subject: [PATCH 16/20] move validation logic into the config dict Signed-off-by: Zen --- src/ugrd/initramfs_config.py | 9 ++++++--- src/ugrd/initramfs_generator.py | 6 ------ 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/ugrd/initramfs_config.py b/src/ugrd/initramfs_config.py index 5543512f..19ce05b6 100644 --- a/src/ugrd/initramfs_config.py +++ b/src/ugrd/initramfs_config.py @@ -533,9 +533,12 @@ def _process_stage(self, stage: str) -> None: def _validate(self) -> None: """Validate config, checks that all values are processed, sets validated flag.""" if self["_processing"]: - return self.logger.critical( - "Unprocessed config values: %s" % c_(", ".join(list(self["_processing"].keys())), "red", bold=True) - ) + unprocessed_values = ", ".join(list(self["_processing"].keys())) + if self["validate"]: + raise ValidationError( + f"Failed to validate config. Unprocessed values: {c_(unprocessed_values, 'red', bold=True)}" + ) + return self.logger.critical(f"Unprocessed config values: {c_(unprocessed_values, 'red', bold=True)}") self.data["validated"] = True def __str__(self) -> str: diff --git a/src/ugrd/initramfs_generator.py b/src/ugrd/initramfs_generator.py index 2af49476..5a52b2da 100644 --- a/src/ugrd/initramfs_generator.py +++ b/src/ugrd/initramfs_generator.py @@ -57,12 +57,6 @@ def build(self) -> None: self.run_build() self.config_dict["stage"] = "final" # Finalize the config, triggering validation - # If validation fails, raise an error if validation mode is enabled - if self.validate and not self.validated: - raise ValidationError( - f"Failed to validate config. Unprocessed values: {', '.join(list(self['_processing'].keys()))}" - ) - self.generate_init() self.pack_build() self.run_checks() From 4daa1abf855c04e9f397eea4bddfa41dac5b3eb8 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 17:21:18 -0500 Subject: [PATCH 17/20] add binaries to late args sets the stage for the busybox module, or any other module which may want to change how binaries are processed Signed-off-by: Zen --- src/ugrd/base/core.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/src/ugrd/base/core.toml b/src/ugrd/base/core.toml index 2e4e63c3..8293b692 100644 --- a/src/ugrd/base/core.toml +++ b/src/ugrd/base/core.toml @@ -11,6 +11,7 @@ validate = true library_paths = [ "/lib64", "/lib" ] old_count = 1 timeout = 15 +_late_args = ["binaries"] [nodes.console] mode = 0o644 From 1abc56ae2a743fa9aadcbda68ca5f776afd18b1f Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 17:34:38 -0500 Subject: [PATCH 18/20] add the debug editor to _late_args because it depends on the binaries helper for error handling, it must be used later for proper error handling Signed-off-by: Zen --- src/ugrd/base/debug.toml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/ugrd/base/debug.toml b/src/ugrd/base/debug.toml index 85227804..7c308801 100644 --- a/src/ugrd/base/debug.toml +++ b/src/ugrd/base/debug.toml @@ -1,3 +1,4 @@ +_late_args = ['editor'] binaries = ['cp', 'ln', 'mv', 'rm', 'find', 'grep', 'dmesg', 'chmod', 'sleep', 'touch'] dependencies = [ "/usr/share/terminfo/l/linux" ] # required by most editors # EDITOR is determined at build time using (in order): @@ -30,5 +31,5 @@ log_file = "ugrd-debug.log" start_shell = "bool" # Start a shell after init_early, before init_pre. Can be enabled by the debug cmdline option. debug_tty2 = "bool" # If true, debug shell will be started on tty2 instead of the current tty. debug_parallel = "bool" # If true, debug shell will be started in parallel to normal init instead of pausing it -editor = "str" # override editor variable +editor = "str" # override editor variable, uses EDITOR from the env or nano if that is undefined no_validate_editor = "bool" # will skip validation of the editor binary, when validation is in use. Otherwise does nothing. From ff75b4ecdc80fe8a22f80d46db913f0b676686f6 Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 18:16:44 -0500 Subject: [PATCH 19/20] modernize dev manual Signed-off-by: Zen --- docs/dev_manual.md | 170 ++++++++++++++++++++++++++++----------------- 1 file changed, 108 insertions(+), 62 deletions(-) diff --git a/docs/dev_manual.md b/docs/dev_manual.md index b9c5997e..e07d860a 100644 --- a/docs/dev_manual.md +++ b/docs/dev_manual.md @@ -1,89 +1,94 @@ # Dev manual -Modules can be created to extend the functionality of the initramfs generator. +Modules are defined using TOML and extend the functionality of the initramfs generator. -Modules only require a TOML definition, and can import other modules to act as meta-modules. +> Modules do not need to have executable code and can serve as meta-modules referencing sets of other modules and setting config -Python functions can be added imported into `init` and `build` runlevels to execute build tasks or output init lines. +Functionality is added by defining `imports` in modules which define functions which can modify all build/config/init generation. > `build` functions are allowed to mutate config, init functions are not. Init is the final build task where bash files are generated. -Within modules, all config values are imported, then processed according to the order of the `custom_parameters` list. +# Configuration Architecture -> If config values have validation which may fail if other config is not loaded, those values can be added to the `_late_args' list to be processed last. +All config is stored in a magic UserDict, not really but it may seem that way. -`_module_name` can be set within a module for logging purposes, it is verified to be accurate when imported but optional. +Essentially all parameters are defined in modules, with a few being built in, mostly to hold the structure for the import system and parameter registration. -## Imports +Parameters defined within modules are imported after `modules` and `imports`, and the last module (highest level) setting a value will take precedence. -µgRD allows python functions to be imported from modules using the `imports` dict. +## Custom parameters -`imports` entries have a key which is the name of the hook to import into, and a value which is a dict of module names and lists of functions to import. +Parameters are defined in the `custom_parameters` dict, where the key is the name, and the value is the expected type. -## Import order +The order of this dict defines the order in which enqueued values will be processed (where relevant) -Imports can be ordered using the `import_order` dict. +### Parameter Initialization -The key name defined in `before` will be run before values in the `after` list (value) by name. +All parameters are initialized to 'empty/zero' values, this is defined in the `_process_custom_parameters` builtin. -Likewise, keys in the `after` list will be run after the key in the `before` value. +### Modifying parameters -> `after` targets are moved before the key when creating the hook order, not literally after. +Scalar values can be updated normally, but this can be changed with a `custom_processing` function. -### Import types +Iterables such as dicts and lists are automatically appended/updated unless a `custom_processing` function is associated with that parameter. -There are two primary categories for imports, `build` and `init`. Build imports are used to mutate the config and build the base structure of the initramfs, while init imports are used to generate the init scripts. +### late_args -`config_processing` imports are used to automatically process config values when they are modified at runtime. +Parameters defined in `_late_args` will only be loaded just before the build phase (after all defined modules are loaded). -The `pack` import is primarily used for packing the CPIO archive. +This can be used to ensure parameters with dynamic processors can change functionality based on config which may be defined in other modules. -The `checks` import is used for static checks, such as ensuring required files are included in the CPIO and have reasonable contents. +A key example is the `no_kmod` parameter which could interfere with a `kernel_version` being set by tooling such as installkernel on a system where no kernel modules are needed. -The `test` import is used for testing the initramfs, and is mostly used by the `test` module for QEMU wrapping. +Another example is the `binaries` parameter which may need to be short circuited for certain utilities if busybox is used. -### Importing functions +# Modules -Functions are imported from modules by specifying the hook they are to be added to in the `imports` dict, with the module name as the key and a list of functions to import as the value. +All µgRD components take the form of modules. Modules consist of a TOML definition which may `import` code from a python file. -For example, the `generate_fstab` function is added to the `build_tasks` book from the `ugrd.fs.mounts` module with: +`_module_name` can be set within a module for logging purposes, it is verified to be accurate when imported but optional. -``` -[imports.build_tasks] -"ugrd.fs.mounts" = [ "generate_fstab" ] -``` +## Imports -## Build imports +µgRD allows python functions to be imported from modules using the `imports` dict. -Build imports are used to mutate config and build the base structure of the initramfs. +`imports` entries have a key which is the name of the hook to import into, and a value which is a dict of module names and lists of functions to import. -### build_enum +For example, to import function `get_foo` into runlevel `bar` from module `baz`: -`build_enum` is used for system enumeration, such as finding the root device, loaded kernel mods, etc. +``` +[imports.bar] +"baz" = ["get_foo"] +``` -### build_pre +### Import hooks (types) -`build_pre` contains build tasks which are run at the very start of the build, such as build directory cleaning and additional config processing based on enumerated info. +There are two primary categories for imports, `build` and `init`. -### build_tasks +`build` imports are used to mutate the config and build the base structure of the initramfs. -`build_tasks` are functions which will be executed after `build_pre`, which make up the majority of the build process. +`init` imports are used to generate the init scripts. -### build_late +For config which requires special validation or handling, `config_processing` hooks can be made to process parameters as soon as they are set. -`build_late` are finalizing build functions, immediately before files are deployed +The `pack`, `checks`, and `test` hooks should be self explanatory and are explained below. -### build_deploy +#### Build imports -`build_deploy` is mostly for builtin functions and is where components are actually copied into the build directory. +Build imports are used to mutate config and build the base structure of the initramfs. -### build_final +The following hooks are used internally and are defined in `InitramfsGenerator.build_tasks`: -`build_final` is the last build hook, where finalizing tasks take place. +* `build_enum` - Used for system enumeration, such as finding the root device, loaded kernel mods, etc. +* `build_pre` - For build tasks run at the very start of the build, such as directory cleaning and possibly late enumeration/config processing. +* `build_tasks` - Functions which will be executed after `build_pre`, which make up the majority of the build process. +* `build_deploy` Where components are actually copied/created in the build directory. +* `build_final` The last default build hook, where finalizing tasks such as image metadata regeneration take place. ## Init imports By default, the following init hooks are available: + * `init_pre` - Where the base initramfs environment is set up; basic mounts are initialized and the kernel cmdline is read. * `init_debug` - Where a shell is started if `start_shell` is enabled in the debug module. * `init_main` - Most important initramfs activities should take place here. @@ -154,24 +159,16 @@ def console_init(self) -> str: return out_str ``` -## pack +### Config processing -Packing functions, such as CPIO generation can be defined in the `pack` import. - -The `cpio` module imports the `make_cpio_list` packing function with: - -``` -[imports.pack] -"ugrd.fs.cpio" = [ "make_cpio" ] -``` -## Config processing - -`config_processing` imports are different from typical imports. They are configured similarly, with a dict of module names and functions to import. +`config_processing` imports are used to automatically process config values when they are modified at runtime. -Instead of running once at a particular build level, `config_processing` functions are run whenever a config value is modified at runtime. +While defined similarly to other imports, they are not associated with any runlevel and run whenever the associated parameter is modified. This can be used to validate config values, or to automatically process them. +> Functions added to `config_processing` should be named in the format `_process_{,_multi}` where _multi is added to run values through @handle_plural + A good example of this is in `base.py`: ``` @@ -193,19 +190,69 @@ def _process_mounts_multi(self, key, mount_config): This module manages mount management, and loads new mounts into fstab objects, also defined in the base module. -The name of `config_prcessing` functions is very important, it must be formatted like `_process_{name}` where the name is the root variable name in the TOML config. +A new root variable named `foo` could be defined, and a function `_process_foo` could be created and imported, raising an error when this value is found, for example. -If the function name has `_multi` at the end, it will be called using the `handle_plural` function, iterating over passed lists/dicts automatically. - -A new root variable named `oops` could be defined, and a function `_process_oops` could be created and imported, raising an error when this value is found, for example. - -This module is loaded in the imports section of the `base.yaml` file: +This module is loaded in the imports section of the `base.toml` file: ``` [imports.config_processing] "ugrd.fs.mounts" = [ "_process_mounts_multi" ] ``` +### Pack + +The `pack` import is primarily used for packing the CPIO archive. + +The `cpio` module imports the `make_cpio_list` packing function with: + +``` +[imports.pack] +"ugrd.fs.cpio" = [ "make_cpio" ] +``` + +### Checks + +The `checks` import is used for static checks, such as ensuring required files are included in the CPIO and have reasonable contents. + +### Test + +The `test` import is used for testing the initramfs, and is mostly used by the `test` module for QEMU wrapping. + +## Importing functions + +Functions are imported from modules by specifying the hook they are to be added to in the `imports` dict, with the module name as the key and a list of functions to import as the value. + +For example, the `generate_fstab` function is added to the `build_tasks` book from the `ugrd.fs.mounts` module with: + +``` +[imports.build_tasks] +"ugrd.fs.mounts" = [ "generate_fstab" ] +``` + +## Import order + +Imports can be ordered using the `import_order` dict. + +The key name defined in `before` will be run before values in the `after` list (value) by name. + +Likewise, keys in the `after` list will be run after the key in the `before` value. + +> `after` targets are moved before the key when creating the hook order, not literally after. + +For example, to run function "foo" before function "bar": + +``` +[import_order.before] +"foo" = "bar" +``` + +To run function "baz" after "foo" and "bar": + +``` +[import_order.after] +"baz" = ["foo", "bar"] +``` + ## Provides/needs Modules can provide/need a certain "tag" to be set by other modules. @@ -217,8 +264,7 @@ When a module has a `needs` string or list of strings, those will be checked aga Needed tags are checked after module imports and before any module config. Provided tags are set upon successful module import. - -## Example module +# Example module The following is an example module which prints "hello world" during the init process: ``` From 0bdedd75642a6ee2824a12740d95d5212bccdd8e Mon Sep 17 00:00:00 2001 From: Zen Date: Sat, 11 Jul 2026 23:42:31 -0500 Subject: [PATCH 20/20] add basic tests to test config dict functionality Signed-off-by: Zen --- tests/test_config.py | 49 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 tests/test_config.py diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..8fa10f37 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,49 @@ +from unittest import TestCase, main + +from ugrd.exceptions import ValidationError +from ugrd.initramfs_generator import InitramfsConfig +from zenlib.logging import loggify + + +@loggify +class TestConfig(TestCase): + def test_custom_parameter(self): + """Tests that a custom parameter type can be defined and used""" + config = InitramfsConfig(logger=self.logger, NO_BASE=True) + config["custom_parameters"]["foo"] = bool + config["foo"] = True + + self.assertEqual(config.get("foo"), True) + + def test_custom_parameter_from_dict(self): + """Tests that a custom parameter can be defined as a dict of strings and used""" + config = InitramfsConfig(logger=self.logger, NO_BASE=True) + config["custom_parameters"] = {"foo": "bool"} + config["foo"] = True + + self.assertEqual(config.get("foo"), True) + + def test_late_arg(self): + """Tests that a late arg is not processed until the late stage""" + config = InitramfsConfig(logger=self.logger, NO_BASE=True) + config["custom_parameters"] = {"foo": "bool"} + config["_late_args"] = "foo" + config["foo"] = True + + # hol up + self.assertEqual(config.get("foo"), False) + + # ready up + config["stage"] = "late" + self.assertEqual(config.get("foo"), True) + + def test_validation(self): + """Tests that unprocessed config raises a ValidationError""" + config = InitramfsConfig(logger=self.logger, NO_BASE=True) + config["foo"] = True + with self.assertRaises(ValidationError): + config["stage"] = "final" + + +if __name__ == "__main__": + main()