coco_pipe.decoding.pipeline =========================== .. py:module:: coco_pipe.decoding.pipeline .. autoapi-nested-parse:: Study-agnostic per-unit runner and sweep orchestration for decoding. Mirrors :mod:`coco_pipe.dim_reduction.pipeline`: a study enumerates the independent decoding units of a sweep (one :class:`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): * :func:`run_decoding_unit` owns the per-unit lifecycle — resume, run, export, record extraction; * :func:`execute_decoding_sweep` runs many units (optionally in parallel via :func:`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 ------- .. autoapisummary:: coco_pipe.decoding.pipeline.DecodingUnit coco_pipe.decoding.pipeline.Sweep Functions --------- .. autoapisummary:: coco_pipe.decoding.pipeline.run_decoding_unit coco_pipe.decoding.pipeline.allocate_inner_jobs coco_pipe.decoding.pipeline.allocate_outer_inner coco_pipe.decoding.pipeline.build_leaderboard coco_pipe.decoding.pipeline.write_run_summary coco_pipe.decoding.pipeline.load_sweep_records coco_pipe.decoding.pipeline.read_sweep coco_pipe.decoding.pipeline.execute_decoding_sweep coco_pipe.decoding.pipeline.execute_decoding_sweep_streaming Module Contents --------------- .. py:class:: DecodingUnit One independent unit of a decoding sweep, ready to run in isolation. The unit carries a fully-built :class:`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. .. py:attribute:: experiment_config :type: coco_pipe.decoding.configs.ExperimentConfig .. py:attribute:: X :type: numpy.ndarray .. py:attribute:: y :type: numpy.ndarray .. py:attribute:: output_dir :type: pathlib.Path .. py:attribute:: context :type: dict[str, Any] .. py:attribute:: run_config :type: dict[str, Any] .. py:attribute:: groups :type: numpy.ndarray | None :value: None .. py:attribute:: feature_names :type: collections.abc.Sequence[str] | None :value: None .. py:attribute:: sample_ids :type: collections.abc.Sequence[Any] | None :value: None .. py:attribute:: sample_metadata :type: pandas.DataFrame | None :value: None .. py:attribute:: observation_level :type: str :value: 'sample' .. py:attribute:: inferential_unit :type: str | None :value: None .. py:attribute:: overwrite :type: bool :value: False .. py:attribute:: include_p_values :type: bool :value: True .. py:function:: 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). .. py:function:: 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 (:class:`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. .. py:function:: 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. .. py:function:: build_leaderboard(records, group_fields, *, primary_metric = 'primary_metric') Best successful record per group, ranked by the primary metric. .. py:function:: 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. .. py:function:: load_sweep_records(output_root) Load the persisted per-unit records for a ``--reports-only`` pass. .. py:class:: Sweep A saved decoding sweep, as returned by :func:`read_sweep`. :ivar root: The sweep's output directory. :vartype root: Path :ivar results: One row per unit and model: the run context, the summary metrics and ``output_dir``, the folder of the unit. :vartype results: pandas.DataFrame :ivar status: ``"SUCCESS"``, ``"PARTIAL"`` (some units failed, or the run did not finish) or ``"FAILED"``. :vartype status: str .. py:attribute:: root :type: pathlib.Path .. py:attribute:: results :type: pandas.DataFrame .. py:attribute:: status :type: str .. py:method:: result(output_dir) Load the full result of one unit. .. py:method:: predictions(rows = None) Return the predictions of the units in ``rows``, in one table. :param rows: A selection of :attr:`results`. All successful units by default. :type rows: pandas.DataFrame, optional :returns: One row per unit, model and sample, with ``output_dir`` first. :rtype: pandas.DataFrame .. py:function:: 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 :meth:`Sweep.result` or :meth:`Sweep.predictions`. A sweep that is still running, or that broke, can be read too: its status is ``"PARTIAL"``. .. py:function:: 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*. .. py:function:: 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 :func:`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.