qmri.pipelines

End-to-end quantitative MRI workflows that combine the pure signal models in qmri with the file handling in qmri.io. Pipelines are file-in / file-out: they load images, run the fit, and write maps and reports.

Provided by the qmri-pipelines package.

Multi-Echo Thermometry

multiecho

Multi-echo MR thermometry pipeline.

This pipeline estimates temperature from multi-echo magnitude images and a segmentation/label map using the dual-resonance model (:func:qmri.thermometry.fit_multiecho_thermometry_image).

At a high level it:

• Loads one or more 4D multi-echo magnitude NIfTI images (the echo dimension is the last axis).
• Loads a 3D segmentation/label map co-located with the images.
• Loads echo times (in seconds) for each image, concatenates them, and sorts all echoes by echo time.
• Determines the magnetic field strength \(B_0\) (Tesla) from an explicit argument or from a JSON sidecar (ImagingFrequency in MHz or MagneticFieldStrength in Tesla).
• Runs region-wise, voxel-wise or bootstrap region-wise fitting and (optionally) writes a temperature map NIfTI and a JSON report.

MultiEchoThermometryReport dataclass

MultiEchoThermometryReport(
    input_files: list[Path],
    segmentation_file: Path,
    output_file: Path | None,
    magnetic_field_tesla: float,
    analysis_method: str,
    n_bootstrap: int | None,
    echo_times: list[float],
    regions: list[RegionThermometryResult],
    acquisition_date_time: list[str],
    processing_date: str,
    processing_time_seconds: float,
)

Structured report from :func:run_multiecho_thermometry.

Attributes:

Name	Type	Description
input_files	list[Path]	The multi-echo input image paths, in the order supplied.
segmentation_file	Path	The segmentation/label-map path.
output_file	Path | None	Path of the saved temperature map, or None if outputs were not written.
magnetic_field_tesla	float	Magnetic field strength used for the calibration.
analysis_method	str	The analysis method that was run.
n_bootstrap	int | None	Number of bootstrap samples (None unless the method was regionwise_bootstrap).
echo_times	list[float]	All echo times in seconds, concatenated and sorted ascending.
regions	list[RegionThermometryResult]	Per-region results, one per non-empty segmentation label.
acquisition_date_time	list[str]	Acquisition date/time string per input image ("Unknown" when no sidecar metadata was found).
processing_date	str	Local date/time the pipeline finished.
processing_time_seconds	float	Wall-clock processing time in seconds.

to_dict

to_dict() -> dict[str, object]

Return a JSON-serialisable dictionary of the report.

