pre

tit.pre ¶

Preprocessing APIs for TI-Toolbox.

This package provides standalone, reusable functions for every stage of the TI-Toolbox preprocessing pipeline: DICOM conversion, SimNIBS head-mesh generation (CHARM), FastSurfer deep segmentation, tissue analysis, and diffusion-weighted imaging (DWI) preprocessing via QSIPrep/QSIRecon.

Structural work runs in the TI-Toolbox image itself (SimNIBS + FastSurfer); only the DWI stages spawn sibling containers (QSIPrep/QSIRecon). The functions here orchestrate those and manage BIDS-compliant file layouts.

Public API¶

run_pipeline Full preprocessing pipeline orchestrator (sequential or parallel). run_dicom_to_nifti Convert DICOM series to BIDS-compliant NIfTI files. run_fastsurfer Run FastSurfer --seg_only deep segmentation. fastsurfer_available Probe for a runnable FastSurfer install. run_charm Generate a SimNIBS head mesh via the charm command. run_tissue_analysis Compute tissue volumes and thickness from segmented NIfTI data. run_qsiprep Preprocess DWI data using QSIPrep (Docker-out-of-Docker). run_qsirecon Reconstruct DWI data using QSIRecon (Docker-out-of-Docker). extract_dti_tensor Extract a DTI tensor for SimNIBS anisotropic conductivity. discover_subjects Discover subject IDs present in a BIDS project tree. check_m2m_exists Check whether a SimNIBS m2m directory already exists.

PreprocessingInputProblem `dataclass` ¶

PreprocessingInputProblem(subject_id: str, step: str, label: str, message: str, path: Path)

Missing or invalid input for one selected preprocessing step.

PreprocessingOutput `dataclass` ¶

PreprocessingOutput(subject_id: str, step: str, label: str, path: Path, cleanup_paths: tuple[Path, ...] = ())

Existing output for one preprocessing step.

run_pipeline ¶

run_pipeline(subject_ids: Iterable[str], *, convert_dicom: bool = False, run_fastsurfer: bool = False, charm_threads: int | None = None, charm_options: dict | None = None, fastsurfer_threads: int | None = None, run_freesurfer: bool = False, freesurfer_recon_all: bool = True, freesurfer_subregions: list[str] | None = None, freesurfer_threads: int | None = None, create_m2m: bool = False, run_tissue_analysis: bool = False, run_qsiprep: bool = False, run_qsirecon: bool = False, qsiprep_config: dict | None = None, qsi_recon_config: dict | None = None, extract_dti: bool = False, skip_existing_outputs: bool = False, replace_existing_outputs: bool = False, stop_event: object | None = None, logger_callback: Callable | None = None, runner: CommandRunner | None = None) -> int

Run the preprocessing pipeline for one or more subjects.

Orchestrates DICOM conversion, SimNIBS CHARM, FastSurfer deep segmentation, tissue analysis, QSIPrep/QSIRecon DWI preprocessing and DTI tensor extraction. Steps are enabled via boolean flags; disabled steps are skipped.

All step flags default to False; pass keyword arguments only.

Parameters¶

