nirs4all.pipeline.dagml.cli_runner module

Assemble dag-ml-cli inputs for a CV+refit run and drive the binary (Phase 2b-ii.3).

This is the integration seam that makes dag-ml execute a nirs4all pipeline end-to-end. Two things the plain compat DSL does NOT provide had to be added as top-level DSL fields (the compat lowerer preserves them verbatim):

• data_bindings — binds the model node’s input x to the data source, carrying the envelope’s fingerprint triple (the runtime keys its data provider by that triple, not by source id). The compat sources step is a no-op, and there is no auto-edge from the graph interface input to the first node, so without a binding the node gets empty data_views. The binding targets the model node specifically because our node runner re-fetches features by the task’s own sample_ids (it does not consume upstream edge data).

• split_invocation.fold_set — a materialized FoldSet. KFold params are inert: dag-ml has no runtime/host splitter that materializes the outer fold set, so the host must compute the folds (build_fold_set) and embed them.

With both, the runtime emits per-fold data_views{sample_ids} and the adapter resolves real X/y. Verified end-to-end against dag-ml-cli run-process-dsl-cv-refit-bundle: the FIT_CV OOF matches direct sklearn KFold. The envelope must be scoped to the CV universe (build_envelope( …, sample_ints=train_pool)) so every relation lives inside the fold set.

nirs4all.pipeline.dagml.cli_runner.assemble_constrained_cv_refit_dsl(pipeline: list[Any], identity: IdentityMap, envelope: dict[str, Any], folds: list[tuple[list[int], list[int]]], *, dsl_id: str = 'nirs4all-pipeline', n_splits: int) → dict[str, Any]

The executable DSL for a CONSTRAINED operator generator: survivor-branch Generator + fold_set (ADR-17 1a + 1b).

Lowers the constrained _or_-pick / _cartesian_ pipeline into ONE canonical Generator step whose branches are the pruned survivor sequences (each [<transforms>, <model>]; lower_constrained_operator_pipeline()), then embeds the fold_set. The data_bindings are deliberately EMPTY here: the survivor-branch graph compiles ONE model node per survivor (Mechanism B namespacing), and the caller (_run_native_operator_generation()) re-binds EVERY model node after compile — exactly as the flat operator-SELECT path re-binds its per-choice model nodes — so an a-priori single binding (which would also force a wrong-path pipeline_to_dsl compile in model_node_id) is neither needed nor correct.

nirs4all.pipeline.dagml.cli_runner.assemble_cv_refit_dsl(pipeline: list[Any], identity: IdentityMap, envelope: dict[str, Any], folds: list[tuple[list[int], list[int]]], *, dsl_id: str = 'nirs4all-pipeline', n_splits: int, source_id: str = 'src0', force_fitted_x_chain: bool = False) → dict[str, Any]

The executable compat DSL: lowered pipeline + embedded fold_set + model data binding.

nirs4all.pipeline.dagml.cli_runner.data_bindings_for(model_id: str, envelope: dict[str, Any], *, source_id: str = 'src0', model_input: dict[str, Any] | None = None) → list[dict[str, Any]]

Bind an aggregate X input or each explicitly declared dense source port.

Named bindings preserve the global envelope fingerprints but attest one projected table each. The cohort’s native source identities must retain their captured user-name mapping, including during PREDICT replay.

nirs4all.pipeline.dagml.cli_runner.data_bindings_for_fitted_x_chain(graph: dict[str, Any], model_id: str, envelope: dict[str, Any], *, source_id: str = 'src0', force: bool = False) → list[dict[str, Any]]

Bind each X node when any step requests a native, node-local fit scope.

nirs4all.pipeline.dagml.cli_runner.data_bindings_for_nodes(model_ids: list[str], envelope: dict[str, Any], *, source_id: str = 'src0') → list[dict[str, Any]]

One DataBinding per model node — the N per-partition model nodes of a separation branch.

nirs4all.pipeline.dagml.cli_runner.model_node_id(pipeline: list[Any], *, dsl_id: str = 'nirs4all-pipeline') → str

