coco_pipe.decoding.pipeline#

Study-agnostic per-unit runner and sweep orchestration for decoding.

Mirrors coco_pipe.dim_reduction.pipeline: a study enumerates the independent decoding units of a sweep (one DecodingUnit each). Two layers live here and are deliberately free of any report/visualization imports (reports are built study-side from the saved result.joblib artifacts):

run_decoding_unit() owns the per-unit lifecycle — resume, run, export, record extraction;
execute_decoding_sweep() runs many units (optionally in parallel via coco_pipe.utils.run_task_batch()) and persists one flat result table plus a resumable runs/ inventory (sweep_runs.json + run_summary.json + leaderboard.json) and a run-status marker.

Classes#

`DecodingUnit`	One independent unit of a decoding sweep, ready to run in isolation.
`Sweep`	A saved decoding sweep, as returned by `read_sweep()`.

Functions#

`run_decoding_unit`(unit, *[, errors])	Run one decoding unit and return one record per model.
`allocate_inner_jobs`(cfg, n_jobs)	Return a copy of the experiment config with a new inner `n_jobs`.
`allocate_outer_inner`(total_jobs, n_units)	Split a job budget into outer workers and the jobs each unit may use.
`build_leaderboard`(records, group_fields, *[, ...])	Best successful record per group, ranked by the primary metric.
`write_run_summary`(output_root, records, failures, *[, ...])	Write `runs/run_summary.json` and the top-level status marker.
`load_sweep_records`(output_root)	Load the persisted per-unit records for a `--reports-only` pass.
`read_sweep`(output_root)	Read a saved decoding sweep.
`execute_decoding_sweep`(units, failures, *, config, ...)	Run a decoding sweep and persist its results + resumable `runs/` inventory.
`execute_decoding_sweep_streaming`(unit_batches, ...[, ...])	Run a decoding sweep from a lazy stream of per-scope unit batches.

Module Contents#

class coco_pipe.decoding.pipeline.DecodingUnit#

One independent unit of a decoding sweep, ready to run in isolation.

The unit carries a fully-built ExperimentConfig plus the data and provenance needed to fit, resume, export, and label it. It is a plain data object (picklable) so it can be dispatched to a process pool.

experiment_config: coco_pipe.decoding.configs.ExperimentConfig#

X: numpy.ndarray#

y: numpy.ndarray#

output_dir: pathlib.Path#

context: dict[str, Any]#

run_config: dict[str, Any]#

groups: numpy.ndarray | None = None#

feature_names: collections.abc.Sequence[str] | None = None#

sample_ids: collections.abc.Sequence[Any] | None = None#

sample_metadata: pandas.DataFrame | None = None#

observation_level: str = 'sample'#

inferential_unit: str | None = None#

overwrite: bool = False#

include_p_values: bool = True#

coco_pipe.decoding.pipeline.run_decoding_unit(unit, *, errors='record')#

Run one decoding unit and return one record per model.

Resumes from a matching completed output when overwrite is false, else runs the experiment, exports it (result.joblib + run_manifest.json), and extracts records. A resume config-hash mismatch is intentionally not caught — it raises so the sweep aborts rather than silently mixing configs. A runtime failure becomes a single status="failed" record when errors="record" (the default for a fault-tolerant sweep).

Parameters:

unit (DecodingUnit)
errors (Literal['raise', 'record'])

Return type:

list[dict[str, Any]]

coco_pipe.decoding.pipeline.allocate_inner_jobs(cfg, n_jobs)#

Return a copy of the experiment config with a new inner n_jobs.

Hands the whole job budget to a lone unit’s folds/permutations when only one outer worker is active. The budget is split across the two nesting levels that actually run concurrently — the outer CV-fold loop (joblib.Parallel over cfg.cv.n_splits folds) and, inside each fold, the per-fold hyperparameter search (GridSearchCV/RandomizedSearchCV). Handing n_jobs to both levels independently (as opposed to splitting it) oversubscribes the machine by a factor of n_splits: each of the n_splits fold workers would spawn its own n_jobs-worker search pool, for up to n_splits * n_jobs processes contending over n_jobs cores.

Parameters:

cfg (coco_pipe.decoding.configs.ExperimentConfig)
n_jobs (int)

Return type:

coco_pipe.decoding.configs.ExperimentConfig

coco_pipe.decoding.pipeline.allocate_outer_inner(total_jobs, n_units)#

Split a job budget into outer workers and the jobs each unit may use.

Every unit gets a worker while there are workers left; with fewer units than workers, each unit takes an equal share of the budget.

Parameters:

total_jobs (int)
n_units (int)

Return type:

tuple[int, int]

coco_pipe.decoding.pipeline.build_leaderboard(records, group_fields, *, primary_metric='primary_metric')#

Best successful record per group, ranked by the primary metric.

Parameters:

records (collections.abc.Sequence[dict[str, Any]])
group_fields (collections.abc.Sequence[str])
primary_metric (str)

Return type:

list[dict[str, Any]]

coco_pipe.decoding.pipeline.write_run_summary(output_root, records, failures, *, primary_mask=None, run_metadata=None, report_path=None)#