subject_ids : iterable of str Subject identifiers without the sub- prefix (e.g. ["ernie", "101"]). Subjects run sequentially. convert_dicom : bool, optional Run DICOM-to-NIfTI conversion (sourcedata/sub-<id>/ to sub-<id>/anat/). run_fastsurfer : bool, optional Run FastSurfer --seg_only deep segmentation. charm_threads : int or None, optional Thread count for SimNIBS charm; None uses its default. charm_options : dict or None, optional charm overrides: {"denoise": bool, "segmentation_final_resolution": 0.5-2.0, "skin_facet_size": 0.5-10.0} (keys optional; unknown keys raise ValueError). None keeps the installed defaults. fastsurfer_threads : int or None, optional Thread count for FastSurfer inference. run_freesurfer : bool, optional Run FreeSurfer (requires a FreeSurfer install, which the standard image does not ship). freesurfer_recon_all : bool, optional With run_freesurfer, run recon-all (default True); set False to run only freesurfer_subregions on an existing reconstruction. freesurfer_subregions : list of str or None, optional FreeSurfer subregion segmentations to add: any of "thalamus", "hippo-amygdala". freesurfer_threads : int or None, optional Thread count for FreeSurfer. create_m2m : bool, optional Run SimNIBS charm to build m2m_<id> (also runs subject_atlas for .annot files). run_tissue_analysis : bool, optional Run tissue-volume and thickness analysis. run_qsiprep : bool, optional Run QSIPrep DWI preprocessing via Docker. run_qsirecon : bool, optional Run QSIRecon reconstruction via Docker. qsiprep_config : dict or None, optional Extra configuration passed to run_qsiprep. qsi_recon_config : dict or None, optional Extra configuration passed to run_qsirecon. extract_dti : bool, optional Extract DTI tensor for SimNIBS anisotropic conductivity. skip_existing_outputs : bool, optional Skip selected preprocessing steps when their output already exists. replace_existing_outputs : bool, optional Remove selected existing outputs before rerunning their steps. stop_event : object or None, optional Threading event used to cancel running steps. logger_callback : callable or None, optional Callback used by the GUI to capture log lines. runner : CommandRunner or None, optional Subprocess runner used to stream command output.

Returns¶

int 0 on success, 1 on failure.

Raises¶

PreprocessError If no subjects are provided, both skip_existing_outputs and replace_existing_outputs are set, a required input is missing for a selected step (checked for every subject before anything runs), or a preprocessing step fails. PreprocessCancelled If stop_event is set during execution.

Examples¶

from tit.pre import run_pipeline run_pipeline(["ernie"], convert_dicom=True, create_m2m=True) # doctest: +SKIP 0 run_pipeline(["ernie", "101"], create_m2m=True, ... charm_options={"denoise": True}, ... skip_existing_outputs=True) # doctest: +SKIP 0

run_dicom_to_nifti ¶

run_dicom_to_nifti(project_dir: str, subject_id: str, *, logger, runner: CommandRunner | None = None) -> None

Ingest a subject's source images into BIDS-named NIfTI files.

Looks for a T1w, T2w, ct, and dwi folder under sourcedata/sub-{subject_id}/ (folder names are matched case-insensitively) and ingests each one that exists. Anatomical images and CT go to the subject's anat/ folder, diffusion images to dwi/ (with their .bval/.bvec sidecars).

Each modality folder is searched recursively. Supported archives (.zip, .tar, .tar.gz, .tgz) are safely extracted to extracted_archives/ first. DICOM files (.dcm/.dicom) are then converted with dcm2niix; if the folder holds no DICOMs but does hold a NIfTI, that file is copied into place instead (compressing a bare .nii on the way). Dotfiles are ignored throughout.

CT is written as anat/sub-{id}_ct.nii.gz. This is a local convention, not BIDS -- see :data:MODALITIES.

Parameters¶

project_dir : str BIDS project root directory. subject_id : str Subject identifier without the sub- prefix. logger : logging.Logger Logger for progress messages. runner : CommandRunner or None, optional Subprocess runner for streaming output.

Raises¶

PreprocessError If an output NIfTI already exists for a modality.

fastsurfer_available ¶

fastsurfer_available() -> bool

Return True when a runnable FastSurfer checkout is present.