Source code in packages/qmri-pipelines/src/qmri/pipelines/thermometry/multiecho.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-serialisable dictionary of the report."""
    return {
        "input_files": [str(f) for f in self.input_files],
        "segmentation_file": str(self.segmentation_file),
        "output_file": str(self.output_file) if self.output_file else None,
        "acquisition_date_time": self.acquisition_date_time,
        "processing_date": self.processing_date,
        "processing_time_seconds": self.processing_time_seconds,
        "magnetic_field_tesla": self.magnetic_field_tesla,
        "analysis_method": self.analysis_method,
        "n_bootstrap": self.n_bootstrap,
        "echo_times": self.echo_times,
        "report": [region.to_dict() for region in self.regions],
    }

run_multiecho_thermometry

run_multiecho_thermometry(
    multiecho_files: Sequence[str | Path],
    segmentation_file: str | Path,
    echo_times_files: Sequence[str | Path],
    *,
    method: RegionAnalysisMethod = "regionwise",
    n_bootstrap: int = 100,
    df_init: DfInitMethod = "multistart",
    magnetic_field_tesla: float | None = None,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, MultiEchoThermometryReport]

Run multi-echo thermometry over a set of images and a segmentation.

Parameters:

Name	Type	Description	Default
multiecho_files	Sequence[str | Path]	One or more 4D multi-echo magnitude NIfTI files. The echo dimension must be the last axis. All images must share the same spatial shape and affine.	required
segmentation_file	str | Path	A 3D segmentation/label-map NIfTI co-located with the multi-echo data. Label 0 is treated as background.	required
echo_times_files	Sequence[str | Path]	Echo-time text files (seconds), one per multi-echo image, in the same order. Each file's length must match the number of echoes in its image.	required
method	RegionAnalysisMethod	Analysis method — "regionwise" (default), "voxelwise" or "regionwise_bootstrap". See :func:qmri.thermometry.fit_multiecho_thermometry_image.	'regionwise'
n_bootstrap	int	Number of bootstrap samples (regionwise_bootstrap only).	100
df_init	DfInitMethod	Frequency starting-value strategy — "multistart" (default), "fixed" or "lombscargle". See :data:qmri.thermometry.DfInitMethod.	'multistart'
magnetic_field_tesla	float | None	Magnetic field strength in Tesla. If None, it is read from the first JSON sidecar containing ImagingFrequency or MagneticFieldStrength.	None
output_dir	str | Path | None	Directory for output files. Defaults to the directory of the first input image. Only used when save_outputs is True.	None
output_prefix	str | None	Prefix for output filenames. Defaults to the first input image's stem.	None
save_outputs	bool	If True (default), write <prefix>_temperature_map.nii.gz and <prefix>_report.json.	True

Returns:

Type	Description
NiftiImage	A tuple (temperature_image, report) where temperature_image is a
MultiEchoThermometryReport	class:qmri.io.NiftiImage of the 3D temperature map (°C) co-located with
tuple[NiftiImage, MultiEchoThermometryReport]	the segmentation, and report is a :class:MultiEchoThermometryReport.

Raises:

Type	Description
ValueError	If no images are supplied, the number of images and echo-time files differ, image dimensions/affines are inconsistent, an image's echo count does not match its echo times, or the magnetic field strength cannot be determined.

Source code in packages/qmri-pipelines/src/qmri/pipelines/thermometry/multiecho.py

def run_multiecho_thermometry(
    multiecho_files: Sequence[str | Path],
    segmentation_file: str | Path,
    echo_times_files: Sequence[str | Path],
    *,
    method: RegionAnalysisMethod = "regionwise",
    n_bootstrap: int = 100,
    df_init: DfInitMethod = "multistart",
    magnetic_field_tesla: float | None = None,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, MultiEchoThermometryReport]:
    r"""Run multi-echo thermometry over a set of images and a segmentation.

    Args:
        multiecho_files: One or more 4D multi-echo magnitude NIfTI files. The
            echo dimension must be the last axis. All images must share the same
            spatial shape and affine.
        segmentation_file: A 3D segmentation/label-map NIfTI co-located with the
            multi-echo data. Label ``0`` is treated as background.
        echo_times_files: Echo-time text files (seconds), one per multi-echo
            image, in the same order. Each file's length must match the number of
            echoes in its image.
        method: Analysis method — ``"regionwise"`` (default), ``"voxelwise"`` or
            ``"regionwise_bootstrap"``. See
            :func:`qmri.thermometry.fit_multiecho_thermometry_image`.
        n_bootstrap: Number of bootstrap samples (``regionwise_bootstrap`` only).
        df_init: Frequency starting-value strategy — ``"multistart"`` (default),
            ``"fixed"`` or ``"lombscargle"``. See
            :data:`qmri.thermometry.DfInitMethod`.
        magnetic_field_tesla: Magnetic field strength in Tesla. If ``None``, it is
            read from the first JSON sidecar containing ``ImagingFrequency`` or
            ``MagneticFieldStrength``.
        output_dir: Directory for output files. Defaults to the directory of the
            first input image. Only used when ``save_outputs`` is ``True``.
        output_prefix: Prefix for output filenames. Defaults to the first input
            image's stem.
        save_outputs: If ``True`` (default), write ``<prefix>_temperature_map.nii.gz``
            and ``<prefix>_report.json``.

    Returns:
        A tuple ``(temperature_image, report)`` where ``temperature_image`` is a
        :class:`qmri.io.NiftiImage` of the 3D temperature map (°C) co-located with
        the segmentation, and ``report`` is a :class:`MultiEchoThermometryReport`.

    Raises:
        ValueError: If no images are supplied, the number of images and
            echo-time files differ, image dimensions/affines are inconsistent,
            an image's echo count does not match its echo times, or the magnetic
            field strength cannot be determined.
    """
    start_time = time.perf_counter()

    multiecho_paths = [Path(f) for f in multiecho_files]
    echo_times_paths = [Path(f) for f in echo_times_files]
    segmentation_path = Path(segmentation_file)

    if not multiecho_paths:
        msg = "At least one multi-echo image must be provided."
        raise ValueError(msg)
    if len(multiecho_paths) != len(echo_times_paths):
        msg = (
            f"Number of multi-echo images ({len(multiecho_paths)}) must match the "
            f"number of echo-time files ({len(echo_times_paths)})."
        )
        raise ValueError(msg)

    echo_times_per_image = [_load_echo_times(p) for p in echo_times_paths]
    images = [load_nifti_image(p) for p in multiecho_paths]
    reference = images[0]

    for image, path in zip(images, multiecho_paths, strict=True):
        if image.data.ndim != 4:
            msg = f"Multi-echo image must be 4D (x, y, z, echo): {path}"
            raise ValueError(msg)
        if image.data.shape[:3] != reference.data.shape[:3]:
            msg = "All multi-echo images must share the same spatial shape."
            raise ValueError(msg)
        if not np.allclose(image.affine, reference.affine):
            msg = "All multi-echo images must share the same affine."
            raise ValueError(msg)
    for image, echo_times, path in zip(
        images, echo_times_per_image, multiecho_paths, strict=True
    ):
        if image.data.shape[-1] != echo_times.shape[0]:
            msg = (
                f"Image {path} has {image.data.shape[-1]} echoes but "
                f"{echo_times.shape[0]} echo times were provided."
            )
            raise ValueError(msg)

    # Determine B0 and acquisition times from JSON sidecars.
    sidecars = [load_sidecar(p) for p in multiecho_paths]
    acquisition_date_time: list[str] = []
    detected_field: float | None = None
    for sidecar in sidecars:
        if detected_field is None:
            detected_field = _magnetic_field_from_sidecar(sidecar)
        if "AcquisitionDateTime" in sidecar:
            acquisition_date_time.append(str(sidecar["AcquisitionDateTime"]))
        elif "AcquisitionTime" in sidecar:
            acquisition_date_time.append(str(sidecar["AcquisitionTime"]))
        else:
            acquisition_date_time.append("Unknown")

    field_tesla = (
        magnetic_field_tesla if magnetic_field_tesla is not None else detected_field
    )
    if field_tesla is None:
        msg = (
            "Magnetic field strength could not be determined. Pass "
            "magnetic_field_tesla explicitly, or provide a JSON sidecar with "
            "'ImagingFrequency' (MHz) or 'MagneticFieldStrength' (Tesla)."
        )
        raise ValueError(msg)

    # Concatenate echoes across images and sort by echo time.
    signal = np.concatenate([image.data for image in images], axis=3)
    all_echo_times = np.concatenate(echo_times_per_image)
    order = np.argsort(all_echo_times)
    signal = signal[:, :, :, order]
    sorted_echo_times = all_echo_times[order]

    segmentation_image = load_nifti_image(segmentation_path)
    if (
        segmentation_image.data.ndim != 3
        or segmentation_image.data.shape != reference.data.shape[:3]
    ):
        msg = (
            "Segmentation must be 3D and match the spatial shape of the "
            "multi-echo images."
        )
        raise ValueError(msg)

    temperature_map, regions = fit_multiecho_thermometry_image(
        signal=signal,
        segmentation=segmentation_image.data,
        echo_times=sorted_echo_times,
        magnetic_field_tesla=field_tesla,
        method=method,
        n_bootstrap=n_bootstrap,
        df_init=df_init,
    )

    temperature_image = NiftiImage(
        data=temperature_map,
        header=reference.header,
        affine=reference.affine,
    )

    out_dir: Path | None = None
    prefix: str | None = None
    output_path: Path | None = None
    if save_outputs:
        out_dir = (
            Path(output_dir) if output_dir is not None else multiecho_paths[0].parent
        )
        out_dir.mkdir(parents=True, exist_ok=True)
        prefix = output_prefix or strip_nifti_suffix(multiecho_paths[0]).stem
        output_path = out_dir / f"{prefix}_temperature_map.nii.gz"
        save_nifti(
            temperature_map,
            output_path,
            header=reference.header,
            affine=reference.affine,
        )

    report = MultiEchoThermometryReport(
        input_files=multiecho_paths,
        segmentation_file=segmentation_path,
        output_file=output_path,
        magnetic_field_tesla=float(field_tesla),
        analysis_method=method,
        n_bootstrap=n_bootstrap if method == "regionwise_bootstrap" else None,
        echo_times=[float(t) for t in sorted_echo_times],
        regions=regions,
        acquisition_date_time=acquisition_date_time,
        processing_date=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        processing_time_seconds=time.perf_counter() - start_time,
    )

    if save_outputs and out_dir is not None and prefix is not None:
        report_path = out_dir / f"{prefix}_report.json"
        with open(report_path, "w", encoding="utf-8") as report_file:
            json.dump(report.to_dict(), report_file, indent=2)

    return temperature_image, report

ASL Perfusion Quantification

asl

Arterial Spin Labelling (ASL) quantification pipeline.

This pipeline produces a cerebral blood flow (CBF) map from an ASL NIfTI using the White Paper consensus equations in :mod:qmri.perfusion.asl.

At a high level it:

• Loads a 4D ASL NIfTI and the per-volume asl_context describing each volume as "control", "label" or "m0scan" (read from an explicit argument, a sibling *_aslcontext.tsv file, or the JSON sidecar).
• Averages the control, label and M0 volumes (or takes M0 from a separate file).
• Resolves the labelling parameters with the precedence explicit argument > JSON sidecar > sensible defaults.
• Quantifies CBF (ml/100g/min) using the pCASL/CASL or PASL equation, depending on ArterialSpinLabelingType.
• Optionally writes a CBF map NIfTI and a JSON report.

ASLQuantificationReport dataclass

ASLQuantificationReport(
    asl_file: Path,
    m0_file: Path | None,
    output_file: Path | None,
    label_type: str,
    asl_context: list[str],
    quantification_parameters: dict[str, float],
    n_valid_voxels: int,
    perfusion_mean: float,
    perfusion_std: float,
    processing_date: str,
    processing_time_seconds: float,
)

Structured report from :func:run_asl_quantification.

Attributes:

Name	Type	Description
asl_file	Path	The input ASL image path.
m0_file	Path | None	A separate M0 image path, or None if M0 came from the ASL file's m0scan volumes.
output_file	Path | None	Path of the saved CBF map, or None if outputs were not written.
label_type	str	The resolved ASL labelling type used for quantification.
asl_context	list[str]	The per-volume context labels used to split the ASL image.
quantification_parameters	dict[str, float]	The resolved labelling parameters actually used for quantification.
n_valid_voxels	int	Number of voxels with non-zero M0 (i.e. voxels where CBF is defined) included in the statistics below.
perfusion_mean	float	Mean CBF (ml/100g/min) over the valid voxels.
perfusion_std	float	Standard deviation of CBF (ml/100g/min) over the valid voxels.
processing_date	str	Local date/time the pipeline finished.
processing_time_seconds	float	Wall-clock processing time in seconds.

to_dict

to_dict() -> dict[str, object]

Return a JSON-serialisable dictionary of the report.

Source code in packages/qmri-pipelines/src/qmri/pipelines/perfusion/asl.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-serialisable dictionary of the report."""
    return {
        "asl_file": str(self.asl_file),
        "m0_file": str(self.m0_file) if self.m0_file else None,
        "output_file": str(self.output_file) if self.output_file else None,
        "label_type": self.label_type,
        "asl_context": self.asl_context,
        "quantification_parameters": self.quantification_parameters,
        "n_valid_voxels": self.n_valid_voxels,
        "perfusion_mean": self.perfusion_mean,
        "perfusion_std": self.perfusion_std,
        "processing_date": self.processing_date,
        "processing_time_seconds": self.processing_time_seconds,
    }