The compiler’s id for the model node (compile-first; do not re-derive the ordinal).

nirs4all.pipeline.dagml.cli_runner.needs_dynamic_feature_axis(pipeline: list[Any]) → bool

A feature selector changes the source axis only after its fit completes.

nirs4all.pipeline.dagml.cli_runner.run_cv_refit_bundle(*, dsl: dict[str, Any], envelope: dict[str, Any], graph: dict[str, Any], dataset_path: str, workdir: Path, dagml_cli: str, venv_python: str, selection_metric: str = 'rmse', sample_metadata: dict[str, dict[str, Any]] | None = None, dataset_pickle: str | None = None, random_state: int | None = None, refit: bool = True, refit_top_k: int = 1) → dict[str, Any]

Write inputs + shim, run dag-ml-cli run-process-dsl-cv-refit-bundle, return outputs.

Returns {returncode, stdout, results} where results are the adapter’s captured NodeResult frames (carrying the FIT_CV validation/OOF predictions).

selection_metric (rmse | accuracy) is forwarded to --selection-metric: when the DSL compiles to multiple plan.variants (native param-level generation) and no variant is pinned, dag-ml runs single-variant FIT_CV per variant, scores each variant’s cross-fold OOF average, and refits the best — so bundle.scores reflects the natively-selected winner.

sample_metadata ({wire_id: {col: value}}) is written for the adapter to honor separation-branch branch_view selectors (each fanned model fits/predicts only its partition). Omit it for non-branch pipelines.

dataset_pickle (a path to a pickled augmented SpectroDataset) is passed to the adapter via N4A_DAGML_DATASET_PICKLE so it loads the exact same synthetic rows the DSL/envelope were built from (a sample_augmentation run; augmentation is not reproducible across processes). Omit it for non-augmentation pipelines.

random_state (the run’s seed) is forwarded to the adapter via N4A_RANDOM_STATE in the PER-CALL child env dict ONLY (never os.environ), so the fresh-python worker seeds its global RNG (init_global_random_state) before fitting — making an unseeded stochastic operator reproducible. Set only for this launch and explicitly dropped otherwise, so no stale seed leaks into a later run or a concurrent dag-ml run.

nirs4all.pipeline.dagml.cli_runner.run_refit_phase_cli(*, dsl: dict[str, Any], envelope: dict[str, Any], graph: dict[str, Any], training_sample_ids: list[str], dataset_path: str, workdir: Path, dagml_cli: str, venv_python: str, dataset_pickle: str | None = None, sample_metadata: dict[str, dict[str, Any]] | None = None, random_state: int | None = None, package_id: str | None = None) → dict[str, Any]

Run one no-splitter REFIT in the native CLI with attested row order.

nirs4all.pipeline.dagml.cli_runner.split_invocation_for(identity: IdentityMap, folds: list[tuple[list[int], list[int]]], *, n_splits: int, shuffle: bool = True) → dict[str, Any]

A split_invocation with an embedded, materialized FoldSet (params alone are inert).

nirs4all.pipeline.dagml.cli_runner.write_launcher_shim(path: Path, venv_python: str) → Path

Write a non-.py executable shim that re-execs the adapter under the project venv.

dag-ml-cli runs a .py adapter under the system python3 (which lacks nirs4all), so the adapter must be a non-.py executable. Result frames are captured by the adapter itself (N4A_DAGML_RESULT_CAPTURE), robust vs tee pipe-buffer line splitting.

The --describe handshake is routed to the adapter run as a standalone file rather than -m nirs4all.…: python -m imports the nirs4all parent package first (its __init__ eagerly pulls sklearn + pandas, ~1.4s) before the module’s --describe short-circuit can fire, so it would re-import the whole library just to print a STATIC capability dict and discard the process. Run by path, the file is __main__ with no package import, so --describe answers in a few ms. The persistent worker keeps the -m form because it needs the package’s relative imports + the real nirs4all stack to materialize data and fit operators.