qsi

tit.pre.qsi

QSI (QSIPrep / QSIRecon) integration for TI-Toolbox.

Provides Docker-out-of-Docker (DooD) integration for running QSIPrep and QSIRecon as sibling containers from within the SimNIBS container. The primary use case is preprocessing DWI data to extract DTI tensors for SimNIBS anisotropic conductivity simulations.

Public API

run_qsiprep Run the QSIPrep DWI preprocessing pipeline. run_qsirecon Run the QSIRecon DWI reconstruction pipeline. extract_dti_tensor Extract a DTI tensor and register it to the SimNIBS T1 grid. check_dti_tensor_exists Check whether an extracted DTI tensor already exists.

Configuration Classes

QSIPrepConfig Configuration dataclass for QSIPrep runs. QSIReconConfig Configuration dataclass for QSIRecon runs. ResourceConfig Resource allocation (CPU, memory, OMP threads) for QSI containers. ReconSpec Enum of available QSIRecon reconstruction specifications. QSIAtlas Enum of available atlases for connectivity analysis.

See Also

tit.pre : Parent preprocessing package. tit.pre.structural.run_pipeline : Full pipeline that can invoke QSI steps.

QSIPrepConfig dataclass

QSIPrepConfig(subject_id: str, output_resolution: float = QSI_DEFAULT_OUTPUT_RESOLUTION, resources: ResourceConfig = ResourceConfig(), image_tag: str = QSI_QSIPREP_IMAGE_TAG, skip_bids_validation: bool = True, denoise_method: str = 'dwidenoise', unringing_method: str = 'mrdegibbs', use_syn_sdc: bool = False, mni_normalization: bool = False)

Configuration for a QSIPrep run.

Attributes

subject_id : str Subject identifier (without 'sub-' prefix). output_resolution : float Target output resolution in mm (default: 2.0). resources : ResourceConfig Resource allocation settings. image_tag : str Docker image tag for QSIPrep. skip_bids_validation : bool Whether to skip BIDS validation (useful for non-BIDS datasets). denoise_method : str Denoising method: 'dwidenoise', 'patch2self', or 'none'. unringing_method : str Unringing method: 'mrdegibbs', 'rpg', or 'none' (already resolved; the pipeline setting 'auto' is resolved by run_qsiprep). use_syn_sdc : bool Pass --use-syn-sdc warn (fieldmap-less SyN distortion correction). Set by run_qsiprep when the subject has no usable fieldmap. mni_normalization : bool Run QSIPrep's anatomical normalization to MNI. Off by default: only QSIRecon atlases/template-space specs need it. Always on with SyN SDC, which uses the MNI transform to place its fieldmap prior.

QSIReconConfig dataclass

QSIReconConfig(subject_id: str, recon_specs: list[str] = (lambda: [QSI_DEFAULT_RECON_SPEC])(), atlases: list[str] | None = None, use_gpu: bool = False, resources: ResourceConfig = ResourceConfig(), image_tag: str = QSI_QSIRECON_IMAGE_TAG, skip_odf_reports: bool = True)

Configuration for a QSIRecon run.

Attributes

subject_id : str Subject identifier (without 'sub-' prefix). recon_specs : list[str] List of reconstruction specs to run. Defaults to ['dsi_studio_gqi'] (scalar maps only; the SimNIBS DTI tensor does not need QSIRecon). atlases : list[str] | None List of atlases for connectivity analysis. None (default) = no connectivity. Set to e.g. ['4S156Parcels', 'AAL116'] if needed. use_gpu : bool Whether to enable GPU acceleration (requires NVIDIA Docker runtime). resources : ResourceConfig Resource allocation settings. image_tag : str Docker image tag for QSIRecon. skip_odf_reports : bool Whether to skip ODF report generation (saves time).

ReconSpec

Bases: StrEnum

Available QSIRecon reconstruction specifications.

Each spec defines a complete reconstruction pipeline with specific algorithms and output formats.

from_string classmethod

from_string(value: str) -> Self

Convert string to ReconSpec enum.

Source code in tit/pre/qsi/config.py

@classmethod
def from_string(cls, value: str) -> Self:
    """Convert string to ReconSpec enum."""
    for spec in cls:
        if spec.value == value:
            return spec
    raise ValueError(f"Unknown recon spec: {value}")