run_asl_quantification

run_asl_quantification(
    asl_file: str | Path,
    *,
    asl_context: Sequence[str] | None = None,
    m0_file: str | Path | None = None,
    label_type: LabelType | None = None,
    post_label_delay: float | None = None,
    label_duration: float | None = None,
    bolus_duration: float | None = None,
    label_efficiency: float | None = None,
    t1_blood: float | None = None,
    partition_coefficient: float | None = None,
    magnetic_field_tesla: float | None = None,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, ASLQuantificationReport]

Quantify cerebral blood flow (CBF) from an ASL image.

The control, label and M0 volumes are identified from asl_context and averaged. CBF is then computed with the White Paper pCASL/CASL or PASL equation, depending on the resolved labelling type.

Labelling parameters are resolved with the precedence explicit argument > JSON sidecar > default. Sidecar keys follow BIDS (ArterialSpinLabelingType, PostLabelingDelay, LabelingDuration, BolusCutOffDelayTime, LabelingEfficiency, BloodBrainPartitionCoefficient, T1ArterialBlood, MagneticFieldStrength).

Parameters:

Name	Type	Description	Default
asl_file	str | Path	A 4D ASL NIfTI containing control/label (and optionally M0) volumes.	required
asl_context	Sequence[str] | None	Per-volume labels ("control", "label", "m0scan"). If None, read from a sibling *_aslcontext.tsv or the JSON sidecar.	None
m0_file	str | Path | None	Optional separate M0 NIfTI. If omitted, M0 is taken from the m0scan volumes of asl_file.	None
label_type	LabelType | None	"pcasl", "casl" or "pasl". If None, read from the sidecar's ArterialSpinLabelingType.	None
post_label_delay	float | None	Post-label delay (s) for pCASL/CASL, or the inversion time (TI, s) for PASL.	None
label_duration	float | None	Label duration (s); pCASL/CASL only.	None
bolus_duration	float | None	Bolus duration (TI1, s); PASL only.	None
label_efficiency	float | None	Labelling efficiency. Defaults to 0.85 (pCASL/CASL) or 0.98 (PASL).	None
t1_blood	float | None	T1 of arterial blood (s). Defaults from field strength (1.35 s at 1.5 T, 1.65 s at 3 T) or 1.65 s.	None
partition_coefficient	float | None	Blood-brain partition coefficient (ml/g). Defaults to 0.9.	None
magnetic_field_tesla	float | None	Field strength (T), used only to pick a default t1_blood when neither it nor the sidecar provides one.	None
output_dir	str | Path | None	Directory for output files. Defaults to the directory of asl_file. Only used when save_outputs is True.	None
output_prefix	str | None	Prefix for output filenames. Defaults to the stem of asl_file.	None
save_outputs	bool	If True (default), write <prefix>_cbf.nii.gz and <prefix>_report.json.	True

