From 61b320238693b29f0fa52a5ea38fbd6709b47aaa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tobias=20H=C3=B6lzer?= Date: Tue, 29 Oct 2024 15:40:06 +0100 Subject: [PATCH] Add --config-file argument to CLI --- darts/src/darts/cli.py | 10 +- darts/src/darts/utils/config.py | 176 +++++++++++++++++++++++--------- 2 files changed, 133 insertions(+), 53 deletions(-) diff --git a/darts/src/darts/cli.py b/darts/src/darts/cli.py index 5b9c161..1f918bf 100644 --- a/darts/src/darts/cli.py +++ b/darts/src/darts/cli.py @@ -10,17 +10,17 @@ from darts import __version__ from darts.native import run_native_orthotile_pipeline -from darts.utils.config import config_parser +from darts.utils.config import ConfigParser from darts.utils.logging import add_logging_handlers, setup_logging logger = logging.getLogger(__name__) console = Console() - +config_parser = ConfigParser() app = cyclopts.App( version=__version__, console=console, - config=config_parser, # config=cyclopts.config.Toml("config.toml", root_keys=["darts"], search_parents=True) + config=config_parser, ) pipeline_group = cyclopts.Group.create_ordered("Pipeline Commands") @@ -53,7 +53,9 @@ def hello(name: str, n: int = 1): # Intercept the logging behavior to add a file handler @app.meta.default def launcher( # noqa: D103 - *tokens: Annotated[str, cyclopts.Parameter(show=False, allow_leading_hyphen=True)], log_dir: Path = Path("logs") + *tokens: Annotated[str, cyclopts.Parameter(show=False, allow_leading_hyphen=True)], + log_dir: Path = Path("logs"), + config_file: Path = Path("config.toml"), ): command, bound = app.parse_args(tokens) add_logging_handlers(command.__name__, console, log_dir) diff --git a/darts/src/darts/utils/config.py b/darts/src/darts/utils/config.py index 27e2eb9..e5af287 100644 --- a/darts/src/darts/utils/config.py +++ b/darts/src/darts/utils/config.py @@ -3,6 +3,7 @@ import logging import tomllib from contextlib import suppress +from pathlib import Path import cyclopts @@ -47,62 +48,139 @@ def flatten_dict(d: dict, parent_key: str = "", sep: str = ".") -> dict[str, dic return dict(items) -def config_parser( - apps: list[cyclopts.App], commands: tuple[str, ...], mapping: dict[str, cyclopts.config.Unset | list[str]] -): - """Parser for cyclopts config. An own implementation is needed to select our own toml structure. +class ConfigParser: + """Parser for cyclopts config. - First, the configuration file at "config.toml" is loaded. - Then, this config is flattened and then mapped to the input arguments of the called function. - Hence parent keys are not considered. + An own implementation is needed to select our own toml structure and source. + Implemented as a class to be able to provide the config-file as a parameter of the CLI. + """ - Args: - apps (list[cyclopts.App]): The cyclopts apps. - commands (tuple[str, ...]): The commands. - mapping (dict[str, cyclopts.config.Unset | list[str]]): The mapping of the arguments. + def __init__(self) -> None: + """Initialize the ConfigParser (no-op).""" + self._config = None - Examples: - Config file `./config.toml`: + def open_config(self, file_path: str | Path) -> None: + """Open the config file, takes the 'darts' key, flattens the resulting dict and saves as config. - ```toml - [darts.hello] # The parent key is completely ignored - name = "Tobias" - ``` + Args: + file_path (str | Path): The path to the config file. - Function signature which is called: + Raises: + FileNotFoundError: If the file does not exist. - ```python - # ... setup code for cyclopts - @app.command() - def hello(name: str): - print(f"Hello {name}") - ``` + """ + if isinstance(file_path, str): + file_path = Path(file_path) - Calling the function from CLI: + if not file_path.exists(): + raise FileNotFoundError(f"Config file '{file_path}' not found.") - ```sh - $ darts hello - Hello Tobias + with file_path.open("rb") as f: + config = tomllib.load(f)["darts"] - $ darts hello --name=Max - Hello Max - ``` + # Flatten the config data () + self._config = flatten_dict(config) - """ - with open("config.toml", "rb") as f: - config_data: dict = tomllib.load(f)["darts"] - - # Flatten the config data () - flat_config = flatten_dict(config_data) - - for key, value in mapping.items(): - if not isinstance(value, cyclopts.config.Unset) or value.related_set(mapping): - continue - - with suppress(KeyError): - new_value = flat_config[key]["value"] - parent_key = flat_config[key]["key"] - if not isinstance(new_value, list): - new_value = [new_value] - mapping[key] = [str(x) for x in new_value] - logger.debug(f"Set cyclopts parameter '{key}' to {new_value} from 'config:{parent_key}' ") + def apply_config(self, mapping: dict[str, cyclopts.config.Unset | list[str]]): + """Apply the loaded config to the cyclopts mapping. + + Args: + mapping (dict[str, cyclopts.config.Unset | list[str]]): The mapping of the arguments. + + """ + for key, value in mapping.items(): + if not isinstance(value, cyclopts.config.Unset) or value.related_set(mapping): + continue + + with suppress(KeyError): + new_value = self._config[key]["value"] + parent_key = self._config[key]["key"] + if not isinstance(new_value, list): + new_value = [new_value] + mapping[key] = [str(x) for x in new_value] + logger.debug(f"Set cyclopts parameter '{key}' to {new_value} from 'config:{parent_key}' ") + + def __call__( + self, apps: list[cyclopts.App], commands: tuple[str, ...], mapping: dict[str, cyclopts.config.Unset | list[str]] + ): + """Parser for cyclopts config. An own implementation is needed to select our own toml structure. + + First, the configuration file at "config.toml" is loaded. + Then, this config is flattened and then mapped to the input arguments of the called function. + Hence parent keys are not considered. + + Args: + apps (list[cyclopts.App]): The cyclopts apps. Unused, but must be provided for the cyclopts hook. + commands (tuple[str, ...]): The commands. Unused, but must be provided for the cyclopts hook. + mapping (dict[str, cyclopts.config.Unset | list[str]]): The mapping of the arguments. + + Examples: + ### Setup the cyclopts App + + ```python + import cyclopts + from darts.utils.config import ConfigParser + + config_parser = ConfigParser() + app = cyclopts.App(config=config_parser) + + # Intercept the logging behavior to add a file handler + @app.meta.default + def launcher( + *tokens: Annotated[str, cyclopts.Parameter(show=False, allow_leading_hyphen=True)], + log_dir: Path = Path("logs"), + config_file: Path = Path("config.toml"), + ): + command, bound = app.parse_args(tokens) + add_logging_handlers(command.__name__, console, log_dir) + return command(*bound.args, **bound.kwargs) + + if __name__ == "__main__": + app.meta() + ``` + + + ### Usage + + Config file `./config.toml`: + + ```toml + [darts.hello] # The parent key is completely ignored + name = "Tobias" + ``` + + Function signature which is called: + + ```python + # ... setup code for cyclopts + @app.command() + def hello(name: str): + print(f"Hello {name}") + ``` + + Calling the function from CLI: + + ```sh + $ darts hello + Hello Tobias + + $ darts hello --name=Max + Hello Max + ``` + + Raises: + ValueError: If no config file is specified. Should not occur if the cyclopts App is setup correctly. + + """ + if self._config is None: + config_param = mapping.get("config-file", None) + if not config_param: + raise ValueError("No config file (--config-file) specified.") + if isinstance(config_param, list): + config_file = config_param[0] + elif isinstance(config_param, cyclopts.config.Unset): + config_file = config_param.iparam.default + # else never happens + self.open_config(config_file) + + self.apply_config(mapping)