list_all classmethod

list_all() -> list[str]

Return list of all spec values.

Source code in tit/pre/qsi/config.py

@classmethod
def list_all(cls) -> list[str]:
    """Return list of all spec values."""
    return [spec.value for spec in cls]

QSIAtlas

Bases: StrEnum

Available atlases for QSIRecon connectivity analysis.

These atlases can be used for structural connectivity matrix generation. The 4S series combines Schaefer cortical parcels with 56 subcortical ROIs.

from_string classmethod

from_string(value: str) -> Self

Convert string to QSIAtlas enum.

Source code in tit/pre/qsi/config.py

@classmethod
def from_string(cls, value: str) -> Self:
    """Convert string to QSIAtlas enum."""
    for atlas in cls:
        if atlas.value == value:
            return atlas
    raise ValueError(f"Unknown atlas: {value}")

list_all classmethod

list_all() -> list[str]

Return list of all atlas values.

Source code in tit/pre/qsi/config.py

@classmethod
def list_all(cls) -> list[str]:
    """Return list of all atlas values."""
    return [atlas.value for atlas in cls]

ResourceConfig dataclass

ResourceConfig(cpus: int | None = None, memory_gb: int | None = None, omp_threads: int = (lambda: QSI_DEFAULT_OMP_THREADS)())

Resource allocation configuration for QSI containers.

Attributes

cpus : int Number of CPUs to allocate to the container. memory_gb : int Memory limit in gigabytes. omp_threads : int Number of OpenMP threads (affects ANTS, MRtrix, etc.).

run_qsiprep

run_qsiprep(project_dir: str, subject_id: str, *, logger: Logger, output_resolution: float | None = None, cpus: int | None = None, memory_gb: int | None = None, omp_threads: int = QSI_DEFAULT_OMP_THREADS, image_tag: str = QSI_QSIPREP_IMAGE_TAG, skip_bids_validation: bool = True, denoise_method: str = 'dwidenoise', unringing_method: str = 'auto', mni_normalization: bool = False, runner: CommandRunner | None = None) -> None

Run QSIPrep preprocessing for a subject's DWI data.

This function spawns a QSIPrep Docker container as a sibling to the current SimNIBS container using Docker-out-of-Docker (DooD). QSIPrep needs an x86-64 Docker host (its SynthSeg step needs AVX); an arm64 host is refused before anything runs.

Susceptibility distortion correction is always on: TOPUP with the subject's reverse phase-encoding fieldmap (IntendedFor written into its sidecar when missing) or reverse-PE DWI series, else fieldmap-less SyN (--use-syn-sdc warn). DWI fieldmaps that exist but cannot be used stop the run instead of silently falling back.

Parameters