Returns:

Type	Description
NiftiImage	A tuple (cbf_image, report) where cbf_image is a
ASLQuantificationReport	class:qmri.io.NiftiImage of the CBF map (ml/100g/min) co-located with
tuple[NiftiImage, ASLQuantificationReport]	the input, and report is an :class:ASLQuantificationReport.

Raises:

Type	Description
ValueError	If the ASL context cannot be determined or mismatches the number of volumes; if control, label or M0 volumes are missing; if the labelling type is unknown; or if a required labelling parameter (post-label delay, label duration, or bolus duration) is missing.

Source code in packages/qmri-pipelines/src/qmri/pipelines/perfusion/asl.py

def run_asl_quantification(
    asl_file: str | Path,
    *,
    asl_context: Sequence[str] | None = None,
    m0_file: str | Path | None = None,
    label_type: LabelType | None = None,
    post_label_delay: float | None = None,
    label_duration: float | None = None,
    bolus_duration: float | None = None,
    label_efficiency: float | None = None,
    t1_blood: float | None = None,
    partition_coefficient: float | None = None,
    magnetic_field_tesla: float | None = None,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, ASLQuantificationReport]:
    r"""Quantify cerebral blood flow (CBF) from an ASL image.

    The control, label and M0 volumes are identified from ``asl_context`` and
    averaged. CBF is then computed with the White Paper pCASL/CASL or PASL
    equation, depending on the resolved labelling type.

    Labelling parameters are resolved with the precedence
    *explicit argument > JSON sidecar > default*. Sidecar keys follow BIDS
    (``ArterialSpinLabelingType``, ``PostLabelingDelay``, ``LabelingDuration``,
    ``BolusCutOffDelayTime``, ``LabelingEfficiency``,
    ``BloodBrainPartitionCoefficient``, ``T1ArterialBlood``,
    ``MagneticFieldStrength``).

    Args:
        asl_file: A 4D ASL NIfTI containing control/label (and optionally M0)
            volumes.
        asl_context: Per-volume labels (``"control"``, ``"label"``,
            ``"m0scan"``). If ``None``, read from a sibling ``*_aslcontext.tsv``
            or the JSON sidecar.
        m0_file: Optional separate M0 NIfTI. If omitted, M0 is taken from the
            ``m0scan`` volumes of ``asl_file``.
        label_type: ``"pcasl"``, ``"casl"`` or ``"pasl"``. If ``None``, read
            from the sidecar's ``ArterialSpinLabelingType``.
        post_label_delay: Post-label delay (s) for pCASL/CASL, or the inversion
            time (TI, s) for PASL.
        label_duration: Label duration (s); pCASL/CASL only.
        bolus_duration: Bolus duration (TI1, s); PASL only.
        label_efficiency: Labelling efficiency. Defaults to 0.85 (pCASL/CASL) or
            0.98 (PASL).
        t1_blood: T1 of arterial blood (s). Defaults from field strength
            (1.35 s at 1.5 T, 1.65 s at 3 T) or 1.65 s.
        partition_coefficient: Blood-brain partition coefficient (ml/g).
            Defaults to 0.9.
        magnetic_field_tesla: Field strength (T), used only to pick a default
            ``t1_blood`` when neither it nor the sidecar provides one.
        output_dir: Directory for output files. Defaults to the directory of
            ``asl_file``. Only used when ``save_outputs`` is ``True``.
        output_prefix: Prefix for output filenames. Defaults to the stem of
            ``asl_file``.
        save_outputs: If ``True`` (default), write ``<prefix>_cbf.nii.gz`` and
            ``<prefix>_report.json``.

    Returns:
        A tuple ``(cbf_image, report)`` where ``cbf_image`` is a
        :class:`qmri.io.NiftiImage` of the CBF map (ml/100g/min) co-located with
        the input, and ``report`` is an :class:`ASLQuantificationReport`.

    Raises:
        ValueError: If the ASL context cannot be determined or mismatches the
            number of volumes; if control, label or M0 volumes are missing; if
            the labelling type is unknown; or if a required labelling parameter
            (post-label delay, label duration, or bolus duration) is missing.
    """
    start_time = time.perf_counter()

    asl_path = Path(asl_file)
    asl_image = load_nifti_image(asl_path)
    sidecar = load_sidecar(asl_path)

    if asl_image.data.ndim != 4:
        msg = f"ASL image must be 4D (x, y, z, volume): {asl_path}"
        raise ValueError(msg)

    context = _resolve_asl_context(asl_path, asl_context, sidecar)
    n_volumes = asl_image.data.shape[-1]
    if len(context) != n_volumes:
        msg = (
            f"ASL context length ({len(context)}) does not match the number of "
            f"volumes in the image ({n_volumes})."
        )
        raise ValueError(msg)

    control = _mean_of_volumes(asl_image.data, context, "control")
    label = _mean_of_volumes(asl_image.data, context, "label")
    if control is None or label is None:
        msg = "ASL context must contain at least one 'control' and one 'label' volume."
        raise ValueError(msg)

    m0_path: Path | None = None
    if m0_file is not None:
        m0_path = Path(m0_file)
        m0 = load_nifti_image(m0_path).data
        if m0.shape != control.shape:
            msg = (
                f"M0 image shape {m0.shape} does not match the ASL volume shape "
                f"{control.shape}."
            )
            raise ValueError(msg)
    else:
        m0_volumes = _mean_of_volumes(asl_image.data, context, "m0scan")
        if m0_volumes is None:
            msg = (
                "No 'm0scan' volume found in the ASL context and no m0_file was "
                "provided."
            )
            raise ValueError(msg)
        m0 = m0_volumes

    resolved_label_type = (label_type or sidecar.get("ArterialSpinLabelingType") or "")
    resolved_label_type = str(resolved_label_type).lower()
    if resolved_label_type not in ("pcasl", "casl", "pasl"):
        msg = (
            "ASL labelling type must be 'pcasl', 'casl' or 'pasl'. Pass "
            "label_type explicitly or set 'ArterialSpinLabelingType' in the "
            f"sidecar (got {resolved_label_type!r})."
        )
        raise ValueError(msg)

    efficiency = _resolve(
        label_efficiency,
        sidecar,
        "LabelingEfficiency",
        _DEFAULT_LABEL_EFFICIENCY[resolved_label_type],
    )
    partition = _resolve(
        partition_coefficient,
        sidecar,
        "BloodBrainPartitionCoefficient",
        _DEFAULT_PARTITION_COEFFICIENT,
    )
    t1b = _resolve_t1_blood(t1_blood, sidecar, magnetic_field_tesla)
    pld = _resolve(post_label_delay, sidecar, "PostLabelingDelay", None)

    # mypy/ruff: defaults above are never None, so narrow for the call below.
    assert efficiency is not None
    assert partition is not None

    parameters: dict[str, float] = {
        "label_efficiency": efficiency,
        "t1_blood": t1b,
        "partition_coefficient": partition,
    }

    if resolved_label_type in ("pcasl", "casl"):
        duration = _resolve(label_duration, sidecar, "LabelingDuration", None)
        if pld is None or duration is None:
            msg = (
                "pCASL/CASL quantification requires both a post-label delay and a "
                "label duration (via arguments or the sidecar)."
            )
            raise ValueError(msg)
        parameters["post_label_delay"] = pld
        parameters["label_duration"] = duration
        result = quantify_pcasl(
            control,
            label,
            m0,
            label_duration=duration,
            post_label_delay=pld,
            label_efficiency=efficiency,
            t1_blood=t1b,
            partition_coefficient=partition,
        )
    else:  # pasl
        bolus = _resolve(bolus_duration, sidecar, "BolusCutOffDelayTime", None)
        if pld is None or bolus is None:
            msg = (
                "PASL quantification requires both an inversion time "
                "(post_label_delay) and a bolus duration (via arguments or the "
                "sidecar)."
            )
            raise ValueError(msg)
        parameters["inversion_time"] = pld
        parameters["bolus_duration"] = bolus
        result = quantify_pasl(
            control,
            label,
            m0,
            bolus_duration=bolus,
            inversion_time=pld,
            label_efficiency=efficiency,
            t1_blood=t1b,
            partition_coefficient=partition,
        )

    perfusion = result.perfusion
    cbf_image = NiftiImage(
        data=perfusion,
        header=asl_image.header,
        affine=asl_image.affine,
    )

    valid = np.abs(m0) > 0
    valid_values = perfusion[valid]
    perfusion_mean = float(np.mean(valid_values)) if valid_values.size else 0.0
    perfusion_std = float(np.std(valid_values)) if valid_values.size else 0.0

    out_dir: Path | None = None
    prefix: str | None = None
    output_path: Path | None = None
    if save_outputs:
        out_dir = Path(output_dir) if output_dir is not None else asl_path.parent
        out_dir.mkdir(parents=True, exist_ok=True)
        prefix = output_prefix or strip_nifti_suffix(asl_path).stem
        output_path = out_dir / f"{prefix}_cbf.nii.gz"
        save_nifti(
            perfusion,
            output_path,
            header=asl_image.header,
            affine=asl_image.affine,
        )

    report = ASLQuantificationReport(
        asl_file=asl_path,
        m0_file=m0_path,
        output_file=output_path,
        label_type=resolved_label_type,
        asl_context=list(context),
        quantification_parameters=parameters,
        n_valid_voxels=int(valid_values.size),
        perfusion_mean=perfusion_mean,
        perfusion_std=perfusion_std,
        processing_date=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        processing_time_seconds=time.perf_counter() - start_time,
    )

    if save_outputs and out_dir is not None and prefix is not None:
        report_path = out_dir / f"{prefix}_report.json"
        with open(report_path, "w", encoding="utf-8") as report_file:
            json.dump(report.to_dict(), report_file, indent=2)

    return cbf_image, report