Capability-probe shape (same contract as the server's other capability probes): filesystem-only, no subprocess, safe to call on any platform.

Source code in tit/pre/fastsurfer.py

def fastsurfer_available() -> bool:
    """Return ``True`` when a runnable FastSurfer checkout is present.

    Capability-probe shape (same contract as the server's other capability
    probes): filesystem-only, no subprocess, safe to call on any platform.
    """
    return fastsurfer_script() is not None

run_fastsurfer ¶

run_fastsurfer(project_dir: str, subject_id: str, *, logger, runner: CommandRunner | None = None, threads: int | None = None) -> None

Run FastSurfer --seg_only for one subject.

Writes derivatives/fastsurfer/sub-<id>/mri/aparc.DKTatlas+aseg.deep.mgz plus a .nii.gz copy and a *_labels.txt sidecar, then returns. Idempotent: an existing segmentation is left alone (only the derived NIfTI/labels are backfilled if missing).

The input is the raw BIDS T1w, the same image recon-all was given. Two reasons, both load-bearing: it keeps this stage independent of charm so the two can run in parallel after DICOM import (the DAG in :mod:tit.jobs.plans relies on that), and it is the image the spike's Dice numbers were measured on -- m2m_<id>/T1.nii.gz is charm's own bias-corrected, re-conformed volume, a different input with unmeasured effect on the network's output.

Parameters¶

project_dir : str BIDS project root. subject_id : str Subject identifier without the sub- prefix. logger : logging.Logger Logger used for progress and streamed command output. runner : CommandRunner or None, optional Subprocess runner used to stream output and honour cancellation. threads : int or None, optional Thread count for the inference. Defaults to $TIT_FASTSURFER_THREADS, else the user-wide preference (automatic: the global CPU limit).

Raises¶

PreprocessError If FastSurfer is not installed, no T1w is found, run_fastsurfer.sh exits non-zero, or the expected segmentation is missing afterwards.

run_charm ¶

run_charm(project_dir: str, subject_id: str, *, logger, runner: CommandRunner | None = None, threads: int | None = None, options: dict | None = None) -> None

Run SimNIBS charm to generate a head mesh for a subject.

Creates an m2m directory at the standard BIDS derivatives location containing the volumetric head model required for TI simulations.

Parameters¶

project_dir : str BIDS project root. subject_id : str Subject identifier without the sub- prefix. logger : logging.Logger Logger used for progress and command output. runner : CommandRunner or None, optional Subprocess runner used to stream output.

Raises¶

PreprocessError If no T1 image is found, the m2m directory already exists, or charm exits with a non-zero code.

run_tissue_analysis ¶

run_tissue_analysis(project_dir: str, subject_id: str, *, tissues: Iterable[str] = DEFAULT_TISSUES, logger: Logger) -> dict

Run tissue analysis for a subject.

Creates a TissueAnalyzer for each requested tissue type, computes volume and thickness statistics, and writes reports and visualizations.

Parameters¶

project_dir : str BIDS project root. subject_id : str Subject identifier without the sub- prefix. tissues : iterable of str, optional Tissue types to analyze. Defaults to ('bone', 'csf', 'skin'). logger : logging.Logger Logger for progress output.

Returns¶

dict Mapping of tissue name to per-tissue analysis results (volume, thickness, voxel counts, report path).

Raises¶

PreprocessError If the segmentation labeling file does not exist.

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

discover_subjects ¶

discover_subjects(project_dir: str | None) -> list[str]

Return sorted, deduplicated subject IDs found in a BIDS project tree.

Returns an empty list when project_dir is None (project not configured).

Discovery order:

sourcedata/sub-*/T1w/ or T2w/ -- any subdir, NIfTI, DICOM, or supported DICOM archive (.zip, .tar, .tar.gz, .tgz).
sourcedata/sub-*/*.tgz (compressed bundles at top level).
sub-*/anat/*T1w*.nii[.gz] or *T2w*.nii[.gz] at project root.

Parameters¶

project_dir : str BIDS project root directory.

Returns¶

list[str] Sorted list of subject identifiers (without the sub- prefix).

check_m2m_exists ¶

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

Return True if the SimNIBS m2m directory already exists.

Checks for <project_dir>/derivatives/SimNIBS/sub-<subject_id>/m2m_<subject_id>.

Parameters¶

project_dir : str BIDS project root directory. subject_id : str Subject identifier without the sub- prefix.

Returns¶

bool True if the m2m directory exists on disk.

find_existing_preprocessing_outputs ¶

find_existing_preprocessing_outputs(project_dir: str, subject_ids: Iterable[str], *, steps: Sequence[str] | None = None, convert_dicom: bool = False, create_m2m: bool = False, run_fastsurfer: bool = False, run_freesurfer: bool = False, freesurfer_recon_all: bool = True, freesurfer_subregions: Sequence[str] = (), run_qsiprep: bool = False, run_qsirecon: bool = False, extract_dti: bool = False) -> list[PreprocessingOutput]

Return outputs that already exist for selected subjects and steps.

Source code in tit/pre/preflight.py

def find_existing_preprocessing_outputs(
    project_dir: str,
    subject_ids: Iterable[str],
    *,
    steps: Sequence[str] | None = None,
    convert_dicom: bool = False,
    create_m2m: bool = False,
    run_fastsurfer: bool = False,
    run_freesurfer: bool = False,
    freesurfer_recon_all: bool = True,
    freesurfer_subregions: Sequence[str] = (),
    run_qsiprep: bool = False,
    run_qsirecon: bool = False,
    extract_dti: bool = False,
) -> list[PreprocessingOutput]:
    """Return outputs that already exist for selected subjects and steps."""
    selected_steps = (
        list(steps)
        if steps is not None
        else selected_preprocessing_steps(
            convert_dicom=convert_dicom,
            create_m2m=create_m2m,
            run_fastsurfer=run_fastsurfer,
            run_freesurfer=run_freesurfer,
            freesurfer_recon_all=freesurfer_recon_all,
            freesurfer_subregions=freesurfer_subregions,
            run_qsiprep=run_qsiprep,
            run_qsirecon=run_qsirecon,
            extract_dti=extract_dti,
        )
    )
    outputs: list[PreprocessingOutput] = []
    for subject_id in subject_ids:
        for step in selected_steps:
            outputs.extend(existing_outputs_for_step(project_dir, subject_id, step))
    return outputs

find_missing_preprocessing_inputs ¶

find_missing_preprocessing_inputs(project_dir: str, subject_ids: Iterable[str], *, steps: Sequence[str] | None = None, convert_dicom: bool = False, create_m2m: bool = False, run_fastsurfer: bool = False, run_freesurfer: bool = False, freesurfer_recon_all: bool = True, freesurfer_subregions: Sequence[str] = (), run_qsiprep: bool = False, run_qsirecon: bool = False, extract_dti: bool = False, skip_existing_outputs: bool = False) -> list[PreprocessingInputProblem]

Return missing inputs that can be detected before running subprocesses.

Source code in tit/pre/preflight.py

def find_missing_preprocessing_inputs(
    project_dir: str,
    subject_ids: Iterable[str],
    *,
    steps: Sequence[str] | None = None,
    convert_dicom: bool = False,
    create_m2m: bool = False,
    run_fastsurfer: bool = False,
    run_freesurfer: bool = False,
    freesurfer_recon_all: bool = True,
    freesurfer_subregions: Sequence[str] = (),
    run_qsiprep: bool = False,
    run_qsirecon: bool = False,
    extract_dti: bool = False,
    skip_existing_outputs: bool = False,
) -> list[PreprocessingInputProblem]:
    """Return missing inputs that can be detected before running subprocesses."""
    subject_ids = list(subject_ids)
    selected_steps = (
        list(steps)
        if steps is not None
        else selected_preprocessing_steps(
            convert_dicom=convert_dicom,
            create_m2m=create_m2m,
            run_fastsurfer=run_fastsurfer,
            run_freesurfer=run_freesurfer,
            freesurfer_recon_all=freesurfer_recon_all,
            freesurfer_subregions=freesurfer_subregions,
            run_qsiprep=run_qsiprep,
            run_qsirecon=run_qsirecon,
            extract_dti=extract_dti,
        )
    )

    problems: list[PreprocessingInputProblem] = []
    for subject_id in subject_ids:
        subject_steps = selected_steps
        if _dicom_conversion_will_run(
            project_dir,
            subject_id,
            convert_dicom=convert_dicom,
            skip_existing_outputs=skip_existing_outputs,
        ):
            # Conversion can produce the T1w and DWI inputs these steps need,
            # so don't flag them as missing yet.
            subject_steps = [
                step
                for step in selected_steps
                if step
                not in (STEP_CHARM, STEP_FASTSURFER, STEP_FREESURFER, STEP_QSIPREP)
            ]
        if STEP_FREESURFER in selected_steps and (
            not skip_existing_outputs
            or not existing_outputs_for_step(project_dir, subject_id, STEP_FREESURFER)
        ):
            # This job's recon-all supplies the subregion prerequisites.
            subject_steps = [
                s for s in subject_steps if s not in FREESURFER_SUBREGION_STEPS.values()
            ]
        if skip_existing_outputs and existing_outputs_for_step(
            project_dir, subject_id, STEP_FREESURFER
        ):
            subject_steps = [s for s in subject_steps if s != STEP_FREESURFER]
        for step in subject_steps:
            found = missing_inputs_for_step(project_dir, subject_id, step)
            if step == STEP_DTI:
                # Stages selected in the same run produce these inputs first.
                produced = [
                    (
                        Path(get_path_manager(project_dir).qsiprep_subject(subject_id))
                        if STEP_QSIPREP in selected_steps
                        else None
                    ),
                    (
                        Path(get_path_manager(project_dir).m2m(subject_id))
                        / const.FILE_T1
                        if STEP_CHARM in selected_steps
                        else None
                    ),
                ]
                found = [p for p in found if p.path not in produced]
            problems.extend(found)
    problems.extend(missing_freesurfer_license(selected_steps, subject_ids))
    return problems

selected_preprocessing_steps ¶

selected_preprocessing_steps(*, convert_dicom: bool = False, create_m2m: bool = False, run_fastsurfer: bool = False, run_freesurfer: bool = False, freesurfer_recon_all: bool = True, freesurfer_subregions: Sequence[str] = (), run_qsiprep: bool = False, run_qsirecon: bool = False, extract_dti: bool = False) -> list[str]

Return output-producing steps enabled by the current run options.

Source code in tit/pre/preflight.py

def selected_preprocessing_steps(
    *,
    convert_dicom: bool = False,
    create_m2m: bool = False,
    run_fastsurfer: bool = False,
    run_freesurfer: bool = False,
    freesurfer_recon_all: bool = True,
    freesurfer_subregions: Sequence[str] = (),
    run_qsiprep: bool = False,
    run_qsirecon: bool = False,
    extract_dti: bool = False,
) -> list[str]:
    """Return output-producing steps enabled by the current run options."""
    steps: list[str] = []
    if convert_dicom:
        steps.append(STEP_DICOM)
    if create_m2m:
        steps.append(STEP_CHARM)
    if run_fastsurfer:
        steps.append(STEP_FASTSURFER)
    if run_freesurfer:
        if freesurfer_recon_all:
            steps.append(STEP_FREESURFER)
        steps.extend(
            FREESURFER_SUBREGION_STEPS[item]
            for item in dict.fromkeys(freesurfer_subregions)
        )
    if run_qsiprep:
        steps.append(STEP_QSIPREP)
    if run_qsirecon:
        steps.append(STEP_QSIRECON)
    if extract_dti:
        steps.append(STEP_DTI)
    return steps

pre

tit.pre ¶

Public API¶

See Also¶

PreprocessingInputProblem dataclass ¶

PreprocessingOutput dataclass ¶

run_pipeline ¶

Parameters¶

Returns¶

Raises¶

Examples¶

See Also¶

run_dicom_to_nifti ¶

Parameters¶

Raises¶

See Also¶

fastsurfer_available ¶

run_fastsurfer ¶

Parameters¶

Raises¶

See Also¶

run_charm ¶

Parameters¶

Raises¶

See Also¶

run_tissue_analysis ¶

Parameters¶

Returns¶

Raises¶

See Also¶

run_qsiprep ¶

Parameters¶

Raises¶

run_qsirecon ¶

Parameters¶

Raises¶

extract_dti_tensor ¶

Raises¶

discover_subjects ¶

Parameters¶

Returns¶

See Also¶

check_m2m_exists ¶

Parameters¶

Returns¶

See Also¶

find_existing_preprocessing_outputs ¶

find_missing_preprocessing_inputs ¶

selected_preprocessing_steps ¶

PreprocessingInputProblem `dataclass` ¶

PreprocessingOutput `dataclass` ¶