Write runs/run_summary.json and the top-level status marker.

Status is SUCCESS when a primary result succeeded and nothing failed, PARTIAL when there is any success, else FAILED — matching the dim-reduction run-status contract. Without primary_mask every successful unit counts as primary.

Parameters:

output_root (pathlib.Path)
records (collections.abc.Sequence[dict[str, Any]])
failures (collections.abc.Sequence[dict[str, Any]])
primary_mask (PrimaryMask | None)
run_metadata (dict[str, Any] | None)
report_path (pathlib.Path | None)

Return type:

dict[str, Any]

coco_pipe.decoding.pipeline.load_sweep_records(output_root)#

Load the persisted per-unit records for a --reports-only pass.

Parameters:: output_root (pathlib.Path)
Return type:: list[dict[str, Any]]

class coco_pipe.decoding.pipeline.Sweep#

A saved decoding sweep, as returned by read_sweep().

Variables:

root (Path) – The sweep’s output directory.
results (pandas.DataFrame) – One row per unit and model: the run context, the summary metrics and output_dir, the folder of the unit.
status (str) – "SUCCESS", "PARTIAL" (some units failed, or the run did not finish) or "FAILED".

root: pathlib.Path#

results: pandas.DataFrame#

status: str#

result(output_dir)#

Load the full result of one unit.

Parameters:: output_dir (str | pathlib.Path)
Return type:: coco_pipe.decoding.result.ExperimentResult

predictions(rows=None)#

Return the predictions of the units in rows, in one table.

Parameters:: rows (pandas.DataFrame, optional) – A selection of results. All successful units by default.
Returns:: One row per unit, model and sample, with output_dir first.
Return type:: pandas.DataFrame

coco_pipe.decoding.pipeline.read_sweep(output_root)#

Read a saved decoding sweep.

Only the results table is read; a unit’s full result is loaded when asked for with Sweep.result() or Sweep.predictions(). A sweep that is still running, or that broke, can be read too: its status is "PARTIAL".

Parameters:: output_root (str | pathlib.Path)
Return type:: Sweep

coco_pipe.decoding.pipeline.execute_decoding_sweep(units, failures, *, config, output_root, results_filename=RESULTS_FILENAME, primary_mask=None, leaderboard_group_fields=(), reallocate_inner_jobs=False, frame_post=None, extra_outputs=None, run_metadata=None)#

Run a decoding sweep and persist its results + resumable runs/ inventory.

Owns the study-agnostic tail shared by classical and foundation decoding: job allocation, per-unit execution, the flat result table, the runs/ inventory, and the run-status marker. Study-specific pre/post steps (p-value correction, capability matrices) inject via frame_post / extra_outputs.

Parameters:

units (list[DecodingUnit])
failures (list[dict[str, Any]])
config (dict[str, Any])
output_root (pathlib.Path)
results_filename (str)
primary_mask (PrimaryMask | None)
leaderboard_group_fields (collections.abc.Sequence[str])
reallocate_inner_jobs (bool)
frame_post (collections.abc.Callable[[pandas.DataFrame], pandas.DataFrame] | None)
extra_outputs (collections.abc.Callable[[pandas.DataFrame, list[dict[str, Any]], pathlib.Path], None] | None)
run_metadata (dict[str, Any] | None)

Return type:

tuple[list[dict[str, Any]], pandas.DataFrame]

coco_pipe.decoding.pipeline.execute_decoding_sweep_streaming(unit_batches, failures, *, config, output_root, results_filename=RESULTS_FILENAME, primary_mask=None, leaderboard_group_fields=(), reallocate_inner_jobs=False, frame_post=None, extra_outputs=None, run_metadata=None)#

Run a decoding sweep from a lazy stream of per-scope unit batches.

Behaves like execute_decoding_sweep() but consumes unit_batches one batch at a time — running and then releasing each batch (and, via the generator, its source data container) before the next is materialized — so only one scope’s worth of arrays is resident at once. This keeps host memory bounded to the largest single scope rather than the whole sweep. failures is the caller-owned list of enumeration-time skips, extended in place with runtime failures (so any extra_outputs closure over it sees them all), and every batch shares the same job budget/parallelism policy as the one-shot variant.

After every batch the records so far are saved to the runs/ inventory and the run is marked _PARTIAL, so a job that breaks leaves a readable record of what finished. The full outputs are written once the stream is exhausted.

Parameters:

unit_batches (collections.abc.Iterable[collections.abc.Sequence[DecodingUnit]])
failures (list[dict[str, Any]])
config (dict[str, Any])
output_root (pathlib.Path)
results_filename (str)
primary_mask (PrimaryMask | None)
leaderboard_group_fields (collections.abc.Sequence[str])
reallocate_inner_jobs (bool)
frame_post (collections.abc.Callable[[pandas.DataFrame], pandas.DataFrame] | None)
extra_outputs (collections.abc.Callable[[pandas.DataFrame, list[dict[str, Any]], pathlib.Path], None] | None)
run_metadata (dict[str, Any] | None)

Return type:

tuple[list[dict[str, Any]], pandas.DataFrame]