preflight

tit.pre.preflight

Preprocessing output preflight and rerun policy helpers.

PreprocessingOutput dataclass

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

Existing output for one preprocessing step.

PreprocessingInputProblem dataclass

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

Missing or invalid input for one selected preprocessing step.

missing_inputs_for_step

missing_inputs_for_step(project_dir: str, subject_id: str, step: str) -> list[PreprocessingInputProblem]

Return missing required inputs for one subject and preprocessing step.

Source code in tit/pre/preflight.py

def missing_inputs_for_step(
    project_dir: str, subject_id: str, step: str
) -> list[PreprocessingInputProblem]:
    """Return missing required inputs for one subject and preprocessing step."""
    pm = get_path_manager(project_dir)
    if step in (STEP_CHARM, STEP_RECON_ALL) and not _has_bids_t1(
        project_dir, subject_id
    ):
        return [
            PreprocessingInputProblem(
                subject_id=subject_id,
                step=step,
                label=STEP_LABELS[step],
                message=(
                    f"{STEP_LABELS[step]} requires a BIDS T1w image for "
                    f"sub-{subject_id}. Run DICOM conversion first or place "
                    f"sub-{subject_id}_T1w.nii[.gz] in the subject anat folder."
                ),
                path=Path(pm.bids_anat(subject_id)),
            )
        ]
    if step == STEP_QSIPREP:
        problems: list[PreprocessingInputProblem] = []
        # QSIPrep needs an anatomical reference (default --anat-modality T1w);
        # without one it fails deep in the workflow ("No T1w images found").
        if not _has_bids_t1(project_dir, subject_id):
            problems.append(
                PreprocessingInputProblem(
                    subject_id=subject_id,
                    step=step,
                    label=STEP_LABELS[step],
                    message=(
                        f"QSIPrep requires a BIDS T1w image for sub-{subject_id}. "
                        f"Run DICOM conversion first or place "
                        f"sub-{subject_id}_T1w.nii[.gz] in the subject anat folder."
                    ),
                    path=Path(pm.bids_anat(subject_id)),
                )
            )
        dwi_ok, dwi_error = validate_bids_dwi(
            project_dir, subject_id, logging.getLogger(__name__)
        )
        if not dwi_ok:
            problems.append(
                PreprocessingInputProblem(
                    subject_id=subject_id,
                    step=step,
                    label=STEP_LABELS[step],
                    message=(
                        f"QSIPrep requires BIDS DWI data for sub-{subject_id}: "
                        f"{dwi_error} Run DICOM conversion with DWI DICOMs or "
                        f"place sub-{subject_id}_dwi.nii[.gz] with .bval/.bvec "
                        "in the subject dwi folder."
                    ),
                    path=Path(pm.bids_dwi(subject_id)),
                )
            )
        return problems
    return []

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_recon: bool = False, 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_recon: bool = False,
    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."""
    selected_steps = (
        list(steps)
        if steps is not None
        else selected_preprocessing_steps(
            convert_dicom=convert_dicom,
            create_m2m=create_m2m,
            run_recon=run_recon,
            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_RECON_ALL, STEP_QSIPREP)
            ]
        for step in subject_steps:
            problems.extend(missing_inputs_for_step(project_dir, subject_id, step))
    return problems

existing_outputs_for_step

existing_outputs_for_step(project_dir: str, subject_id: str, step: str) -> list[PreprocessingOutput]

Return existing outputs for one subject and preprocessing step.

Source code in tit/pre/preflight.py

def existing_outputs_for_step(
    project_dir: str, subject_id: str, step: str
) -> list[PreprocessingOutput]:
    """Return existing outputs for one subject and preprocessing step."""
    pm = get_path_manager(project_dir)
    if step == STEP_DICOM:
        return _dicom_outputs(project_dir, subject_id)
    if step == STEP_CHARM:
        return _single_path_output(
            project_dir, subject_id, step, Path(pm.m2m(subject_id))
        )
    if step == STEP_RECON_ALL:
        return _single_path_output(
            project_dir, subject_id, step, Path(pm.freesurfer_subject(subject_id))
        )
    if step == STEP_QSIPREP:
        return _single_path_output(
            project_dir, subject_id, step, Path(pm.qsiprep_subject(subject_id))
        )
    if step == STEP_QSIRECON:
        return _single_path_output(
            project_dir, subject_id, step, Path(pm.qsirecon_subject(subject_id))
        )
    if step == STEP_DTI:
        return _single_path_output(
            project_dir,
            subject_id,
            step,
            Path(pm.m2m(subject_id)) / const.FILE_DTI_TENSOR,
        )
    raise ValueError(f"Unknown preprocessing step: {step}")

selected_preprocessing_steps

selected_preprocessing_steps(*, convert_dicom: bool = False, create_m2m: bool = False, run_recon: bool = False, 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_recon: bool = False,
    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_recon:
        steps.append(STEP_RECON_ALL)
    if run_qsiprep:
        steps.append(STEP_QSIPREP)
    if run_qsirecon:
        steps.append(STEP_QSIRECON)
    if extract_dti:
        steps.append(STEP_DTI)
    return steps

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_recon: bool = False, 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_recon: bool = False,
    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_recon=run_recon,
            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

remove_preprocessing_output

remove_preprocessing_output(output: PreprocessingOutput) -> None

Remove an existing preprocessing output before an explicit rerun.

Source code in tit/pre/preflight.py

def remove_preprocessing_output(output: PreprocessingOutput) -> None:
    """Remove an existing preprocessing output before an explicit rerun."""
    for path in output.paths_to_remove:
        if not path.exists():
            continue
        if path.is_dir():
            shutil.rmtree(path)
        else:
            path.unlink()