Controller

Orchestration: reads the YAML recipe, loads HDF5 under multiprocessing, and dispatches pipeline steps per run.

XSpect.controller.pipeline

Pipeline: the user-facing entry point for YAML-driven analysis.

Usage: pipeline = Pipeline.from_yaml("my_analysis.yaml") pipeline.run(cores=16, batch_size=2000) results = pipeline.results

Pipeline

YAML-driven analysis pipeline.

Parses a YAML config, creates experiment/run objects, dispatches registered steps, and collects results.

Source code in XSpect/controller/pipeline.py

class Pipeline:
    """
    YAML-driven analysis pipeline.

    Parses a YAML config, creates experiment/run objects, dispatches
    registered steps, and collects results.
    """

    def __init__(self, config: PipelineConfig):
        self.config = config
        self.results = {}
        self.analyzed_runs = []
        self._status_log = []

    @classmethod
    def from_yaml(cls, path: str) -> "Pipeline":
        """
        Create a Pipeline from a YAML configuration file.

        Parameters
        ----------
        path : str
            Path to the YAML config file.

        Returns
        -------
        Pipeline
            Configured pipeline ready to run.
        """
        config = parse_yaml(path)
        return cls(config)

    def run(self, cores: int = 1, batch_size: int = 2000) -> None:
        """
        Execute the pipeline.

        1. Creates an experiment from config
        2. For each run number, creates a spectroscopy_run and executes pipeline steps
        3. After all runs, executes reduction steps
        4. Populates self.results

        Parameters
        ----------
        cores : int
            Number of parallel workers for batch processing.
        batch_size : int
            Number of shots per batch.
        """
        self._status_log.append("Pipeline execution started")
        logger.info(
            "Pipeline execution started (cores=%d, batch_size=%d)", cores, batch_size
        )

        exp = self._create_experiment()

        detector_configs = [
            (dc.hdf5_path, dc.name, dc.transpose, dc.row_range)
            for dc in self.config.data.detector_keys
        ]
        scalar_keys = [(dk.hdf5_path, dk.friendly_name) for dk in self.config.data.keys]
        logger.info("Runs to process: %s", list(self.config.data.runs))

        for run_number in self.config.data.runs:
            self._status_log.append(f"Processing run {run_number}")
            logger.info("── Run %s: creating run object", run_number)
            run = self._create_run(exp, run_number)
            self._load_data(run, skip_detector=True)
            total = getattr(run, "total_shots", None)
            logger.info("Run %s: %s total shots", run_number, total)

            if hasattr(run, "total_shots") and run.total_shots > batch_size:
                n_batches = -(-run.total_shots // batch_size)  # ceil
                logger.info(
                    "Run %s: BATCHED path — %d shots / %d per batch = %d batches on %d cores",
                    run_number,
                    run.total_shots,
                    batch_size,
                    n_batches,
                    cores,
                )
                # Pre-run the pipeline on the parent run (no detector loaded).
                # Detector-dependent steps return gracefully; scalar-only steps
                # (e.g. make_ccm_axis, ccm_binning, time_binning) fire here and
                # produce a globally consistent axis.  The resulting attributes
                # are then injected into every batch so each batch shares the
                # same ccm_bins / ccm_energies / time_bins instead of deriving
                # its own from an incomplete slice of the data.
                run_pipeline(run, self.config.pipeline)
                precomputed_attrs = self._collect_scalar_precomputed(run)
                logger.info(
                    "Run %s: pre-pass complete, dispatching batches…", run_number
                )

                run_batched(
                    run,
                    self.config.pipeline,
                    cores=cores,
                    batch_size=batch_size,
                    detector_configs=detector_configs,
                    scalar_keys=scalar_keys,
                    precomputed_attrs=precomputed_attrs,
                )
            else:
                logger.info(
                    "Run %s: SINGLE path — loading detector into memory", run_number
                )
                self._load_detector(run)
                run_pipeline(run, self.config.pipeline)

            self.analyzed_runs.append(run)
            self._status_log.append(f"Completed run {run_number}")
            logger.info("Run %s: complete", run_number)

        if self.config.reduction:
            self._status_log.append("Running reduction steps")
            logger.info("Running %d reduction step(s)", len(self.config.reduction))
            reduction_results = run_reductions(
                self.analyzed_runs, self.config.reduction
            )
            self.results.update(reduction_results)

        for run in self.analyzed_runs:
            for key, value in run.results.items():
                self.results[f"run_{run.run_number}.{key}"] = value
            self._collect_run_attributes(run)

        self._status_log.append("Pipeline execution complete")

    def _collect_scalar_precomputed(self, run) -> dict:
        """Collect scalar-derived attributes set during the pre-pipeline run.

        Returns attributes that are either non-array scalars or small arrays
        whose first dimension does NOT equal total_shots (i.e. axis/bin arrays
        like ccm_bins, ccm_energies, time_bins), plus per-shot arrays that
        should be sliced per-batch (ccm_bin_indices, timing_bin_indices).

        These are later injected into each batch so that axis steps (make_ccm_axis,
        time_binning) see a pre-populated result and skip re-derivation.

        Raw input data keys (scalar_keys and detector_keys) are excluded: they
        are reloaded fresh from HDF5 in each batch worker and must not be
        overwritten by a pre-pipeline-mutated (e.g. union_shots-filtered) copy.
        """
        import numpy as np

        # Names that will be reloaded per-batch from HDF5 — never inject these.
        input_names = {dk.friendly_name for dk in self.config.data.keys}
        input_names |= {dc.name for dc in self.config.data.detector_keys}

        skip = {
            "spec_experiment",
            "run_number",
            "run_file",
            "status",
            "status_datetime",
            "verbose",
            "end_index",
            "start_index",
            "results",
            "total_shots",
            "run_shots",
            "xray",
            "laser",
            "simultaneous",
            "h5",
        } | input_names

        total = getattr(run, "total_shots", None)
        attrs = {}
        for attr, value in vars(run).items():
            if attr in skip or attr.startswith("_"):
                continue
            if isinstance(value, np.ndarray):
                attrs[attr] = value
            elif not isinstance(value, np.ndarray) and value is not None:
                # Skip non-array objects (h5 handles, etc.) but keep scalars
                if isinstance(value, (int, float, bool, str)):
                    attrs[attr] = value
        return attrs

    def _collect_run_attributes(self, run):
        """Collect pipeline-generated attributes from a run into self.results."""
        skip = {
            "spec_experiment",
            "run_number",
            "run_file",
            "status",
            "status_datetime",
            "verbose",
            "end_index",
            "start_index",
            "results",
            "total_shots",
            "run_shots",
            "xray",
            "laser",
            "simultaneous",
            "h5",
        }
        import numpy as np

        for attr, value in vars(run).items():
            if attr in skip or attr.startswith("_"):
                continue
            if isinstance(value, np.ndarray):
                self.results[attr] = value

    def _create_experiment(self):
        """Create experiment object from config. Wraps the directory lookup failure gracefully."""
        cfg = self.config.experiment
        try:
            exp = spectroscopy_experiment(
                cfg.lcls_run,
                cfg.hutch,
                cfg.experiment_id,
                smalldata_dir=cfg.smalldata_dir,
            )
        except Exception:
            exp = _MockExperiment(cfg.lcls_run, cfg.hutch, cfg.experiment_id)
        return exp

    def _create_run(self, exp, run_number: int):
        """Create a spectroscopy_run for the given run number."""
        max_shots = self.config.data.max_shots
        end_index = max_shots if max_shots is not None else -1
        try:
            run = spectroscopy_run(exp, run_number, end_index=end_index)
        except Exception:
            run = spectroscopy_run.__new__(spectroscopy_run)
            run.spec_experiment = exp
            run.run_number = run_number
            run.run_file = None
            run.status = []
            run.status_datetime = []
            run.verbose = False
            run.end_index = end_index
            run.start_index = 0
            run.results = {}
        return run

    def _load_data(self, run, skip_detector=False):
        """Load data keys into the run if the HDF5 file is accessible.

        Parameters
        ----------
        skip_detector : bool
            If True, skip loading detector arrays (used when batching will
            reload per-batch from HDF5 to avoid loading multi-GB arrays).
        """
        if run.run_file is None or not _file_exists(run.run_file):
            return

        keys = [dk.hdf5_path for dk in self.config.data.keys]
        names = [dk.friendly_name for dk in self.config.data.keys]
        if keys:
            run.load_run_keys(keys, names)

        run.get_run_shot_properties()

        if not skip_detector:
            for det_config in self.config.data.detector_keys:
                kwargs = {}
                if det_config.rois is not None:
                    kwargs["rois"] = det_config.rois
                    kwargs["combine"] = det_config.combine_rois
                if det_config.row_range is not None:
                    kwargs["row_range"] = det_config.row_range
                run.load_run_key_delayed(
                    [det_config.hdf5_path],
                    [det_config.name],
                    **kwargs,
                )

            # Close the h5py file handle so the run object is picklable for multiprocessing
            if hasattr(run, "h5"):
                run.h5.close()
                del run.h5

    def _load_detector(self, run):
        """Load detector data into run (for non-batched path)."""
        if run.run_file is None or not _file_exists(run.run_file):
            return
        for det_config in self.config.data.detector_keys:
            kwargs = {}
            if det_config.rois is not None:
                kwargs["rois"] = det_config.rois
                kwargs["combine"] = det_config.combine_rois
            kwargs["transpose"] = det_config.transpose
            if det_config.row_range is not None:
                kwargs["row_range"] = det_config.row_range
            run.load_run_key_delayed(
                [det_config.hdf5_path],
                [det_config.name],
                **kwargs,
            )
        if hasattr(run, "h5"):
            run.h5.close()
            del run.h5

from_yaml(path) classmethod

Create a Pipeline from a YAML configuration file.

Parameters:

Name	Type	Description	Default
path	str	Path to the YAML config file.	required

Returns:

Type	Description
Pipeline	Configured pipeline ready to run.

Source code in XSpect/controller/pipeline.py

@classmethod
def from_yaml(cls, path: str) -> "Pipeline":
    """
    Create a Pipeline from a YAML configuration file.

    Parameters
    ----------
    path : str
        Path to the YAML config file.

    Returns
    -------
    Pipeline
        Configured pipeline ready to run.
    """
    config = parse_yaml(path)
    return cls(config)

run(cores=1, batch_size=2000)

Execute the pipeline.

1. Creates an experiment from config
2. For each run number, creates a spectroscopy_run and executes pipeline steps
3. After all runs, executes reduction steps
4. Populates self.results

Parameters:

Name	Type	Description	Default
cores	int	Number of parallel workers for batch processing.	1
batch_size	int	Number of shots per batch.	2000

Source code in XSpect/controller/pipeline.py

def run(self, cores: int = 1, batch_size: int = 2000) -> None:
    """
    Execute the pipeline.

    1. Creates an experiment from config
    2. For each run number, creates a spectroscopy_run and executes pipeline steps
    3. After all runs, executes reduction steps
    4. Populates self.results

    Parameters
    ----------
    cores : int
        Number of parallel workers for batch processing.
    batch_size : int
        Number of shots per batch.
    """
    self._status_log.append("Pipeline execution started")
    logger.info(
        "Pipeline execution started (cores=%d, batch_size=%d)", cores, batch_size
    )

    exp = self._create_experiment()

    detector_configs = [
        (dc.hdf5_path, dc.name, dc.transpose, dc.row_range)
        for dc in self.config.data.detector_keys
    ]
    scalar_keys = [(dk.hdf5_path, dk.friendly_name) for dk in self.config.data.keys]
    logger.info("Runs to process: %s", list(self.config.data.runs))

    for run_number in self.config.data.runs:
        self._status_log.append(f"Processing run {run_number}")
        logger.info("── Run %s: creating run object", run_number)
        run = self._create_run(exp, run_number)
        self._load_data(run, skip_detector=True)
        total = getattr(run, "total_shots", None)
        logger.info("Run %s: %s total shots", run_number, total)

        if hasattr(run, "total_shots") and run.total_shots > batch_size:
            n_batches = -(-run.total_shots // batch_size)  # ceil
            logger.info(
                "Run %s: BATCHED path — %d shots / %d per batch = %d batches on %d cores",
                run_number,
                run.total_shots,
                batch_size,
                n_batches,
                cores,
            )
            # Pre-run the pipeline on the parent run (no detector loaded).
            # Detector-dependent steps return gracefully; scalar-only steps
            # (e.g. make_ccm_axis, ccm_binning, time_binning) fire here and
            # produce a globally consistent axis.  The resulting attributes
            # are then injected into every batch so each batch shares the
            # same ccm_bins / ccm_energies / time_bins instead of deriving
            # its own from an incomplete slice of the data.
            run_pipeline(run, self.config.pipeline)
            precomputed_attrs = self._collect_scalar_precomputed(run)
            logger.info(
                "Run %s: pre-pass complete, dispatching batches…", run_number
            )

            run_batched(
                run,
                self.config.pipeline,
                cores=cores,
                batch_size=batch_size,
                detector_configs=detector_configs,
                scalar_keys=scalar_keys,
                precomputed_attrs=precomputed_attrs,
            )
        else:
            logger.info(
                "Run %s: SINGLE path — loading detector into memory", run_number
            )
            self._load_detector(run)
            run_pipeline(run, self.config.pipeline)

        self.analyzed_runs.append(run)
        self._status_log.append(f"Completed run {run_number}")
        logger.info("Run %s: complete", run_number)

    if self.config.reduction:
        self._status_log.append("Running reduction steps")
        logger.info("Running %d reduction step(s)", len(self.config.reduction))
        reduction_results = run_reductions(
            self.analyzed_runs, self.config.reduction
        )
        self.results.update(reduction_results)

    for run in self.analyzed_runs:
        for key, value in run.results.items():
            self.results[f"run_{run.run_number}.{key}"] = value
        self._collect_run_attributes(run)

    self._status_log.append("Pipeline execution complete")

enable_logging(level=logging.INFO, log_file=None)

Attach handlers to the XSpect logger so pipeline progress is visible.

Call once before Pipeline.run():

from XSpect.controller.pipeline import enable_logging
enable_logging()                          # stderr only
enable_logging(log_file="xspect.log")     # stderr + file
enable_logging(log_file="xspect.log", level=logging.DEBUG)

Safe to call multiple times; it will not add duplicate handlers.

Parameters:

Name	Type	Description	Default
level	int	Logging level (default logging.INFO). Use logging.DEBUG for per-step lines.	INFO
log_file	str or None	If given, also write logs to this file (appended). Worker subprocesses do NOT inherit this handler, but the main process logs every batch's completion, so the file captures full pipeline progress including the point of any hang/OOM.	None

Source code in XSpect/controller/pipeline.py

def enable_logging(level=logging.INFO, log_file=None):
    """Attach handlers to the XSpect logger so pipeline progress is visible.

    Call once before ``Pipeline.run()``:

        from XSpect.controller.pipeline import enable_logging
        enable_logging()                          # stderr only
        enable_logging(log_file="xspect.log")     # stderr + file
        enable_logging(log_file="xspect.log", level=logging.DEBUG)

    Safe to call multiple times; it will not add duplicate handlers.

    Parameters
    ----------
    level : int
        Logging level (default logging.INFO). Use logging.DEBUG for per-step lines.
    log_file : str or None
        If given, also write logs to this file (appended). Worker subprocesses
        do NOT inherit this handler, but the main process logs every batch's
        completion, so the file captures full pipeline progress including the
        point of any hang/OOM.
    """
    logger.setLevel(level)

    fmt = logging.Formatter(
        "[XSpect %(asctime)s %(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
    )

    # stderr handler (for notebook/terminal)
    if not any(getattr(h, "_xspect_stream", False) for h in logger.handlers):
        sh = logging.StreamHandler()
        sh.setFormatter(fmt)
        sh._xspect_stream = True
        logger.addHandler(sh)

    # file handler
    if log_file is not None:
        log_file = os.path.abspath(os.path.expanduser(log_file))
        already = any(
            getattr(h, "_xspect_file", None) == log_file for h in logger.handlers
        )
        if not already:
            fh = logging.FileHandler(log_file, mode="a")
            fh.setFormatter(fmt)
            fh._xspect_file = log_file
            logger.addHandler(fh)
            logger.info("XSpect logging to file: %s", log_file)

    return logger

XSpect.controller.config_parser

YAML configuration parser for XSpect pipelines.

Parses a YAML file into frozen dataclass structures and validates that all referenced steps exist in the registry.

ConfigValidationError

Bases: ValueError

Raised when YAML configuration is invalid.

Source code in XSpect/controller/config_parser.py

class ConfigValidationError(ValueError):
    """Raised when YAML configuration is invalid."""

    pass

parse_yaml(path, validate_steps=True)

Parse a YAML pipeline configuration file.

Parameters:

Name	Type	Description	Default
path	str	Path to the YAML file.	required
validate_steps	bool	If True, validate that all step names exist in the registry.	True

Returns:

Type	Description
PipelineConfig	Frozen dataclass with all configuration sections.

Raises:

Type	Description
ConfigValidationError	If the YAML is missing required sections or contains invalid entries.

Source code in XSpect/controller/config_parser.py

def parse_yaml(path: str, validate_steps: bool = True) -> PipelineConfig:
    """
    Parse a YAML pipeline configuration file.

    Parameters
    ----------
    path : str
        Path to the YAML file.
    validate_steps : bool
        If True, validate that all step names exist in the registry.

    Returns
    -------
    PipelineConfig
        Frozen dataclass with all configuration sections.

    Raises
    ------
    ConfigValidationError
        If the YAML is missing required sections or contains invalid entries.
    """
    path = Path(path)
    if not path.exists():
        raise ConfigValidationError(f"Config file not found: {path}")

    with open(path, "r") as f:
        raw = yaml.safe_load(f)

    if not isinstance(raw, dict):
        raise ConfigValidationError("YAML root must be a mapping")

    if "experiment" not in raw:
        raise ConfigValidationError("Missing required section: 'experiment'")
    if "data" not in raw:
        raise ConfigValidationError("Missing required section: 'data'")
    if "pipeline" not in raw:
        raise ConfigValidationError("Missing required section: 'pipeline'")

    experiment = _parse_experiment(raw["experiment"])
    data = _parse_data(raw["data"])
    pipeline_steps = _parse_steps(raw["pipeline"], "pipeline")

    reduction_steps = []
    if "reduction" in raw and raw["reduction"]:
        reduction_steps = _parse_steps(raw["reduction"], "reduction")

    output = OutputConfig()
    if "output" in raw and raw["output"]:
        output = OutputConfig(
            format=raw["output"].get("format", "hdf5"),
            path=raw["output"].get("path", "./results/"),
        )

    if validate_steps:
        _validate_step_names(pipeline_steps, "pipeline")
        _validate_step_names(reduction_steps, "reduction")

    return PipelineConfig(
        experiment=experiment,
        data=data,
        pipeline=pipeline_steps,
        reduction=reduction_steps,
        output=output,
    )

XSpect.controller.batch_manager

Batch manager for parallel pipeline execution.

Splits runs into shot-range batches, optionally parallelizes via multiprocessing.Pool, and reconverges batch results.

reconverge_results(batch_results)

Merge results from multiple batches by summing numeric arrays.

For numpy arrays: sums across batches (appropriate for photon-counting spectroscopy where batch spectra should be summed). For scalars: sums them. For non-numeric values: takes the last batch's value (geometry axes, etc.).

Parameters:

Name	Type	Description	Default
batch_results	list[dict]	List of attribute dicts from each batch.	required

Returns:

Type	Description
dict	Merged results with summed arrays.

Source code in XSpect/controller/batch_manager.py

def reconverge_results(batch_results: list[dict]) -> dict:
    """
    Merge results from multiple batches by summing numeric arrays.

    For numpy arrays: sums across batches (appropriate for photon-counting
    spectroscopy where batch spectra should be summed).
    For scalars: sums them.
    For non-numeric values: takes the last batch's value (geometry axes, etc.).

    Parameters
    ----------
    batch_results : list[dict]
        List of attribute dicts from each batch.

    Returns
    -------
    dict
        Merged results with summed arrays.
    """
    if not batch_results:
        return {}

    if len(batch_results) == 1:
        return batch_results[0]

    merged = {}
    all_keys = set()
    for br in batch_results:
        all_keys.update(br.keys())

    for key in all_keys:
        values = [br[key] for br in batch_results if key in br]
        if len(values) == 1:
            merged[key] = values[0]
            continue

        first = values[0]
        if isinstance(first, np.ndarray):
            if (
                key in ("ccm_energies", "ccm_bins", "time_bins", "time_delays")
                or key.endswith("_energy")
                or key.endswith("_axis")
                or key.endswith("_bins")
                or key.endswith("_delays")
            ):
                merged[key] = first
            else:
                shapes = [v.shape for v in values]
                if all(s == shapes[0] for s in shapes):
                    merged[key] = np.nansum(values, axis=0)
                else:
                    try:
                        merged[key] = np.concatenate(values, axis=0)
                    except ValueError:
                        merged[key] = values[-1]
        elif isinstance(first, (int, float)):
            # Per-detector geometry scalars (angles, etc.) are identical across
            # batches — take the mean rather than summing.
            if key.endswith("_angle"):
                merged[key] = float(np.mean(values))
            else:
                merged[key] = sum(values)
        else:
            merged[key] = values[-1]

    return merged

run_batched(run, pipeline_steps, cores=1, batch_size=2000, detector_configs=None, scalar_keys=None, precomputed_attrs=None)

Execute pipeline steps on a run with optional batch parallelism.

If cores == 1, runs sequentially without spawning a Pool. If cores > 1, splits into batches and uses multiprocessing (reloading data from HDF5 in each worker to avoid pickling large arrays).

After all batches complete, reconverged attributes are set on the original run object so downstream code can access them.

Parameters:

Name	Type	Description	Default
run	spectroscopy_run	The run object. Must have a total_shots attribute or equivalent.	required
pipeline_steps	list[StepConfig]	Steps to execute on each batch.	required
cores	int	Number of worker processes.	1
batch_size	int	Shots per batch.	2000
detector_configs	list of tuples	[(hdf5_path, name, transpose), ...] for parallel reloading.	None
scalar_keys	list of tuples	[(hdf5_path, friendly_name), ...] for parallel reloading.	None
precomputed_attrs	dict	Scalar/axis attributes pre-computed on the full run (e.g. ccm_bins, ccm_energies, ccm_bin_indices). Static attributes are copied directly into each batch; per-shot arrays (shape[0] == total_shots) are sliced.	None

Source code in XSpect/controller/batch_manager.py

def run_batched(
    run,
    pipeline_steps: list[StepConfig],
    cores: int = 1,
    batch_size: int = 2000,
    detector_configs=None,
    scalar_keys=None,
    precomputed_attrs=None,
) -> None:
    """
    Execute pipeline steps on a run with optional batch parallelism.

    If cores == 1, runs sequentially without spawning a Pool.
    If cores > 1, splits into batches and uses multiprocessing (reloading
    data from HDF5 in each worker to avoid pickling large arrays).

    After all batches complete, reconverged attributes are set on the
    original run object so downstream code can access them.

    Parameters
    ----------
    run : spectroscopy_run
        The run object. Must have a total_shots attribute or equivalent.
    pipeline_steps : list[StepConfig]
        Steps to execute on each batch.
    cores : int
        Number of worker processes.
    batch_size : int
        Shots per batch.
    detector_configs : list of tuples, optional
        [(hdf5_path, name, transpose), ...] for parallel reloading.
    scalar_keys : list of tuples, optional
        [(hdf5_path, friendly_name), ...] for parallel reloading.
    precomputed_attrs : dict, optional
        Scalar/axis attributes pre-computed on the full run (e.g. ccm_bins,
        ccm_energies, ccm_bin_indices).  Static attributes are copied directly
        into each batch; per-shot arrays (shape[0] == total_shots) are sliced.
    """
    total_shots = getattr(run, "total_shots", None)
    if total_shots is None:
        run_pipeline(run, pipeline_steps)
        return

    batches = split_into_batches(total_shots, batch_size)

    if not batches:
        return

    logger.info("run_batched: %d batches, cores=%d", len(batches), cores)

    if cores <= 1 or len(batches) == 1:
        batch_results = []
        for i, batch_range in enumerate(batches):
            logger.info(
                "  [seq] batch %d/%d shots %s", i + 1, len(batches), batch_range
            )
            result = _process_batch_sequential(
                batch_range,
                run,
                pipeline_steps,
                detector_configs=detector_configs,
                scalar_keys=scalar_keys,
                precomputed_attrs=precomputed_attrs,
            )
            batch_results.append(result)
    else:
        run_file = getattr(run, "run_file", None)
        start_index = getattr(run, "start_index", 0)
        if run_file is None:
            batch_results = []
            for batch_range in batches:
                result = _process_batch_sequential(
                    batch_range,
                    run,
                    pipeline_steps,
                    precomputed_attrs=precomputed_attrs,
                )
                batch_results.append(result)
        else:
            if detector_configs is None:
                detector_configs = []
            if scalar_keys is None:
                scalar_keys = []
            process_fn = partial(
                _process_batch_parallel,
                run_file=run_file,
                start_index=start_index,
                detector_configs=detector_configs,
                scalar_keys=scalar_keys,
                steps=pipeline_steps,
                precomputed_attrs=precomputed_attrs,
                parent_total_shots=total_shots,
            )
            logger.info(
                "  [parallel] dispatching %d batches across %d workers "
                "(≈%.1f GB/batch est. for detector load)",
                len(batches),
                cores,
                _estimate_batch_gb(detector_configs, batch_size),
            )
            batch_results = []
            try:
                with Pool(processes=cores) as pool:
                    # imap_unordered surfaces results as they complete, so a
                    # hang/crash is visible at the batch level instead of a
                    # silent block on pool.map.
                    for n_done, result in enumerate(
                        pool.imap_unordered(process_fn, batches), start=1
                    ):
                        batch_results.append(result)
                        logger.info(
                            "  [parallel] %d/%d batches done", n_done, len(batches)
                        )
            except Exception as exc:
                # A worker killed by the OOM-killer surfaces here as an opaque
                # error (e.g. "process died with signal"). Make it actionable.
                logger.error(
                    "Batch worker failed after %d/%d batches: %s. "
                    "This is often an out-of-memory kill — reduce batch_size or "
                    "cores, or tighten row_range in the YAML.",
                    len(batch_results),
                    len(batches),
                    exc,
                )
                raise

    merged = reconverge_results(batch_results)
    for key, value in merged.items():
        setattr(run, key, value)
    run.update_status(f"Batched execution complete: {len(batches)} batches reconverged")

split_into_batches(total_shots, batch_size)

Split a range of shots into contiguous batches.

Parameters:

Name	Type	Description	Default
total_shots	int	Total number of shots in the run.	required
batch_size	int	Maximum shots per batch.	required

Returns:

Type	Description
list of (start, end) tuples	Each tuple defines a half-open range [start, end).

Source code in XSpect/controller/batch_manager.py

def split_into_batches(total_shots: int, batch_size: int) -> list[tuple[int, int]]:
    """
    Split a range of shots into contiguous batches.

    Parameters
    ----------
    total_shots : int
        Total number of shots in the run.
    batch_size : int
        Maximum shots per batch.

    Returns
    -------
    list of (start, end) tuples
        Each tuple defines a half-open range [start, end).
    """
    if total_shots <= 0:
        return []
    if batch_size <= 0:
        raise ValueError("batch_size must be positive")

    batches = []
    start = 0
    while start < total_shots:
        end = min(start + batch_size, total_shots)
        batches.append((start, end))
        start = end
    return batches