Magnetisation Transfer Ratio

mtr

Magnetisation Transfer Ratio (MTR) pipeline.

This pipeline calculates an MTR map from images acquired with and without bound-pool saturation, using :func:qmri.transfer.calculate_mtr.

At a high level it:

• Loads the saturated and unsaturated images, either as two separate 3D NIfTI files or as a single 4D file whose last axis holds the two volumes (ordered [unsaturated, saturated]).
• Checks the two images are co-located (matching spatial shape and affine).
• Computes the MTR (percentage units) voxel-wise.
• Optionally writes an MTR map NIfTI and a JSON report.

The MTR is defined as

\[\text{MTR} = 100 \cdot \frac{S_0 - S_s}{S_0}\]

where \(S_0\) is the unsaturated signal and \(S_s\) the saturated signal.

MTRReport dataclass

MTRReport(
    input_files: list[Path],
    mode: str,
    output_file: Path | None,
    n_valid_voxels: int,
    mtr_mean: float,
    mtr_std: float,
    mtr_min: float,
    mtr_max: float,
    processing_date: str,
    processing_time_seconds: float,
)

Structured report from :func:run_mtr.

Attributes:

Name	Type	Description
input_files	list[Path]	The input image paths. Two entries (unsaturated, saturated) in separate-file mode, or one entry in combined-file mode.
mode	str	"separate" (two files) or "combined" (single 4D file).
output_file	Path | None	Path of the saved MTR map, or None if outputs were not written.
n_valid_voxels	int	Number of voxels with a non-zero unsaturated signal (i.e. voxels where MTR is defined) included in the statistics below.
mtr_mean	float	Mean MTR (pu) over the valid voxels.
mtr_std	float	Standard deviation of MTR (pu) over the valid voxels.
mtr_min	float	Minimum MTR (pu) over the valid voxels.
mtr_max	float	Maximum MTR (pu) over the valid voxels.
processing_date	str	Local date/time the pipeline finished.
processing_time_seconds	float	Wall-clock processing time in seconds.

