docker_builder

tit.pre.qsi.docker_builder

Docker command builder for QSI containers.

This module constructs Docker run commands for QSIPrep and QSIRecon, handling volume mounts, resource allocation, and pipeline arguments.

DockerBuildError

Bases: Exception

Raised when Docker command construction fails.

DockerPaths dataclass

DockerPaths(bids_dir: str = '/data', output_dir: str = '/out', work_dir: str = '/work', license_file: str = '/opt/freesurfer/license.txt')

Container paths for QSI containers.

QSI containers expect BIDS data at /data and outputs at /out.

DockerCommandBuilder

DockerCommandBuilder(project_dir: str, paths: DockerPaths | None = None)

Builds Docker commands for QSIPrep and QSIRecon.

This class handles the complexity of constructing Docker run commands with proper volume mounts, resource limits, and pipeline arguments.

Parameters

project_dir : str Path to the BIDS project directory (container path). paths : DockerPaths | None Container path configuration. Uses defaults if not provided.

Raises

DockerBuildError If Docker is not available or required paths cannot be resolved.

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

def __init__(
    self,
    project_dir: str,
    paths: DockerPaths | None = None,
) -> None:
    # Validate Docker availability

    self.project_dir = project_dir
    self.paths = paths or DockerPaths()

    self._host_project_dir = get_host_project_dir()
    self._fs_license = get_freesurfer_license_path()

build_qsiprep_cmd

build_qsiprep_cmd(config: QSIPrepConfig) -> list[str]

Build Docker command for QSIPrep.

Parameters

config : QSIPrepConfig QSIPrep configuration.

Returns

list[str] Complete Docker command as a list of arguments.

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

def build_qsiprep_cmd(self, config: QSIPrepConfig) -> list[str]:
    """
    Build Docker command for QSIPrep.

    Parameters
    ----------
    config : QSIPrepConfig
        QSIPrep configuration.

    Returns
    -------
    list[str]
        Complete Docker command as a list of arguments.
    """
    image = f"{const.QSI_QSIPREP_IMAGE}:{config.image_tag}"
    inherited_cpus, inherited_mem_gb = get_inherited_dood_resources()
    effective_cpus = (
        config.resources.cpus
        if config.resources.cpus is not None
        else inherited_cpus
    )
    effective_mem_gb = (
        config.resources.memory_gb
        if config.resources.memory_gb is not None
        else inherited_mem_gb
    )

    # Build base docker run command
    cmd = [
        "docker",
        "run",
        "--rm",
        "--name",
        f"qsiprep_{config.subject_id}_{uuid.uuid4().hex[:8]}",
    ]

    # Resource limits
    cmd.extend(
        [
            "--cpus",
            str(effective_cpus),
            "--memory",
            format_memory_limit(effective_mem_gb),
        ]
    )

    # Environment variables
    cmd.extend(
        [
            "-e",
            f"OMP_NUM_THREADS={config.resources.omp_threads}",
        ]
    )

    # Volume mounts - mount host project directory
    # BIDS data is at project root
    cmd.extend(["-v", f"{self._host_project_dir}:{self.paths.bids_dir}:ro"])

    qsiprep_output = str(Path(self._host_project_dir) / "derivatives" / "qsiprep")
    cmd.extend(["-v", f"{qsiprep_output}:{self.paths.output_dir}"])

    work_dir = str(Path(self._host_project_dir) / "derivatives" / ".qsiprep_work")
    cmd.extend(["-v", f"{work_dir}:{self.paths.work_dir}"])

    if self._fs_license:
        cmd.extend(["-v", f"{self._fs_license}:{self.paths.license_file}:ro"])

    cmd.append(image)

    # QSIPrep arguments
    cmd.extend(
        [
            self.paths.bids_dir,
            self.paths.output_dir,
            "participant",
        ]
    )

    cmd.extend(["--participant-label", config.subject_id])
    cmd.extend(["--output-resolution", str(config.output_resolution)])
    cmd.extend(["-w", self.paths.work_dir])
    cmd.extend(["--nthreads", str(effective_cpus)])
    cmd.extend(["--omp-nthreads", str(config.resources.omp_threads)])
    cmd.extend(["--mem-mb", str(effective_mem_gb * 1024)])
    cmd.extend(["--fs-license-file", self.paths.license_file])

    if config.skip_bids_validation:
        cmd.append("--skip-bids-validation")

    if config.denoise_method != "dwidenoise":
        cmd.extend(["--denoise-method", config.denoise_method])

    if config.unringing_method != "mrdegibbs":
        cmd.extend(["--unringing-method", config.unringing_method])

    if config.distortion_group_merge != "none":
        cmd.extend(["--distortion-group-merge", config.distortion_group_merge])

    return cmd