project_dir : str Path to the BIDS project root directory. subject_id : str Subject identifier (without 'sub-' prefix). logger : logging.Logger Logger for status messages. output_resolution : float or None, optional Isotropic output voxel size in mm. None (default) uses the native DWI voxel size (smallest axis, rounded to 0.1 mm). cpus : int, optional Number of CPUs to allocate. Default: 8. memory_gb : int, optional Memory limit in GB. Default: 32. omp_threads : int, optional Threads per process. Default: constants.QSI_DEFAULT_OMP_THREADS (min(cpu_count - 1, 8), matching QSIPrep's own default). image_tag : str, optional QSIPrep Docker image tag. Default from constants.QSI_QSIPREP_IMAGE_TAG. skip_bids_validation : bool, optional Skip BIDS validation. Default: True. denoise_method : str, optional Denoising method. Default: 'dwidenoise'. unringing_method : str, optional 'mrdegibbs', 'rpg', 'none' or 'auto' (default): rpg when the DWI sidecar has PartialFourier < 1, else mrdegibbs. mni_normalization : bool, optional Run the anatomical normalization to MNI (needed only for QSIRecon atlases and template-space specs). Default: False; forced on by SyN SDC. runner : CommandRunner | None, optional Command runner for subprocess execution.

Raises

PreprocessError If QSIPrep fails or prerequisites are not met.

Source code in tit/pre/qsi/qsiprep.py

def run_qsiprep(
    project_dir: str,
    subject_id: str,
    *,
    logger: logging.Logger,
    output_resolution: float | None = None,
    cpus: int | None = None,
    memory_gb: int | None = None,
    omp_threads: int = const.QSI_DEFAULT_OMP_THREADS,
    image_tag: str = const.QSI_QSIPREP_IMAGE_TAG,
    skip_bids_validation: bool = True,
    denoise_method: str = "dwidenoise",
    unringing_method: str = "auto",
    mni_normalization: bool = False,
    runner: CommandRunner | None = None,
) -> None:
    """
    Run QSIPrep preprocessing for a subject's DWI data.

    This function spawns a QSIPrep Docker container as a sibling to the
    current SimNIBS container using Docker-out-of-Docker (DooD). QSIPrep needs
    an x86-64 Docker host (its SynthSeg step needs AVX); an arm64 host is
    refused before anything runs.

    Susceptibility distortion correction is always on: TOPUP with the subject's
    reverse phase-encoding fieldmap (``IntendedFor`` written into its sidecar when
    missing) or reverse-PE DWI series, else fieldmap-less SyN
    (``--use-syn-sdc warn``). DWI fieldmaps that exist but cannot be used stop the
    run instead of silently falling back.

    Parameters
    ----------
    project_dir : str
        Path to the BIDS project root directory.
    subject_id : str
        Subject identifier (without 'sub-' prefix).
    logger : logging.Logger
        Logger for status messages.
    output_resolution : float or None, optional
        Isotropic output voxel size in mm. ``None`` (default) uses the native DWI
        voxel size (smallest axis, rounded to 0.1 mm).
    cpus : int, optional
        Number of CPUs to allocate. Default: 8.
    memory_gb : int, optional
        Memory limit in GB. Default: 32.
    omp_threads : int, optional
        Threads per process. Default: ``constants.QSI_DEFAULT_OMP_THREADS``
        (min(cpu_count - 1, 8), matching QSIPrep's own default).
    image_tag : str, optional
        QSIPrep Docker image tag. Default from ``constants.QSI_QSIPREP_IMAGE_TAG``.
    skip_bids_validation : bool, optional
        Skip BIDS validation. Default: True.
    denoise_method : str, optional
        Denoising method. Default: 'dwidenoise'.
    unringing_method : str, optional
        'mrdegibbs', 'rpg', 'none' or 'auto' (default): rpg when the DWI sidecar
        has ``PartialFourier`` < 1, else mrdegibbs.
    mni_normalization : bool, optional
        Run the anatomical normalization to MNI (needed only for QSIRecon atlases
        and template-space specs). Default: False; forced on by SyN SDC.
    runner : CommandRunner | None, optional
        Command runner for subprocess execution.

    Raises
    ------
    PreprocessError
        If QSIPrep fails or prerequisites are not met.
    """
    from tit.telemetry import track_operation
    from tit import constants as _const

    with track_operation(_const.TELEMETRY_OP_PRE_QSIPREP):
        logger.info(f"Starting QSIPrep for subject {subject_id}")
        ok, preflight_error = validate_dood_environment(
            project_dir, require_x86_64=True
        )
        if not ok:
            raise PreprocessError(f"QSI Docker preflight failed: {preflight_error}")

        # Validate DWI data exists. This runs before the container starts
        # because QSIPrep surfaces a bad gradient table or an incomplete
        # sidecar only after the anatomical workflow has finished, an hour in.
        is_valid, error_msg = validate_bids_dwi(project_dir, subject_id, logger)
        if not is_valid:
            raise PreprocessError(f"DWI validation failed: {error_msg}")

        is_valid, error_msg = ensure_total_readout_time(
            project_dir, subject_id, logger=logger
        )
        if not is_valid:
            raise PreprocessError(f"DWI sidecar validation failed: {error_msg}")

        pm = get_path_manager(project_dir)
        output_dir = Path(pm.qsiprep_subject(subject_id))
        # Docker `-v` can leave an empty directory behind; only a non-empty
        # one is a real output.
        if output_dir.exists() and any(output_dir.iterdir()):
            raise PreprocessError(
                f"QSIPrep output already exists at {output_dir}. "
                "Remove the directory manually before rerunning. The working "
                f"directory at {Path(pm.derivatives()) / '.qsiprep_work'} is kept "
                "on purpose: QSIPrep reuses the nodes that already finished, so "
                "a rerun after a failure skips the anatomical workflow. Delete "
                "it too only if you want to start from scratch."
            )

        sdc = plan_distortion_correction(project_dir, subject_id, logger=logger)
        if sdc.blocking_error:
            raise PreprocessError(sdc.blocking_error)
        logger.info(f"Distortion correction: {sdc.describe()}")
        if sdc.use_syn and not mni_normalization:
            logger.info(
                "MNI normalization enabled: SyN distortion correction needs the "
                "anat-to-MNI transform."
            )

        if unringing_method == "auto":
            unringing_method, reason = choose_unringing_method(project_dir, subject_id)
            logger.info(f"Unringing: {unringing_method} ({reason})")

        if output_resolution is None:
            output_resolution = native_dwi_resolution(project_dir, subject_id)
            if output_resolution is None:
                output_resolution = const.QSI_DEFAULT_OUTPUT_RESOLUTION
                logger.warning(
                    "Could not read the DWI voxel size; using "
                    f"{output_resolution:g} mm output resolution."
                )
            else:
                logger.info(
                    f"Output resolution: {output_resolution:g} mm (native DWI voxel size)"
                )

        # Create output directories
        output_dir.parent.mkdir(parents=True, exist_ok=True)
        work_dir = Path(pm.derivatives()) / ".qsiprep_work"
        work_dir.mkdir(parents=True, exist_ok=True)

        # Build configuration
        config = QSIPrepConfig(
            subject_id=subject_id,
            output_resolution=output_resolution,
            resources=ResourceConfig(
                cpus=cpus,
                memory_gb=memory_gb,
                omp_threads=omp_threads,
            ),
            image_tag=image_tag,
            skip_bids_validation=skip_bids_validation,
            denoise_method=denoise_method,
            unringing_method=unringing_method,
            use_syn_sdc=sdc.use_syn,
            mni_normalization=mni_normalization,
        )

        try:
            # Build Docker command
            builder = DockerCommandBuilder(project_dir)
            cmd = builder.build_qsiprep_cmd(config)
        except DockerBuildError as e:
            raise PreprocessError(f"Failed to build QSIPrep command: {e}")

        # Ensure image is available
        if not pull_image_if_needed(const.QSI_QSIPREP_IMAGE, image_tag, logger):
            raise PreprocessError(
                f"Failed to pull QSIPrep image: {const.QSI_QSIPREP_IMAGE}:{image_tag}"
            )

        # Log the command for debugging
        logger.debug(f"QSIPrep command: {' '.join(cmd)}")

        # Run the container
        if runner is None:
            runner = CommandRunner()

        logger.info(f"Running QSIPrep for subject {subject_id}...")
        returncode = runner.run(cmd, logger=logger)

        if returncode != 0:
            crash_count = _report_crashfiles(output_dir, logger)
            raise PreprocessError(
                _format_qsiprep_failure(returncode, runner, crash_count)
            )

        # Validate output
        is_valid, error_msg = validate_qsiprep_output(project_dir, subject_id)
        if not is_valid:
            raise PreprocessError(f"QSIPrep output validation failed: {error_msg}")

    logger.info(f"QSIPrep completed successfully for subject {subject_id}")

run_qsirecon

run_qsirecon(project_dir: str, subject_id: str, *, logger: Logger, recon_specs: list[str] | None = None, atlases: list[str] | None = None, use_gpu: bool = False, cpus: int | None = None, memory_gb: int | None = None, omp_threads: int = QSI_DEFAULT_OMP_THREADS, image_tag: str = QSI_QSIRECON_IMAGE_TAG, skip_odf_reports: bool = True, runner: CommandRunner | None = None) -> None

Run QSIRecon reconstruction for a subject's preprocessed DWI data.

This function spawns QSIRecon Docker containers as siblings to the current SimNIBS container using Docker-out-of-Docker (DooD).

QSIRecon is optional: it adds tractography, scalar maps and connectivity on top of QSIPrep output. The SimNIBS DTI tensor does not need it. Multiple reconstruction specs can be run sequentially.

Parameters

project_dir : str Path to the BIDS project root directory. subject_id : str Subject identifier (without 'sub-' prefix). logger : logging.Logger Logger for status messages. recon_specs : list[str] | None, optional List of reconstruction specifications to run. Default: ['dsi_studio_gqi'] (GQI scalar maps, no tractography). Other specs (mrtrix_, dipy_, amico_noddi, pyafq_*, etc.) remain available. atlases : list[str] | None, optional List of atlases for connectivity analysis. Default: None (no connectivity). Needs a QSIPrep run with MNI normalization enabled. use_gpu : bool, optional Enable GPU acceleration. Default: False. cpus : int | None, optional Number of CPUs to allocate. None = inherit from current container. memory_gb : int | None, optional Memory limit in GB. None = inherit from current container. omp_threads : int, optional Threads per process. Default: constants.QSI_DEFAULT_OMP_THREADS (min(cpu_count - 1, 8), matching QSIPrep's own default). image_tag : str, optional QSIRecon Docker image tag. Default from constants.QSI_QSIRECON_IMAGE_TAG. skip_odf_reports : bool, optional Skip ODF report generation. Default: True. runner : CommandRunner | None, optional Command runner for subprocess execution.

Raises

PreprocessError If QSIRecon fails or prerequisites are not met.

Source code in tit/pre/qsi/qsirecon.py

def run_qsirecon(
    project_dir: str,
    subject_id: str,
    *,
    logger: logging.Logger,
    recon_specs: list[str] | None = None,
    atlases: list[str] | None = None,
    use_gpu: bool = False,
    cpus: int | None = None,
    memory_gb: int | None = None,
    omp_threads: int = const.QSI_DEFAULT_OMP_THREADS,
    image_tag: str = const.QSI_QSIRECON_IMAGE_TAG,
    skip_odf_reports: bool = True,
    runner: CommandRunner | None = None,
) -> None:
    """
    Run QSIRecon reconstruction for a subject's preprocessed DWI data.

    This function spawns QSIRecon Docker containers as siblings to the
    current SimNIBS container using Docker-out-of-Docker (DooD).

    QSIRecon is optional: it adds tractography, scalar maps and connectivity on
    top of QSIPrep output. The SimNIBS DTI tensor does not need it. Multiple
    reconstruction specs can be run sequentially.

    Parameters
    ----------
    project_dir : str
        Path to the BIDS project root directory.
    subject_id : str
        Subject identifier (without 'sub-' prefix).
    logger : logging.Logger
        Logger for status messages.
    recon_specs : list[str] | None, optional
        List of reconstruction specifications to run. Default: ['dsi_studio_gqi']
        (GQI scalar maps, no tractography). Other specs (mrtrix_*, dipy_*,
        amico_noddi, pyafq_*, etc.) remain available.
    atlases : list[str] | None, optional
        List of atlases for connectivity analysis. Default: None (no
        connectivity). Needs a QSIPrep run with MNI normalization enabled.
    use_gpu : bool, optional
        Enable GPU acceleration. Default: False.
    cpus : int | None, optional
        Number of CPUs to allocate. None = inherit from current container.
    memory_gb : int | None, optional
        Memory limit in GB. None = inherit from current container.
    omp_threads : int, optional
        Threads per process. Default: ``constants.QSI_DEFAULT_OMP_THREADS``
        (min(cpu_count - 1, 8), matching QSIPrep's own default).
    image_tag : str, optional
        QSIRecon Docker image tag. Default from ``constants.QSI_QSIRECON_IMAGE_TAG``.
    skip_odf_reports : bool, optional
        Skip ODF report generation. Default: True.
    runner : CommandRunner | None, optional
        Command runner for subprocess execution.

    Raises
    ------
    PreprocessError
        If QSIRecon fails or prerequisites are not met.
    """
    if recon_specs is None:
        recon_specs = [const.QSI_DEFAULT_RECON_SPEC]

    from tit.telemetry import track_operation
    from tit import constants as _const

    with track_operation(_const.TELEMETRY_OP_PRE_QSIRECON):
        logger.info(
            f"Starting QSIRecon for subject {subject_id} with specs: {recon_specs}, atlases: {atlases}"
        )
        ok, preflight_error = validate_dood_environment(
            project_dir, require_gpu=use_gpu
        )
        if not ok:
            raise PreprocessError(f"QSI Docker preflight failed: {preflight_error}")

        # Validate QSIPrep output exists
        is_valid, error_msg = validate_qsiprep_output(project_dir, subject_id)
        if not is_valid:
            raise PreprocessError(
                f"QSIPrep output validation failed: {error_msg}. "
                "Run QSIPrep first before running QSIRecon."
            )

        # Atlases are warped from MNI; QSIPrep only writes that transform when its
        # (opt-in) MNI normalization ran.
        qsiprep_anat = Path(get_path_manager(project_dir).qsiprep_subject(subject_id))
        if atlases and not any(
            qsiprep_anat.glob("**/anat/*_to-MNI152NLin2009cAsym_*xfm.h5")
        ):
            raise PreprocessError(
                "QSIRecon atlases need QSIPrep's MNI normalization, which this "
                "subject's QSIPrep run skipped. Re-run QSIPrep with MNI "
                "normalization enabled, or run QSIRecon without atlases."
            )

        # No mkdir here — Docker's `-v` creates host directories automatically.
        # Creating them from SimNIBS fails on Docker Desktop due to phantom
        # bind-mount entries left by previous sibling containers.
        # QSIRecon 26 writes its reconstructions under
        # <out>/derivatives/qsirecon-<suffix>/sub-<id>/ and reserves
        # <out>/sub-<id>/ for logs and figures. Guarding on the latter would
        # block every rerun after the first attempt, including one that
        # crashed and produced nothing but a log.
        recon_root = Path(get_path_manager(project_dir).qsirecon()) / "derivatives"
        existing = [
            candidate
            for candidate in recon_root.glob(f"qsirecon-*/sub-{subject_id}")
            if any(candidate.iterdir())
        ]
        if existing:
            raise PreprocessError(
                "QSIRecon output already exists at "
                f"{', '.join(str(path) for path in sorted(existing))}. "
                "Remove those directories manually before rerunning."
            )

        # Build configuration
        config = QSIReconConfig(
            subject_id=subject_id,
            recon_specs=recon_specs,
            atlases=atlases,
            use_gpu=use_gpu,
            resources=ResourceConfig(
                cpus=cpus,
                memory_gb=memory_gb,
                omp_threads=omp_threads,
            ),
            image_tag=image_tag,
            skip_odf_reports=skip_odf_reports,
        )

        try:
            # Build Docker command builder
            builder = DockerCommandBuilder(project_dir)
        except DockerBuildError as e:
            raise PreprocessError(f"Failed to initialize Docker: {e}")

        # Ensure image is available
        if not pull_image_if_needed(const.QSI_QSIRECON_IMAGE, image_tag, logger):
            raise PreprocessError(
                f"Failed to pull QSIRecon image: {const.QSI_QSIRECON_IMAGE}:{image_tag}"
            )

        # Create runner if not provided
        if runner is None:
            runner = CommandRunner()

        # Run each recon spec
        for spec in recon_specs:
            logger.info(f"Running QSIRecon spec: {spec}")

            try:
                cmd = builder.build_qsirecon_cmd(config, spec)
            except DockerBuildError as e:
                raise PreprocessError(f"Failed to build QSIRecon command: {e}")

            # Log the command for debugging
            logger.debug(f"QSIRecon command: {' '.join(cmd)}")

            # Run the container
            logger.info(f"Running QSIRecon {spec} for subject {subject_id}...")
            returncode = runner.run(cmd, logger=logger)

            if returncode != 0:
                raise PreprocessError(
                    f"QSIRecon {spec} failed with exit code {returncode}"
                )

            logger.info(f"QSIRecon {spec} completed for subject {subject_id}")

        logger.info(f"QSIRecon completed successfully for subject {subject_id}")

extract_dti_tensor

extract_dti_tensor(project_dir: str, subject_id: str, *, logger: Logger) -> Path

Fit, register and write DTI_coregT1_tensor.nii.gz for subject_id.

Reads QSIPrep output and the charm m2m folder; writes the tensor and DTI_coregT1_qc.json into the m2m folder, then a DTI QC report.

Raises

tit.pre.utils.PreprocessError Missing inputs, an existing tensor, an m2m T1 that is not the T1w QSIPrep used, or a failed QC gate (the QC JSON and the report are still written; the tensor is not).

Source code in tit/pre/qsi/dti_extractor.py

def extract_dti_tensor(
    project_dir: str,
    subject_id: str,
    *,
    logger: logging.Logger,
) -> Path:
    """Fit, register and write ``DTI_coregT1_tensor.nii.gz`` for *subject_id*.

    Reads QSIPrep output and the charm m2m folder; writes the tensor and
    ``DTI_coregT1_qc.json`` into the m2m folder, then a DTI QC report.

    Raises
    ------
    tit.pre.utils.PreprocessError
        Missing inputs, an existing tensor, an m2m T1 that is not the T1w QSIPrep
        used, or a failed QC gate (the QC JSON and the report are still written; the
        tensor is not).
    """
    import nibabel as nib
    from scipy.ndimage import binary_dilation

    pm = get_path_manager(project_dir)
    m2m_dir = Path(pm.m2m(subject_id))
    output_path = m2m_dir / const.FILE_DTI_TENSOR
    qc_path = m2m_dir / const.FILE_DTI_QC
    t1_path = m2m_dir / const.FILE_T1
    labels_path = m2m_dir / "final_tissues.nii.gz"
    for path, fix in ((t1_path, "Run charm first."), (labels_path, "Run charm first.")):
        if not path.is_file():
            raise PreprocessError(f"{path} not found. {fix}")
    if output_path.exists():
        raise PreprocessError(
            f"DTI tensor already exists at {output_path}. Remove it before rerunning."
        )
    raw_t1_path = _find_nifti(Path(pm.bids_anat(subject_id)), f"sub-{subject_id}_T1w")
    if raw_t1_path is None:
        raise PreprocessError(
            f"sub-{subject_id}_T1w not found in {pm.bids_anat(subject_id)}; the "
            "transform chain starts from the T1w QSIPrep used."
        )
    inputs = qsiprep_inputs(Path(pm.qsiprep_subject(subject_id)))

    t1 = nib.load(str(t1_path))
    raw = nib.load(str(raw_t1_path))
    if raw.shape[:3] != t1.shape[:3] or not np.allclose(
        raw.affine, t1.affine, atol=1e-3
    ):
        raise PreprocessError(
            f"The m2m T1 grid ({t1_path}) differs from the raw T1w QSIPrep used "
            f"({raw_t1_path}). The QSIPrep transform chain is only valid when charm "
            "ran on that same T1w file."
        )
    labels = np.asanyarray(nib.load(str(labels_path)).dataobj)
    labels = labels[..., 0] if labels.ndim == 4 else labels
    if labels.shape != t1.shape[:3]:
        raise PreprocessError(f"{labels_path} is not on the T1 grid {t1.shape[:3]}")

    logger.info(f"DTI: WLS tensor fit (b <= {DTI_BMAX:g}) on {inputs['dwi'].name}")
    tensors_acpc, acpc_affine, fit_mask = fit_tensor(
        inputs["dwi"], inputs["grad"], inputs["dwi_mask"]
    )

    chain = tm.acpc_to_t1_world(
        raw.affine, raw.shape[:3], read_itk_transform(inputs["xfm"])
    )
    acpc_t1 = nib.load(str(inputs["acpc_t1"]))
    acpc_mask = np.asanyarray(nib.load(str(inputs["acpc_mask"])).dataobj) > 0
    logger.info("DTI: checking the QSIPrep ACPC -> T1 chain against an NCC refinement")
    ncc, disp = check_registration(
        chain,
        np.asanyarray(acpc_t1.dataobj, dtype=np.float32),
        acpc_t1.affine,
        acpc_mask,
        np.asanyarray(t1.dataobj, dtype=np.float32),
        t1.affine,
    )

    logger.info("DTI: resampling onto the m2m T1 grid with tensor rotation")
    tensors, valid = resample_tensor(
        tensors_acpc, acpc_affine, fit_mask, chain, t1.shape[:3], t1.affine
    )
    del tensors_acpc
    wmgm = np.isin(labels, (1, 2))
    mask = (
        valid
        & binary_dilation(wmgm, iterations=MASK_DILATE_VOX)
        & np.isin(labels, BRAIN_LABELS)
    )
    tensors[~mask] = 0

    eigenvalues, _ = tm.eig_desc(tensors[mask])
    fa, md = tm.fa_md(eigenvalues)
    wm_in_mask = (labels == 1)[mask]
    qc = DtiQc(
        ncc_chain=float(ncc),
        chain_vs_ncc_mm=disp,
        pct_pd=(
            float(100 * (eigenvalues[:, -1] > 0).mean()) if len(eigenvalues) else 0.0
        ),
        pct_wm_covered=float(100 * mask[labels == 1].mean()),
        pct_gm_covered=float(100 * mask[labels == 2].mean()),
        pct_wmgm_zero=float(100 * (~mask[wmgm]).mean()),
        wm_md_median=float(np.median(md[wm_in_mask])) if wm_in_mask.any() else 0.0,
        wm_fa_median=float(np.median(fa[wm_in_mask])) if wm_in_mask.any() else 0.0,
        n_out_of_brain=int((mask & ~np.isin(labels, BRAIN_LABELS)).sum()),
        acpc_to_t1=chain.round(6).tolist(),
    )
    qc.gate()
    logger.info(
        f"DTI QC: {json.dumps({k: v for k, v in asdict(qc).items() if k not in ('acpc_to_t1', 'thresholds')})}"
    )
    stored = None
    if qc.passed:
        stored = tm.world_to_simnibs(tensors, t1.affine).astype(np.float32)
        stored[~mask] = 0
    t6w = np.zeros(t1.shape[:3] + (6,), np.float32)
    t6w[mask] = tm.unsym(tensors[mask])
    del tensors
    if stored is not None:
        _save_nifti_gz(stored, t1.affine, output_path)
        logger.info(f"DTI tensor saved to: {output_path}")
        del stored

    from .dti_advisories import DtiVolumes

    warp_path = m2m_dir / "toMNI" / "Conform2MNI_nonl.nii.gz"
    vols = DtiVolumes.from_arrays(
        t1.affine,
        np.asanyarray(t1.dataobj, dtype=np.float32),
        labels,
        t6w,
        (
            np.asanyarray(nib.load(str(warp_path)).dataobj)
            if warp_path.is_file()
            else None
        ),
    )
    del t6w
    record = asdict(qc)
    logger.info(
        "DTI: measuring advisories (orientation, flip test, residual shift, conductivity)"
    )
    record_advisories(
        record,
        vols,
        project_dir,
        subject_id,
        recorded_by="extract_dti_tensor",
        logger=logger,
    )
    qc_path.write_text(json.dumps(record, indent=1) + "\n")

    from tit.reporting.generators.dti_qc import create_dti_qc_report

    try:
        report = create_dti_qc_report(project_dir, subject_id, record, vols)
        logger.info(f"DTI QC report: {report}")
    except Exception as exc:  # the tensor and QC record stand without their report
        logger.error(f"DTI QC report could not be written: {exc}")
    if not qc.passed:
        raise PreprocessError(
            "DTI QC gate failed ("
            + ", ".join(qc.failures)
            + f"); see {qc_path} and the DTI QC report. "
            "The tensor was not written. Check QSIPrep's report for this subject, and "
            "that charm ran on the same T1w QSIPrep used."
        )
    return output_path

check_dti_tensor_exists

check_dti_tensor_exists(project_dir: str, subject_id: str) -> bool

Whether DTI_coregT1_tensor.nii.gz exists in subject_id's m2m folder.

Source code in tit/pre/qsi/dti_extractor.py

def check_dti_tensor_exists(project_dir: str, subject_id: str) -> bool:
    """Whether ``DTI_coregT1_tensor.nii.gz`` exists in *subject_id*'s m2m folder."""
    m2m_dir = get_path_manager(project_dir).m2m(subject_id)
    return os.path.isdir(m2m_dir) and (Path(m2m_dir) / const.FILE_DTI_TENSOR).exists()