to_dict

to_dict() -> dict[str, object]

Return a JSON-serialisable dictionary of the report.

Source code in packages/qmri-pipelines/src/qmri/pipelines/transfer/mtr.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-serialisable dictionary of the report."""
    return {
        "input_files": [str(f) for f in self.input_files],
        "mode": self.mode,
        "output_file": str(self.output_file) if self.output_file else None,
        "n_valid_voxels": self.n_valid_voxels,
        "mtr_mean": self.mtr_mean,
        "mtr_std": self.mtr_std,
        "mtr_min": self.mtr_min,
        "mtr_max": self.mtr_max,
        "processing_date": self.processing_date,
        "processing_time_seconds": self.processing_time_seconds,
    }

run_mtr

run_mtr(
    saturated_file: str | Path,
    unsaturated_file: str | Path | None = None,
    *,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, MTRReport]

Calculate a Magnetisation Transfer Ratio (MTR) map.

Two input modes are supported:

• Separate (unsaturated_file given): saturated_file and unsaturated_file are individual NIfTI images of the same shape and affine.
• Combined (unsaturated_file is None): saturated_file is a single 4D NIfTI whose last axis holds two volumes ordered [unsaturated, saturated].

Parameters:

Name	Type	Description	Default
saturated_file	str | Path	The image with bound-pool saturation, or — in combined mode — a 4D file containing both the unsaturated and saturated volumes (in that order).	required
unsaturated_file	str | Path | None	The image without bound-pool saturation. Omit to use combined mode.	None
output_dir	str | Path | None	Directory for output files. Defaults to the directory of saturated_file. Only used when save_outputs is True.	None
output_prefix	str | None	Prefix for output filenames. Defaults to the stem of saturated_file.	None
save_outputs	bool	If True (default), write <prefix>_mtr_map.nii.gz and <prefix>_report.json.	True

Returns:

Type	Description
NiftiImage	A tuple (mtr_image, report) where mtr_image is a
MTRReport	class:qmri.io.NiftiImage of the MTR map in percentage units (pu),
tuple[NiftiImage, MTRReport]	co-located with the input, and report is an :class:MTRReport.

Raises:

Type	Description
ValueError	If a combined file is not 4D with two volumes, or the two separate images differ in shape or affine.

Source code in packages/qmri-pipelines/src/qmri/pipelines/transfer/mtr.py

def run_mtr(
    saturated_file: str | Path,
    unsaturated_file: str | Path | None = None,
    *,
    output_dir: str | Path | None = None,
    output_prefix: str | None = None,
    save_outputs: bool = True,
) -> tuple[NiftiImage, MTRReport]:
    r"""Calculate a Magnetisation Transfer Ratio (MTR) map.

    Two input modes are supported:

    - **Separate** (``unsaturated_file`` given): ``saturated_file`` and
      ``unsaturated_file`` are individual NIfTI images of the same shape and
      affine.
    - **Combined** (``unsaturated_file`` is ``None``): ``saturated_file`` is a
      single 4D NIfTI whose last axis holds two volumes ordered
      ``[unsaturated, saturated]``.

    Args:
        saturated_file: The image with bound-pool saturation, or — in combined
            mode — a 4D file containing both the unsaturated and saturated
            volumes (in that order).
        unsaturated_file: The image without bound-pool saturation. Omit to use
            combined mode.
        output_dir: Directory for output files. Defaults to the directory of
            ``saturated_file``. Only used when ``save_outputs`` is ``True``.
        output_prefix: Prefix for output filenames. Defaults to the stem of
            ``saturated_file``.
        save_outputs: If ``True`` (default), write ``<prefix>_mtr_map.nii.gz``
            and ``<prefix>_report.json``.

    Returns:
        A tuple ``(mtr_image, report)`` where ``mtr_image`` is a
        :class:`qmri.io.NiftiImage` of the MTR map in percentage units (pu),
        co-located with the input, and ``report`` is an :class:`MTRReport`.

    Raises:
        ValueError: If a combined file is not 4D with two volumes, or the two
            separate images differ in shape or affine.
    """
    start_time = time.perf_counter()

    saturated_path = Path(saturated_file)

    if unsaturated_file is None:
        mode = "combined"
        input_files = [saturated_path]
        signal_nosat, signal_sat, reference = _load_combined(saturated_path)
    else:
        mode = "separate"
        unsaturated_path = Path(unsaturated_file)
        input_files = [unsaturated_path, saturated_path]
        signal_nosat, signal_sat, reference = _load_separate(
            saturated_path, unsaturated_path
        )

    mtr_map = calculate_mtr(signal_nosat, signal_sat).mtr

    mtr_image = NiftiImage(
        data=mtr_map,
        header=reference.header,
        affine=reference.affine,
    )

    valid = np.abs(signal_nosat) > 0
    valid_values = mtr_map[valid]
    if valid_values.size:
        mtr_mean = float(np.mean(valid_values))
        mtr_std = float(np.std(valid_values))
        mtr_min = float(np.min(valid_values))
        mtr_max = float(np.max(valid_values))
    else:
        mtr_mean = mtr_std = mtr_min = mtr_max = 0.0

    out_dir: Path | None = None
    prefix: str | None = None
    output_path: Path | None = None
    if save_outputs:
        out_dir = Path(output_dir) if output_dir is not None else saturated_path.parent
        out_dir.mkdir(parents=True, exist_ok=True)
        prefix = output_prefix or strip_nifti_suffix(saturated_path).stem
        output_path = out_dir / f"{prefix}_mtr_map.nii.gz"
        save_nifti(
            mtr_map,
            output_path,
            header=reference.header,
            affine=reference.affine,
        )

    report = MTRReport(
        input_files=input_files,
        mode=mode,
        output_file=output_path,
        n_valid_voxels=int(valid_values.size),
        mtr_mean=mtr_mean,
        mtr_std=mtr_std,
        mtr_min=mtr_min,
        mtr_max=mtr_max,
        processing_date=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        processing_time_seconds=time.perf_counter() - start_time,
    )

    if save_outputs and out_dir is not None and prefix is not None:
        report_path = out_dir / f"{prefix}_report.json"
        with open(report_path, "w", encoding="utf-8") as report_file:
            json.dump(report.to_dict(), report_file, indent=2)

    return mtr_image, report