build_qsirecon_cmd

build_qsirecon_cmd(config: QSIReconConfig, recon_spec: str) -> list[str]

Build Docker command for QSIRecon with a specific recon spec.

Parameters

config : QSIReconConfig QSIRecon configuration. recon_spec : str The reconstruction specification to use.

Returns

list[str] Complete Docker command as a list of arguments.

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

def build_qsirecon_cmd(self, config: QSIReconConfig, recon_spec: str) -> list[str]:
    """
    Build Docker command for QSIRecon with a specific recon spec.

    Parameters
    ----------
    config : QSIReconConfig
        QSIRecon configuration.
    recon_spec : str
        The reconstruction specification to use.

    Returns
    -------
    list[str]
        Complete Docker command as a list of arguments.
    """
    image = f"{const.QSI_QSIRECON_IMAGE}:{config.image_tag}"
    inherited_cpus, inherited_mem_gb = get_inherited_dood_resources()
    effective_cpus = (
        config.resources.cpus
        if config.resources.cpus is not None
        else inherited_cpus
    )
    effective_mem_gb = (
        config.resources.memory_gb
        if config.resources.memory_gb is not None
        else inherited_mem_gb
    )

    cmd = [
        "docker",
        "run",
        "--rm",
        "--name",
        f"qsirecon_{config.subject_id}_{recon_spec.replace('-', '_')}_{uuid.uuid4().hex[:8]}",
    ]

    if config.use_gpu:
        cmd.extend(["--gpus", "all"])

    cmd.extend(
        [
            "--cpus",
            str(effective_cpus),
            "--memory",
            format_memory_limit(effective_mem_gb),
        ]
    )

    cmd.extend(
        [
            "-e",
            f"OMP_NUM_THREADS={config.resources.omp_threads}",
        ]
    )

    qsiprep_output = str(Path(self._host_project_dir) / "derivatives" / "qsiprep")
    cmd.extend(["-v", f"{qsiprep_output}:{self.paths.bids_dir}:ro"])

    qsirecon_output = str(Path(self._host_project_dir) / "derivatives" / "qsirecon")
    cmd.extend(["-v", f"{qsirecon_output}:{self.paths.output_dir}"])

    work_dir = str(Path(self._host_project_dir) / "derivatives" / ".qsirecon_work")
    cmd.extend(["-v", f"{work_dir}:{self.paths.work_dir}"])

    cmd.extend(["-v", f"{self._fs_license}:{self.paths.license_file}:ro"])

    cmd.append(image)

    cmd.extend(
        [
            self.paths.bids_dir,
            self.paths.output_dir,
            "participant",
        ]
    )

    cmd.extend(["--participant-label", config.subject_id])
    cmd.extend(["--recon-spec", recon_spec])
    cmd.extend(["-w", self.paths.work_dir])
    cmd.extend(["--nthreads", str(effective_cpus)])
    cmd.extend(["--omp-nthreads", str(config.resources.omp_threads)])
    cmd.extend(["--mem-mb", str(effective_mem_gb * 1024)])
    cmd.extend(["--fs-license-file", self.paths.license_file])

    if config.atlases:
        for atlas in config.atlases:
            cmd.extend(["--atlases", atlas])

    if config.skip_odf_reports:
        cmd.append("--skip-odf-reports")

    return cmd

get_output_dir

get_output_dir(pipeline: str) -> Path

Get the output directory path for a pipeline.

Parameters

pipeline : str Pipeline name ('qsiprep' or 'qsirecon').

Returns

Path Path to the output directory.

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

def get_output_dir(self, pipeline: str) -> Path:
    """
    Get the output directory path for a pipeline.

    Parameters
    ----------
    pipeline : str
        Pipeline name ('qsiprep' or 'qsirecon').

    Returns
    -------
    Path
        Path to the output directory.
    """
    return Path(self._host_project_dir) / "derivatives" / pipeline