utils

tit.pre.qsi.utils

Utility functions for QSI integration.

This module provides path resolution, validation, and helper functions for the QSI Docker-out-of-Docker integration.

SdcPlan dataclass

SdcPlan(mode: str, fieldmaps: tuple[Path, ...] = (), problems: tuple[str, ...] = ())

How QSIPrep will correct susceptibility distortion for one subject.

mode is "fieldmap" (reverse-PE fmap/*_epi images, TOPUP), "reverse-pe-dwi" (DWI series with opposite phase encoding, TOPUP) or "syn" (fieldmap-less SyN). problems names DWI fieldmaps that cannot be used and why.

blocking_error property

blocking_error: str | None

Set when DWI fieldmaps exist but none is usable: falling back would hide it.

check_image_exists

check_image_exists(image: str, tag: str) -> bool

Check if a Docker image exists locally.

Parameters

image : str Docker image name (e.g., 'pennlinc/qsiprep'). tag : str Image tag (e.g., '26.1.1').

Returns

bool True if the image exists locally.

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

def check_image_exists(image: str, tag: str) -> bool:
    """
    Check if a Docker image exists locally.

    Parameters
    ----------
    image : str
        Docker image name (e.g., 'pennlinc/qsiprep').
    tag : str
        Image tag (e.g., '26.1.1').

    Returns
    -------
    bool
        True if the image exists locally.
    """
    try:
        result = subprocess.run(
            ["docker", "image", "inspect", f"{image}:{tag}"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        return result.returncode == 0
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False

pull_image_if_needed

pull_image_if_needed(image: str, tag: str, logger: Logger) -> bool

Pull a Docker image if it doesn't exist locally.

Parameters

image : str Docker image name. tag : str Image tag. logger : logging.Logger Logger for status messages.

Returns

bool True if image is available (either existed or was pulled successfully).

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

def pull_image_if_needed(image: str, tag: str, logger: logging.Logger) -> bool:
    """
    Pull a Docker image if it doesn't exist locally.

    Parameters
    ----------
    image : str
        Docker image name.
    tag : str
        Image tag.
    logger : logging.Logger
        Logger for status messages.

    Returns
    -------
    bool
        True if image is available (either existed or was pulled successfully).
    """
    full_image = f"{image}:{tag}"

    if check_image_exists(image, tag):
        logger.debug(f"Docker image {full_image} already exists locally")
        return True

    logger.info(f"Pulling Docker image {full_image}...")
    try:
        result = subprocess.run(
            ["docker", "pull", full_image],
            capture_output=True,
            text=True,
            timeout=1800,  # 30 minutes timeout for large images
        )
        if result.returncode == 0:
            logger.info(f"Successfully pulled {full_image}")
            return True
        else:
            logger.error(f"Failed to pull {full_image}: {result.stderr}")
            return False
    except subprocess.TimeoutExpired:
        logger.error(f"Timed out pulling {full_image}")
        return False
    except (FileNotFoundError, OSError) as e:
        logger.error(f"Error pulling {full_image}: {e}")
        return False

validate_dood_environment

validate_dood_environment(project_dir: str, *, require_gpu: bool = False, require_x86_64: bool = False) -> tuple[bool, str | None]

Validate Docker-outside-of-Docker prerequisites before QSI runs.

require_x86_64 rejects an arm64 Docker host (Apple Silicon), where QSIPrep cannot run; the check reads docker info's Architecture line.

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

def validate_dood_environment(
    project_dir: str,
    *,
    require_gpu: bool = False,
    require_x86_64: bool = False,
) -> tuple[bool, str | None]:
    """Validate Docker-outside-of-Docker prerequisites before QSI runs.

    *require_x86_64* rejects an arm64 Docker host (Apple Silicon), where QSIPrep
    cannot run; the check reads ``docker info``'s ``Architecture`` line.
    """
    if shutil.which("docker") is None:
        return False, "Docker CLI not found in PATH."

    local_project_dir = os.environ.get(const.ENV_LOCAL_PROJECT_DIR)
    if not local_project_dir:
        return (
            False,
            f"{const.ENV_LOCAL_PROJECT_DIR} is not set; sibling Docker containers "
            "cannot mount the host project directory.",
        )

    if not Path(project_dir).exists():
        return False, f"Project directory does not exist: {project_dir}"

    try:
        result = subprocess.run(
            ["docker", "info"],
            capture_output=True,
            text=True,
            timeout=15,
        )
    except subprocess.TimeoutExpired:
        return False, "Docker did not respond to `docker info` within 15 seconds."
    except OSError as exc:
        return False, f"Docker is not accessible: {exc}"

    if result.returncode != 0:
        detail = (result.stderr or result.stdout or "docker info failed").strip()
        return False, f"Docker daemon is not accessible: {detail}"

    if require_gpu:
        docker_info = f"{result.stdout}\n{result.stderr}".lower()
        if "nvidia" not in docker_info:
            return False, "Docker GPU runtime is not available."

    if require_x86_64:
        arch = next(
            (
                line.split(":", 1)[1].strip()
                for line in (result.stdout or "").splitlines()
                if line.strip().startswith("Architecture:")
            ),
            "",
        )
        if arch.lower() in ("aarch64", "arm64"):
            return False, (
                f"QSIPrep needs an x86-64 (Linux or Windows) Docker host; this one is "
                f"{arch}. On Apple Silicon QSIPrep 26 fails at SynthSeg, whose "
                "TensorFlow needs AVX instructions that emulation does not provide. "
                "Run QSIPrep on an x86-64 machine and copy derivatives/qsiprep/"
                "sub-<id> into this project; DTI extraction then runs here."
            )

    return True, None

nifti_stem

nifti_stem(path: Path) -> str

Return path's name without its .nii or .nii.gz extension.

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

def nifti_stem(path: Path) -> str:
    """Return *path*'s name without its ``.nii`` or ``.nii.gz`` extension."""
    name = path.name
    for suffix in (".nii.gz", ".nii"):
        if name.lower().endswith(suffix):
            return name[: -len(suffix)]
    return path.stem

read_nifti_dims

read_nifti_dims(path: Path) -> tuple[int, ...] | None

Return the 8-element dim field of a NIfTI header, or None.

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

def read_nifti_dims(path: Path) -> tuple[int, ...] | None:
    """Return the 8-element ``dim`` field of a NIfTI header, or ``None``."""
    parsed = _read_nifti_header(path)
    return parsed[0] if parsed else None

read_nifti_zooms

read_nifti_zooms(path: Path) -> tuple[float, float, float] | None

Return the three spatial voxel sizes (mm) of a NIfTI header, or None.

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

def read_nifti_zooms(path: Path) -> tuple[float, float, float] | None:
    """Return the three spatial voxel sizes (mm) of a NIfTI header, or ``None``."""
    parsed = _read_nifti_header(path)
    if not parsed:
        return None
    zooms = tuple(abs(float(z)) for z in parsed[1][1:4])
    return zooms if all(z > 0 for z in zooms) else None  # type: ignore[return-value]

validate_bids_dwi

validate_bids_dwi(project_dir: str, subject_id: str, logger: Logger) -> tuple[bool, str | None]

Validate that usable DWI data exists for a subject in BIDS format.

Checks that every *_dwi.nii* under the subject's dwi/ folder has a gradient table that matches it: same basename, parseable, one b-value and one direction per volume, and enough distinct directions to fit a tensor.

Parameters

project_dir : str Path to the BIDS project root. subject_id : str Subject identifier (without 'sub-' prefix). logger : logging.Logger Logger for status messages.

Returns

tuple[bool, str | None] (is_valid, error_message). If valid, error_message is None.

See Also

ensure_total_readout_time : The sidecar metadata QSIPrep needs alongside this.

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

def validate_bids_dwi(
    project_dir: str, subject_id: str, logger: logging.Logger
) -> tuple[bool, str | None]:
    """
    Validate that usable DWI data exists for a subject in BIDS format.

    Checks that every ``*_dwi.nii*`` under the subject's ``dwi/`` folder has a
    gradient table that matches it: same basename, parseable, one b-value and
    one direction per volume, and enough distinct directions to fit a tensor.

    Parameters
    ----------
    project_dir : str
        Path to the BIDS project root.
    subject_id : str
        Subject identifier (without 'sub-' prefix).
    logger : logging.Logger
        Logger for status messages.

    Returns
    -------
    tuple[bool, str | None]
        (is_valid, error_message). If valid, error_message is None.

    See Also
    --------
    ensure_total_readout_time : The sidecar metadata QSIPrep needs alongside this.
    """
    dwi_dir = Path(get_path_manager(project_dir).bids_dwi(subject_id))

    if not dwi_dir.exists():
        return False, f"DWI directory not found: {dwi_dir}"

    dwi_files = sorted(
        path for path in dwi_dir.glob("*_dwi.nii*") if not path.name.startswith(".")
    )
    if not dwi_files:
        return False, f"No DWI NIfTI files found in {dwi_dir}"

    for dwi_file in dwi_files:
        error = _validate_gradient_table(dwi_file, Path(project_dir), logger)
        if error:
            return False, error

    logger.debug(f"Found valid DWI data for sub-{subject_id}")
    return True, None

ensure_total_readout_time

ensure_total_readout_time(project_dir: str, subject_id: str, *, logger: Logger, repair: bool = True) -> tuple[bool, str | None]

Make sure every DWI sidecar carries the metadata QSIPrep dereferences.

QSIPrep formats TotalReadoutTime into an FSL acqp line for every run, including ones with no fieldmap and no TOPUP, and its sidecar reader has no fallback when the key is absent -- the run dies with TypeError: must be real number, not NoneType only after the anatomical workflow has finished, an hour in.

A missing value is derived from EstimatedTotalReadoutTime or from EffectiveEchoSpacing and the phase-encode matrix size when the sidecar carries them. Failing that, and only when the subject has no fieldmap and a single phase-encoding direction, a conventional placeholder is written: in that configuration no susceptibility correction is estimated, so the readout time is a common scale factor that cancels. Anything else is reported rather than guessed, because a wrong readout time does bias distortion correction once a fieldmap is present.

Parameters

project_dir : str Path to the BIDS project root. subject_id : str Subject identifier (without 'sub-' prefix). logger : logging.Logger Logger for status messages. repair : bool, optional Write the derived value back into the sidecar. When False a missing value is reported as an error instead. Default: True.

Returns

tuple[bool, str | None] (is_ok, error_message). If ok, error_message is None.

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

def ensure_total_readout_time(
    project_dir: str,
    subject_id: str,
    *,
    logger: logging.Logger,
    repair: bool = True,
) -> tuple[bool, str | None]:
    """Make sure every DWI sidecar carries the metadata QSIPrep dereferences.

    QSIPrep formats ``TotalReadoutTime`` into an FSL ``acqp`` line for *every*
    run, including ones with no fieldmap and no TOPUP, and its sidecar reader
    has no fallback when the key is absent -- the run dies with
    ``TypeError: must be real number, not NoneType`` only after the anatomical
    workflow has finished, an hour in.

    A missing value is derived from ``EstimatedTotalReadoutTime`` or from
    ``EffectiveEchoSpacing`` and the phase-encode matrix size when the sidecar
    carries them. Failing that, and only when the subject has no fieldmap and
    a single phase-encoding direction, a conventional placeholder is written:
    in that configuration no susceptibility correction is estimated, so the
    readout time is a common scale factor that cancels. Anything else is
    reported rather than guessed, because a wrong readout time does bias
    distortion correction once a fieldmap is present.

    Parameters
    ----------
    project_dir : str
        Path to the BIDS project root.
    subject_id : str
        Subject identifier (without 'sub-' prefix).
    logger : logging.Logger
        Logger for status messages.
    repair : bool, optional
        Write the derived value back into the sidecar. When *False* a missing
        value is reported as an error instead. Default: True.

    Returns
    -------
    tuple[bool, str | None]
        (is_ok, error_message). If ok, error_message is None.
    """
    dwi_dir = Path(get_path_manager(project_dir).bids_dwi(subject_id))
    dwi_files = sorted(
        path for path in dwi_dir.glob("*_dwi.nii*") if not path.name.startswith(".")
    )

    has_fieldmaps = _subject_has_fieldmaps(project_dir, subject_id)
    pe_directions: set[str] = set()
    pending: list[tuple[Path, dict]] = []

    for dwi_file in dwi_files:
        stem = nifti_stem(dwi_file)
        sidecar = dwi_file.with_name(f"{stem}.json")
        if not sidecar.is_file():
            return False, (
                f"{dwi_file.name} has no {stem}.json sidecar. QSIPrep reads "
                "PhaseEncodingDirection and TotalReadoutTime from it."
            )
        try:
            with open(sidecar, "r", encoding="utf-8") as handle:
                metadata = json.load(handle)
        except (OSError, json.JSONDecodeError) as exc:
            return False, f"{sidecar} could not be read as JSON: {exc}"

        pe_dir = metadata.get("PhaseEncodingDirection")
        if not pe_dir:
            return False, (
                f"{sidecar.name} has no PhaseEncodingDirection. QSIPrep needs it "
                "to build the eddy acquisition parameters and cannot run without it."
            )
        pe_directions.add(pe_dir)

        readout = metadata.get("TotalReadoutTime")
        if isinstance(readout, (int, float)) and readout > 0:
            continue
        pending.append((sidecar, metadata))

    for sidecar, metadata in pending:
        value, provenance = _derive_total_readout_time(metadata)
        if value is None:
            if has_fieldmaps or len(pe_directions) > 1:
                return False, (
                    f"{sidecar.name} has no TotalReadoutTime and nothing to derive "
                    "it from (EstimatedTotalReadoutTime, or EffectiveEchoSpacing "
                    "with ReconMatrixPE). This subject has fieldmaps or more than "
                    "one phase-encoding direction, so the value affects distortion "
                    "correction and must come from the acquisition -- add it to "
                    "the sidecar before rerunning."
                )
            value = const.QSI_FALLBACK_TOTAL_READOUT_TIME
            provenance = (
                "placeholder (no fieldmap and a single phase-encoding "
                "direction, so the value cancels)"
            )

        if not repair:
            return False, (
                f"{sidecar.name} has no TotalReadoutTime. QSIPrep will fail on it. "
                f"Derivable value: {value:g} s from {provenance}."
            )

        metadata["TotalReadoutTime"] = value
        try:
            with open(sidecar, "w", encoding="utf-8") as handle:
                json.dump(metadata, handle, indent=2, sort_keys=True)
                handle.write("\n")
        except OSError as exc:
            return False, f"Could not write TotalReadoutTime into {sidecar}: {exc}"

        logger.warning(
            f"Added TotalReadoutTime={value:g}s to {sidecar.name} "
            f"[{provenance}]. QSIPrep cannot run without this field."
        )

    return True, None

plan_distortion_correction

plan_distortion_correction(project_dir: str, subject_id: str, *, logger: Logger, repair: bool = True) -> SdcPlan

Choose QSIPrep's susceptibility distortion correction for subject_id.

An fmap/*_epi image is treated as a DWI fieldmap when its IntendedFor already names a DWI, or, with no IntendedFor, when its acq label says dwi/dti or it has no acq label and the DWI's matrix. It is usable when its sidecar has a PhaseEncodingDirection opposite to a DWI's on the same axis and a TotalReadoutTime (derived like the DWI's when absent). With repair, a usable fieldmap missing IntendedFor or TotalReadoutTime gets them written into its own sidecar; nothing else is touched. Unusable DWI fieldmaps are reported in problems, never guessed at.

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

def plan_distortion_correction(
    project_dir: str,
    subject_id: str,
    *,
    logger: logging.Logger,
    repair: bool = True,
) -> SdcPlan:
    """Choose QSIPrep's susceptibility distortion correction for *subject_id*.

    An ``fmap/*_epi`` image is treated as a DWI fieldmap when its ``IntendedFor``
    already names a DWI, or, with no ``IntendedFor``, when its ``acq`` label says
    dwi/dti or it has no ``acq`` label and the DWI's matrix. It is usable when its
    sidecar has a ``PhaseEncodingDirection`` opposite to a DWI's on the same axis and
    a ``TotalReadoutTime`` (derived like the DWI's when absent). With *repair*, a
    usable fieldmap missing ``IntendedFor`` or ``TotalReadoutTime`` gets them written
    into its own sidecar; nothing else is touched. Unusable DWI fieldmaps are
    reported in ``problems``, never guessed at.
    """
    series = _dwi_series(project_dir, subject_id)
    dwi_pes = {
        pe
        for pe in (_pe(meta.get("PhaseEncodingDirection")) for _, meta in series)
        if pe
    }
    if any((axis, -sign) in dwi_pes for axis, sign in dwi_pes):
        return SdcPlan("reverse-pe-dwi")

    dwi_dims = {
        dims[1:4] for dims in (read_nifti_dims(path) for path, _ in series) if dims
    }
    dwi_names = [path.name for path, _ in series]
    dwi_sources = {
        str(v) for _, meta in series for v in _as_list(meta.get("B0FieldSource"))
    }
    fmap_dir = Path(get_path_manager(project_dir).bids_datatype(subject_id, "fmap"))
    usable: list[Path] = []
    problems: list[str] = []
    for fmap in sorted(fmap_dir.glob("*_epi.nii*")):
        if fmap.name.startswith("."):
            continue
        stem = nifti_stem(fmap)
        acq = next(
            (part[4:].lower() for part in stem.split("_") if part.startswith("acq-")),
            None,
        )
        dims = read_nifti_dims(fmap)
        looks_dwi = (
            any(tag in acq for tag in _DWI_FMAP_ACQ)
            if acq
            else bool(dims and dims[1:4] in dwi_dims)
        )
        sidecar = fmap.with_name(f"{stem}.json")
        if not sidecar.is_file():
            if looks_dwi:
                problems.append(
                    f"{fmap.name} has no {sidecar.name} sidecar, so its phase-encoding "
                    "direction is unknown."
                )
            continue
        meta = _read_sidecar(sidecar)
        intended = [str(v) for v in _as_list(meta.get("IntendedFor"))]
        wired = any("dwi/" in v for v in intended) or any(
            str(v) in dwi_sources for v in _as_list(meta.get("B0FieldIdentifier"))
        )
        if not wired and (intended or not looks_dwi):
            continue  # a fieldmap for another modality

        pe = _pe(meta.get("PhaseEncodingDirection"))
        if pe is None:
            problems.append(f"{sidecar.name} has no valid PhaseEncodingDirection.")
            continue
        if not any(axis == pe[0] for axis, _ in dwi_pes):
            problems.append(
                f"{sidecar.name} phase-encodes along {pe[0]}, not the DWI's axis."
            )
            continue
        if (pe[0], -pe[1]) not in dwi_pes:
            problems.append(
                f"{sidecar.name} has the same PhaseEncodingDirection as the DWI "
                f"({meta['PhaseEncodingDirection']}); TOPUP needs the opposite one."
            )
            continue
        changed = []
        readout = meta.get("TotalReadoutTime")
        if not (isinstance(readout, (int, float)) and readout > 0):
            value, provenance = _derive_total_readout_time(meta)
            if value is None:
                problems.append(
                    f"{sidecar.name} has no TotalReadoutTime and nothing to derive it "
                    "from (EstimatedTotalReadoutTime, or EffectiveEchoSpacing with "
                    "ReconMatrixPE)."
                )
                continue
            meta["TotalReadoutTime"] = value
            changed.append(f"TotalReadoutTime={value:g}s [{provenance}]")
        if not wired:
            meta["IntendedFor"] = [f"dwi/{name}" for name in dwi_names]
            changed.append("IntendedFor=" + ", ".join(meta["IntendedFor"]))
        if changed and repair:
            try:
                _write_sidecar(sidecar, meta)
            except OSError as exc:
                problems.append(f"Could not write {sidecar.name}: {exc}")
                continue
            logger.warning(f"Added {'; '.join(changed)} to {sidecar.name}.")
        usable.append(fmap)

    if usable and problems:
        logger.warning("Ignoring unusable DWI fieldmap(s): " + " ".join(problems))
    return SdcPlan("fieldmap" if usable else "syn", tuple(usable), tuple(problems))

choose_unringing_method

choose_unringing_method(project_dir: str, subject_id: str) -> tuple[str, str]

("rpg" | "mrdegibbs", reason) from the DWI sidecars' PartialFourier.

mrdegibbs assumes full k-space; TORTOISE's rpg handles partial-Fourier data.

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

def choose_unringing_method(project_dir: str, subject_id: str) -> tuple[str, str]:
    """``("rpg" | "mrdegibbs", reason)`` from the DWI sidecars' ``PartialFourier``.

    mrdegibbs assumes full k-space; TORTOISE's rpg handles partial-Fourier data.
    """
    fractions = [
        meta.get("PartialFourier") for _, meta in _dwi_series(project_dir, subject_id)
    ]
    partial = [f for f in fractions if isinstance(f, (int, float)) and 0 < f < 1]
    if partial:
        return "rpg", f"PartialFourier {min(partial):g} < 1"
    return "mrdegibbs", "no PartialFourier < 1 in the DWI sidecar"

native_dwi_resolution

native_dwi_resolution(project_dir: str, subject_id: str) -> float | None

The DWI's smallest voxel dimension rounded half-up to 0.1 mm, or None.

QSIPrep resamples to an isotropic grid; the finest native axis keeps every acquired sample without inventing resolution.

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

def native_dwi_resolution(project_dir: str, subject_id: str) -> float | None:
    """The DWI's smallest voxel dimension rounded half-up to 0.1 mm, or ``None``.

    QSIPrep resamples to an isotropic grid; the finest native axis keeps every
    acquired sample without inventing resolution.
    """
    zooms = [
        z
        for path, _ in _dwi_series(project_dir, subject_id)
        for z in (read_nifti_zooms(path) or ())
    ]
    if not zooms:
        return None
    return max(0.1, math.floor(min(zooms) * 10 + 0.5) / 10)

validate_qsiprep_output

validate_qsiprep_output(project_dir: str, subject_id: str) -> tuple[bool, str | None]

Validate that QSIPrep output exists for a subject.

Parameters

project_dir : str Path to the project root. subject_id : str Subject identifier.

Returns

tuple[bool, str | None] (is_valid, error_message). If valid, error_message is None.

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

def validate_qsiprep_output(
    project_dir: str, subject_id: str
) -> tuple[bool, str | None]:
    """
    Validate that QSIPrep output exists for a subject.

    Parameters
    ----------
    project_dir : str
        Path to the project root.
    subject_id : str
        Subject identifier.

    Returns
    -------
    tuple[bool, str | None]
        (is_valid, error_message). If valid, error_message is None.
    """
    qsiprep_dir = Path(get_path_manager(project_dir).qsiprep_subject(subject_id))

    if not qsiprep_dir.exists():
        return False, f"QSIPrep output directory not found: {qsiprep_dir}"

    # Check for preprocessed DWI
    dwi_dir = qsiprep_dir / "dwi"
    if not dwi_dir.exists():
        return False, f"QSIPrep DWI output not found: {dwi_dir}"

    # Check for at least one preprocessed DWI file
    preproc_files = list(dwi_dir.glob("*_dwi.nii*"))
    if not preproc_files:
        return False, f"No preprocessed DWI files found in {dwi_dir}"

    return True, None

qsiprep_has_mni_transform

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

Whether the subject's QSIPrep output holds the anat-to-MNI transform.

QSIPrep writes anat/*_to-MNI152NLin2009cAsym_*xfm.h5 only when its MNI normalization ran (opt-in, or forced by SyN distortion correction); QSIRecon warps its atlases through it (the guard in :func:tit.pre.qsi.qsirecon.run_qsirecon).

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

def qsiprep_has_mni_transform(project_dir: str, subject_id: str) -> bool:
    """Whether the subject's QSIPrep output holds the anat-to-MNI transform.

    QSIPrep writes ``anat/*_to-MNI152NLin2009cAsym_*xfm.h5`` only when its MNI
    normalization ran (opt-in, or forced by SyN distortion correction); QSIRecon warps
    its atlases through it (the guard in :func:`tit.pre.qsi.qsirecon.run_qsirecon`).
    """
    subject = Path(get_path_manager(project_dir).qsiprep_subject(subject_id))
    return any(subject.glob("**/anat/*_to-MNI152NLin2009cAsym_*xfm.h5"))

format_memory_limit

format_memory_limit(memory_gb: int) -> str

Format memory limit for Docker --memory flag.

Parameters

memory_gb : int Memory limit in gigabytes.

Returns

str Formatted memory string (e.g., '32g').

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

def format_memory_limit(memory_gb: int) -> str:
    """
    Format memory limit for Docker --memory flag.

    Parameters
    ----------
    memory_gb : int
        Memory limit in gigabytes.

    Returns
    -------
    str
        Formatted memory string (e.g., '32g').
    """
    return f"{memory_gb}g"

get_container_resource_limits

get_container_resource_limits() -> tuple[int | None, int | None]

Return (cpu_limit, mem_limit_bytes) for the current container.

• cpu_limit: integer number of CPUs available via cgroups/cpuset if limited, otherwise None.
• mem_limit_bytes: memory limit in bytes via cgroups if limited, otherwise None.

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

def get_container_resource_limits() -> tuple[int | None, int | None]:
    """
    Return (cpu_limit, mem_limit_bytes) for the *current* container.

    - cpu_limit: integer number of CPUs available via cgroups/cpuset if limited,
      otherwise None.
    - mem_limit_bytes: memory limit in bytes via cgroups if limited,
      otherwise None.
    """
    # ---- Memory ----
    mem_limit_bytes: int | None = None

    # cgroup v2
    mem_max = _read_first_line("/sys/fs/cgroup/memory.max")
    if mem_max and mem_max != "max":
        try:
            val = int(mem_max)
            # Treat extremely large values as effectively unlimited
            if val > 1 << 60:
                mem_limit_bytes = None
            else:
                mem_limit_bytes = val
        except ValueError:
            mem_limit_bytes = None
    else:
        # cgroup v1
        mem_v1 = _read_first_line("/sys/fs/cgroup/memory/memory.limit_in_bytes")
        if mem_v1:
            try:
                val = int(mem_v1)
                if val > 1 << 60:
                    mem_limit_bytes = None
                else:
                    mem_limit_bytes = val
            except ValueError:
                mem_limit_bytes = None

    # ---- CPU ----
    cpu_limit: int | None = None

    # Prefer cpuset if present
    cpuset = _read_first_line(
        "/sys/fs/cgroup/cpuset.cpus.effective"
    ) or _read_first_line("/sys/fs/cgroup/cpuset/cpuset.cpus")
    cpuset_count = _parse_cpuset(cpuset) if cpuset else None
    if cpuset_count:
        cpu_limit = cpuset_count

    # cgroup v2 cpu.max
    cpu_max = _read_first_line("/sys/fs/cgroup/cpu.max")
    if cpu_max and cpu_max.strip():
        parts = cpu_max.split()
        if len(parts) >= 2 and parts[0] != "max":
            try:
                quota = int(parts[0])
                period = int(parts[1])
                if quota > 0 and period > 0:
                    derived = max(1, math.floor(quota / period))
                    cpu_limit = min(cpu_limit, derived) if cpu_limit else derived
            except ValueError:
                pass
    else:
        # cgroup v1 cpu quota
        quota_s = _read_first_line("/sys/fs/cgroup/cpu/cpu.cfs_quota_us")
        period_s = _read_first_line("/sys/fs/cgroup/cpu/cpu.cfs_period_us")
        if quota_s and period_s:
            try:
                quota = int(quota_s)
                period = int(period_s)
                if quota > 0 and period > 0:
                    derived = max(1, math.floor(quota / period))
                    cpu_limit = min(cpu_limit, derived) if cpu_limit else derived
            except ValueError:
                pass

    return cpu_limit, mem_limit_bytes

get_inherited_dood_resources

get_inherited_dood_resources() -> tuple[int, int]

Determine DooD resource defaults that match the current container.

Returns (cpus, memory_gb) with conservative rounding.

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

def get_inherited_dood_resources() -> tuple[int, int]:
    """
    Determine DooD resource defaults that match the current container.

    Returns (cpus, memory_gb) with conservative rounding.
    """
    cpu_limit, mem_limit_bytes = get_container_resource_limits()

    from tit.cpu import effective_cpus

    # No cgroup limit still does not mean "every core the host has": affinity and cpuset count.
    cpus = cpu_limit or effective_cpus()

    if mem_limit_bytes is None:
        mem_limit_bytes = _get_total_mem_bytes_from_proc()

    if mem_limit_bytes is None:
        # Last-resort fallback: keep existing historical default
        return int(cpus), int(const.QSI_DEFAULT_MEMORY_GB)

    # Convert bytes -> GiB (floor), ensure minimum 4GB
    mem_gb = max(4, int(mem_limit_bytes // (1024**3)))
    return int(cpus), int(mem_gb)