Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,12 @@ using [PEP 440](https://packaging.python.org/en/latest/specifications/version-sp

### Added

- Applies the configuration bundled in a pipeline file (`configs` in the `.adg`) as the lowest argument layer, under `--arguments` and command-line flags; `--no-bundled-config` ignores it and `adagio pipeline config` prints it as an arguments file.
- Applies the variable values and input paths saved with a pipeline (`configs` in the `.adg`) as the lowest argument layer, under `--arguments` and command-line flags. `--help` shows a variable's value as the parameter's default.
- Runs each plugin and task in the execution environment saved with the pipeline (`configs[0].execution` in the `.adg`, the shape of a `--config` file), so a downloaded pipeline runs without a separate run-config file. A `--config` file overrides the saved environments. The run header lists each plugin's environment and whether it came from the pipeline file or `--config`.

### Changed

- Reads variable values from the bundled configuration rather than the pipeline signature; signatures written by older releases still supply their defaults.
- Reads variable values from the values saved with the pipeline rather than the pipeline signature; signatures written by older releases still supply their defaults.

### Fixed

Expand Down
32 changes: 32 additions & 0 deletions src/adagio/bundled_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,18 @@
Files written before configs existed kept variable defaults inside the spec
(``signature.parameters[].default`` and ``meta.editor_variables[].value``);
those read as a bundled config too, so an older file keeps working unchanged.

The same config may carry an ``execution`` object: the environments and
resources the app would run each plugin and task with, in the shape of a
``--config`` file. The CLI applies them under any ``--config`` file the same
way, so a downloaded pipeline runs without one.
"""

from typing import Any

from pydantic import ValidationError

from .cli.config import AdagioRunConfig
from .model.arguments import AdagioArguments


Expand Down Expand Up @@ -43,6 +51,30 @@ def bundled_config(data: Any) -> dict[str, Any] | None:
return _legacy_config(data.get("spec", data))


def saved_run_config(data: Any) -> AdagioRunConfig | None:
"""The execution settings saved with a pipeline file, or ``None``.

Raises ``SystemExit`` with a readable message when the saved settings do
not have the shape of a ``--config`` file.
"""
if not isinstance(data, dict) or not isinstance(data.get("configs"), list):
return None
for entry in data["configs"]:
if not isinstance(entry, dict):
continue
execution = entry.get("execution")
if execution is None:
return None
try:
return AdagioRunConfig.model_validate(execution)
except ValidationError as error:
raise SystemExit(
"The execution environments saved with this pipeline are not "
f"valid: {error}"
) from error
return None


def _legacy_config(spec: Any) -> dict[str, Any] | None:
"""Variable values an older spec stored in its signature or editor metadata."""
if not isinstance(spec, dict):
Expand Down
11 changes: 11 additions & 0 deletions src/adagio/cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

from ..executors.base import TaskEnvironmentOverride
from ..executors.defaults import EnvironmentLayer

try:
import tomllib
Expand Down Expand Up @@ -181,3 +182,13 @@ def named_environment_overrides(
if (override := raw_override.to_task_environment_override()) is not None
}
return resolved or None


def environment_layer(run_config: AdagioRunConfig, *, source: str) -> EnvironmentLayer:
"""The environments ``run_config`` sets, as one layer named ``source``."""
return EnvironmentLayer(
source=source,
default_override=default_environment_override(run_config),
plugin_overrides=named_environment_overrides(run_config.plugins) or {},
task_overrides=named_environment_overrides(run_config.tasks) or {},
)
36 changes: 12 additions & 24 deletions src/adagio/cli/dynamic.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,17 @@ def _display_type_label(*, spec_type: str | None, type_hint: Any, is_input: bool
return _pipeline_type_label(type_hint)


_OUTPUT_PATH_RULE = (
"Overrides --output-dir for this output. A path ending in / is a directory "
"the output is written into under its own name."
)


def _output_path_help(description: str | None) -> str:
cleaned = (description or "").strip()
if cleaned:
return f"{cleaned} Overrides --output-dir for this output."
return "Overrides --output-dir for this output."
return f"{cleaned} {_OUTPUT_PATH_RULE}"
return _OUTPUT_PATH_RULE


def _render_pipeline_type(
Expand Down Expand Up @@ -283,7 +289,6 @@ def build_dynamic_run(
"--pipeline",
"-p",
"--arguments",
"--no-bundled-config",
"--config",
"--show-params",
"--cache-dir",
Expand Down Expand Up @@ -322,18 +327,6 @@ def build_dynamic_run(
help="Path to a JSON arguments file. Values are applied before CLI overrides.",
),
]
annotations["no_bundled_config"] = Annotated[
bool,
CliParameter(
name=("--no-bundled-config",),
negative=(),
group=command_group,
help=(
"Ignore the configuration bundled in the pipeline file. By default it "
"supplies values under the arguments file and these options."
),
),
]
annotations["show_params"] = Annotated[
ShowParamsMode,
CliParameter(
Expand All @@ -347,7 +340,10 @@ def build_dynamic_run(
CliParameter(
name=("--config",),
group=command_group,
help="Path to a TOML runtime config file.",
help=(
"Path to a TOML or JSON runtime config file. Its environments "
"override those saved with the pipeline."
),
),
]
annotations["cache_dir"] = Annotated[
Expand Down Expand Up @@ -423,12 +419,6 @@ def build_dynamic_run(
default=None,
annotation=annotations["arguments_file"],
),
inspect.Parameter(
name="no_bundled_config",
kind=inspect.Parameter.KEYWORD_ONLY,
default=False,
annotation=annotations["no_bundled_config"],
),
inspect.Parameter(
name="show_params",
kind=inspect.Parameter.KEYWORD_ONLY,
Expand Down Expand Up @@ -635,7 +625,6 @@ def add_param_spec(spec: ParamSpec) -> None:
def run(
pipeline: Path,
arguments_file: Path | None = None,
no_bundled_config: bool = False,
show_params: ShowParamsMode = ShowParamsMode.REQUIRED,
config_file: Path | None = None,
output_dir: Path | None = None,
Expand All @@ -645,7 +634,6 @@ def run(
**kwargs: Any,
) -> None:
_ = show_params
kwargs["no_bundled_config"] = no_bundled_config
kwargs["output_dir"] = output_dir
kwargs["recycle_pool"] = recycle_pool
kwargs["log_dir"] = log_dir
Expand Down
5 changes: 3 additions & 2 deletions src/adagio/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ def main(argv: list[str] | None = None) -> None:

def _run_pipeline_app(argv: list[str]) -> None:
from ..app.parsers.pipeline import parse_inputs, parse_outputs, parse_parameters
from ..bundled_config import bundled_config
from ..bundled_config import bundled_config, saved_run_config
from ..executors.cache_support import CACHE_DIR_HELP, REUSE_HELP
from .config import load_run_config
from .dynamic import build_dynamic_run
Expand Down Expand Up @@ -214,7 +214,7 @@ def run(
# What --help shows as defaults and what --show-params counts as
# missing follow the same layers as the run: the bundled config
# underneath, the arguments file over it.
bundled = None if "--no-bundled-config" in argv else bundled_config(data)
bundled = bundled_config(data)
if bundled is not None:
arguments_data = {
"inputs": {
Expand All @@ -226,6 +226,7 @@ def run(
**(arguments_data or {}).get("parameters", {}),
},
}
saved_run_config(data)
if config_path_str:
load_run_config(Path(config_path_str))
visible_inputs, visible_params, visible_outputs = _filter_visible_specs(
Expand Down
21 changes: 0 additions & 21 deletions src/adagio/cli/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
from cyclopts import App, Parameter
from rich.console import Console

from ..bundled_config import bundled_config
from ..describe import render_pipeline_text
from ..model.pipeline import AdagioPipeline
from .pipeline_sources import PipelineResolutionError, resolve_pipeline_reference
Expand All @@ -20,29 +19,9 @@ def run_pipeline_cli(argv: list[str]) -> None:
help="Inspect pipeline definitions.",
)
app.command(show_pipeline, name="show")
app.command(show_pipeline_config, name="config")
app(argv)


def show_pipeline_config(pipeline: Path) -> None:
"""Print the configuration bundled in a pipeline file as an arguments file.

Save it, edit it, and pass it back with ``adagio run --arguments`` to
override the bundled values; ``--no-bundled-config`` ignores them instead.
"""
with ExitStack() as exit_stack:
try:
pipeline_path = resolve_pipeline_reference(pipeline, exit_stack=exit_stack)
except PipelineResolutionError as error:
raise SystemExit(str(error)) from error

data = json.loads(pipeline_path.read_text(encoding="utf-8"))
config = bundled_config(data)
if config is None:
raise SystemExit("This pipeline bundles no configuration.")
print(json.dumps({"version": 1, **config}, indent=2))


def show_pipeline(
pipeline: Path,
*,
Expand Down
80 changes: 65 additions & 15 deletions src/adagio/cli/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,14 @@

from rich import box
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from rich.text import Text

from ..bundled_config import apply_bundled_config, bundled_config
from ..bundled_config import apply_bundled_config, bundled_config, saved_run_config
from .config import (
default_environment_override,
environment_layer,
load_run_config,
named_environment_overrides,
)
Expand Down Expand Up @@ -68,7 +70,6 @@ def run_pipeline_from_kwargs(
recycle_pool = kwargs.pop("recycle_pool", None)
log_dir = kwargs.pop("log_dir", None)
targets_raw = kwargs.pop("targets", None)
no_bundled_config = bool(kwargs.pop("no_bundled_config", False))

with ExitStack() as exit_stack:
try:
Expand All @@ -89,6 +90,7 @@ def run_pipeline_from_kwargs(
parsed_pipeline = AdagioPipeline.model_validate(pipeline_data)
arguments = parsed_pipeline.signature.to_default_arguments()
run_config = load_run_config(config_file)
saved_execution = saved_run_config(data)
output_names = [output.name for output in parsed_pipeline.signature.outputs]

input_names = {name for _, name in input_bindings}
Expand All @@ -97,14 +99,13 @@ def run_pipeline_from_kwargs(

# Layers, lowest first: the config bundled in the file, the arguments
# file, then command-line flags.
if not no_bundled_config:
apply_bundled_config(
arguments,
bundled_config(data),
input_names=input_names,
parameter_names=param_names,
output_names=output_name_set,
)
apply_bundled_config(
arguments,
bundled_config(data),
input_names=input_names,
parameter_names=param_names,
output_names=output_name_set,
)

if arguments_file is not None:
file_data = json.loads(arguments_file.read_text(encoding="utf-8"))
Expand Down Expand Up @@ -209,18 +210,32 @@ def run_pipeline_from_kwargs(
console.print(f"[bold]Cache:[/bold] {describe_cache_config(cache_config)}")

from ..executors import select_default_executor
from ..executors.defaults import ConfigurableTaskEnvironmentResolver

executor = select_default_executor(
default_override=default_environment_override(run_config),
plugin_overrides=named_environment_overrides(
# Environments, lowest first: those saved with the pipeline, then the
# --config file.
environments = {
"default_override": default_environment_override(run_config),
"plugin_overrides": named_environment_overrides(
run_config.plugins if run_config is not None else {}
),
task_overrides=named_environment_overrides(
"task_overrides": named_environment_overrides(
run_config.tasks if run_config is not None else {}
),
)
"lower_layers": (
[environment_layer(saved_execution, source="pipeline file")]
if saved_execution is not None
else []
),
}
executor = select_default_executor(**environments)

if not suppress_header:
_print_environments(
console,
parsed_pipeline,
ConfigurableTaskEnvironmentResolver(**environments),
)
console.print(f"[bold]Executing pipeline[/bold] ({executor.mode_label})")

target_ids: set[str] | None = None
Expand All @@ -239,6 +254,41 @@ def run_pipeline_from_kwargs(
)


def _print_environments(console: Console, pipeline: Any, resolver: Any) -> None:
"""List each plugin's environment and where it came from.

A task whose environment differs from the rest of its plugin's gets a row
of its own, so a per-task choice is never hidden behind the plugin's.
"""
from ..model.task import PluginActionTask

rows: dict[str, str] = {}
plugin_rows: dict[str, str] = {}
for task in pipeline.iter_tasks():
if not isinstance(task, PluginActionTask):
continue
try:
environment = resolver.resolve(task=task)
except ValueError:
description = "[red]not configured[/red]"
else:
source = resolver.source_of(task=task) or "default"
description = (
f"{environment.kind} {escape(environment.reference)} "
f"[dim]({source})[/dim]"
)
if task.plugin not in plugin_rows:
plugin_rows[task.plugin] = description
rows[task.plugin] = description
elif plugin_rows[task.plugin] != description:
rows[f"{task.plugin} ({task.name or task.id})"] = description
if not rows:
return
console.print("[bold]Environments:[/bold]")
for label, description in rows.items():
console.print(f" {escape(label)}: {description}", highlight=False)


def _is_missing(value: Any) -> bool:
"""Treat placeholders and null values as missing."""
return value is None or value == "" or value == "<fill me>" or value == [] or value == {}
Expand Down
2 changes: 2 additions & 0 deletions src/adagio/executors/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ def select_default_executor(
default_override=None,
plugin_overrides=None,
task_overrides=None,
lower_layers=(),
launchers=None,
):
from .defaults import ConfigurableTaskEnvironmentResolver
Expand All @@ -29,6 +30,7 @@ def select_default_executor(
default_override=default_override,
plugin_overrides=plugin_overrides,
task_overrides=task_overrides,
lower_layers=lower_layers,
),
launchers=launchers or builtin_task_environment_launchers(),
)
Loading
Loading