Add --config-file argument to CLI

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)

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)