konfai.utils package#

Submodules#

konfai.utils.ITK module#

SimpleITK-based helpers for geometric transforms, resampling, and masking.

konfai.utils.ITK.compose_transform(transform_files, image=None)[source]#

Return type:: sitk.CompositeTransform

konfai.utils.ITK.flatten_transform(transform_files)[source]#

Return type:: sitk.AffineTransform

konfai.utils.ITK.apply_to_image_rigid_transform(image, transform_files)[source]#

Return type:: sitk.Image

konfai.utils.ITK.apply_to_data_transform(data, transform_files)[source]#

Return type:: np.ndarray

konfai.utils.ITK.resample_itk(image_reference, image, transform_files, mask=False, default_pixel_value=None, torch_resample=False)[source]#

Return type:: sitk.Image

konfai.utils.ITK.parametermap_to_transform(path_src)[source]#

Return type:: sitk.Transform | list[sitk.Transform]

konfai.utils.ITK.resample_isotropic(image, spacing=None)[source]#

Return type:: Image

konfai.utils.ITK.resample_resize(image, size=None)[source]#

konfai.utils.ITK.box_with_mask(mask, label, dilatations)[source]#

Return type:: ndarray

konfai.utils.ITK.crop_with_mask(image, box)[source]#

Return type:: Image

konfai.utils.ITK.format_mask_label(mask, labels)[source]#

Return type:: Image

konfai.utils.ITK.get_flat_label(mask, labels=None)[source]#

Return type:: Image

konfai.utils.ITK.clip_and_cast(image, min_value, max_value, dtype)[source]#

Return type:: Image

konfai.utils.config module#

Configuration helpers that map YAML trees to KonfAI Python objects.

class konfai.utils.config.Range(min, max)[source]#

Bases: object

UI hint attached to a parameter’s type — its inclusive numeric bounds.

Use Annotated[int, Range(0, 100)] (or float) in a config-bound signature: the binder ignores the metadata and validates the base type, while a UI reads the bounds to size a spinbox. Introspection-only.

min: float#

max: float#

class konfai.utils.config.Choices(values)[source]#

Bases: object

UI hint attached to a parameter’s type — its allowed values.

Use Annotated[str, Choices([...])] for a fixed list, or Annotated[str, Choices(resolver)] where resolver is a zero-arg callable the app owns (e.g. one that lists a model registry it already fetches). resolve() returns the list — a reader calls it lazily, so the app resolves its own values and no tool re-fetches. Introspection-only; the binder ignores it (a value outside the list is still accepted, e.g. a local path). For a small FIXED, binder-validated set, prefer Literal[...].

resolve()[source]#

Return type:: list

class konfai.utils.config.Config(key)[source]#

Bases: object

Context manager for reading and updating a subtree of the active YAML config.

Parameters:: key (str) – Dot-separated path pointing to the configuration subtree to inspect or materialize.

create_dictionary(data, keys, i)[source]#

Return type:: dict

merge(dict1, dict2)[source]#

Return type:: dict

get_value(name, default)[source]#

Return type:: object

konfai.utils.config.config(key=None)[source]#

Attach a KonfAI configuration key to a class or callable.

Parameters:: key (str | None) – Configuration branch handled by the decorated object.
Returns:: Decorator storing the key on the decorated object.
Return type:: Callable

konfai.utils.config.apply_config(konfai_args=None)[source]#

Recursively instantiate callables from the active KonfAI configuration.

Parameters:: konfai_args (str | None) – Root configuration path used to resolve nested constructor arguments.
Returns:: Decorator that injects configuration-backed arguments at call time.
Return type:: Callable

konfai.utils.dataset module#

Dataset file abstractions and image conversion utilities for KonfAI.

class konfai.utils.dataset.Attribute(attributes=None)[source]#

Bases: dict[str, Any]

Metadata container storing repeated values with a stack-like naming scheme.

pop(key, default=None)[source]#

If the key is not found, return the default if given; otherwise, raise a KeyError.

Return type:: Any

get_np_array(key)[source]#

Return type:: ndarray

get_tensor(key)[source]#

Return type:: Tensor

pop_np_array(key)[source]#

Return type:: ndarray

pop_tensor(key)[source]#

Return type:: Tensor

is_info(key, value)[source]#

Return type:: bool

konfai.utils.dataset.is_an_image(attributes)[source]#

Return whether the given attribute set contains image geometry metadata.

Return type:: bool

konfai.utils.dataset.data_to_image(data, attributes)[source]#

Convert a NumPy array and KonfAI attributes into a SimpleITK image.

Return type:: Image

konfai.utils.dataset.image_to_data(image)[source]#

Convert a SimpleITK image into a channel-first NumPy array and attributes.

Return type:: tuple[ndarray, Attribute]

konfai.utils.dataset.get_infos(filename)[source]#

Read shape and metadata from an image file without loading its full pixel data.

Return type:: tuple[list[int], Attribute]

konfai.utils.dataset.read_landmarks(filename)[source]#

Read Slicer-style fiducial landmarks from disk.

Return type:: ndarray | None

konfai.utils.dataset.write_landmarks(data, filename)[source]#

Write landmarks to the Slicer Markups fiducial CSV-like format.

Return type:: None

class konfai.utils.dataset.DataStream[source]#

Bases: ABC

One dataset entry written incrementally, region by region. Obtained from Dataset.open_data_stream, which returns None when the write format cannot serve region writes (the caller then assembles the volume and uses Dataset.write). Use as a context manager: a clean exit finalizes the entry, an exception removes the partial one so a reader never sees a half-written volume.

The entry lives under a temporary name until the clean exit renames it into place: an existence probe (is_dataset_exist) or a concurrent reader never sees the entry while it is being written, a replaced entry stays readable until its replacement is complete, and a hard-killed writer leaves only temporary debris, never a plausible-looking partial volume under the final name. The temporary name is unique per stream (PID + sequence): two writers of the same entry (a case landing on two workers) each own their temporary, and whichever finalizes last publishes — a complete entry either way, never an interleaving of the two.

static temporary_suffix()[source]#

The per-stream unique suffix a backend appends to its temporary name.

Return type:: str

abstractmethod write_slice(slices, data)[source]#

Write data into the region slices (channel-first indices, step 1).

Return type:: None

close()[source]#

Finalize the entry under its final name.

Return type:: None

abort(error=None)[source]#

Remove the partial entry.

Return type:: None

class konfai.utils.dataset.Dataset(filename, file_format)[source]#

Bases: object

Filesystem or HDF5-backed dataset abstraction used across KonfAI.

class AbstractFile[source]#

Bases: ABC

abstractmethod file_to_data(group, name)[source]#

Return type:: tuple[ndarray, Attribute]

abstractmethod file_to_data_slice(group, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

abstractmethod file_to_data_statistics(group, name, channels=None)[source]#

Return type:: dict[str, float]

abstractmethod data_to_file(name, data, attributes=None)[source]#

Return type:: None

open_data_stream(name, shape, dtype, attributes)[source]#

Open name for incremental region writes; None when this backend cannot.

Return type:: DataStream | None

abstractmethod get_names(group)[source]#

Return type:: list[str]

abstractmethod get_group()[source]#

Return type:: list[str]

abstractmethod is_exist(group, name=None)[source]#

Return type:: bool

abstractmethod get_infos(group, name)[source]#

Return type:: tuple[list[int], Attribute]

class H5File(filename, read)[source]#

Bases: AbstractFile

h5: h5py.File | None#

file_to_data(groups, name)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_slice(groups, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_statistics(groups, name, channels=None)[source]#

Return type:: dict[str, float]

data_to_file(name, data, attributes=None)[source]#

Return type:: None

open_data_stream(name, shape, dtype, attributes)[source]#

Open name for incremental region writes; None when this backend cannot.

Return type:: DataStream | None

is_exist(group, name=None)[source]#

Return type:: bool

get_names(groups, h5_group=None)[source]#

get_group()[source]#

Return type:: list[str]

get_infos(groups, name)[source]#

Return type:: tuple[list[int], Attribute]

class SitkFile(filename, read, file_format)[source]#

Bases: AbstractFile

file_to_data(group, name)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_slice(group, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_statistics(group, name, channels=None)[source]#

Return type:: dict[str, float]

is_vtk_polydata(obj)[source]#

Return type:: bool

data_to_file(name, data, attributes=None)[source]#

Return type:: None

open_data_stream(name, shape, dtype, attributes)[source]#

Open name for incremental region writes; None when this backend cannot.

Return type:: DataStream | None

is_exist(group, name=None)[source]#

Return type:: bool

get_names(group)[source]#

Return type:: list[str]

get_group()[source]#

Return type:: list[str]

get_infos(group, name)[source]#

Return type:: tuple[list[int], Attribute]

class OmeZarrFile(filename, read, level=0)[source]#

Bases: AbstractFile

OME-NGFF backend using chunked Zarr reads for KonfAI patches.

level selects the multiscale pyramid resolution to read (0 = full resolution, higher = coarser); it comes from the omezarr@<level> dataset-spec suffix.

file_to_data(group, name)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_slice(group, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_statistics(group, name, channels=None)[source]#

Return type:: dict[str, float]

data_to_file(name, data, attributes=None)[source]#

Return type:: None

open_data_stream(name, shape, dtype, attributes)[source]#

Open name for incremental region writes; None when this backend cannot.

Return type:: DataStream | None

get_names(group)[source]#

Return type:: list[str]

get_group()[source]#

Return type:: list[str]

is_exist(group, name=None)[source]#

Return type:: bool

get_infos(group, name)[source]#

Return type:: tuple[list[int], Attribute]

class DicomFile(filename, read)[source]#

Bases: AbstractFile

DICOM series backend with header-only metadata and slice-level reads.

file_to_data(group, name)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_slice(group, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

file_to_data_statistics(group, name, channels=None)[source]#

Return type:: dict[str, float]

data_to_file(name, data, attributes=None)[source]#

Return type:: None

get_names(group)[source]#

Return type:: list[str]

get_group()[source]#

Return type:: list[str]

is_exist(group, name=None)[source]#

Return type:: bool

get_infos(group, name)[source]#

Return type:: tuple[list[int], Attribute]

class File(filename, read, file_format, level=0)[source]#

Bases: object

file: AbstractFile | None#

rebase(prefix)[source]#

Prepend prefix to this dataset’s path, re-deriving is_directory from the format.

Return type:: None

concurrent_write_safe()[source]#

Whether writes to different entries land in disjoint files, so a background writer may flush one entry while another thread writes elsewhere in the dataset.

Mirrors the backend dispatch in File.__enter__: everything that is not a single-store backend is a SitkFile directory, one image file per (group, name). A single store (one HDF5 file, one zarr hierarchy, a DICOM series) shares handles and metadata across entries and must stay serial.

Return type:: bool

write(group, name, data, attributes=None)[source]#

Return type:: None

can_stream_data(attributes)[source]#

Whether open_data_stream can serve this dataset’s write format.

H5 and OME-Zarr always can; MetaImage mha needs image geometry to write its header up front; every other format only writes whole volumes (use write).

Return type:: bool

open_data_stream(group, name, shape, dtype, attributes=None)[source]#

Open one entry for incremental region writes.

Returns None when the write format cannot serve region writes; the caller then assembles the volume and uses write. The returned stream is a context manager: a clean exit finalizes the entry, an exception removes the partial one.

Return type:: DataStream | None

read_data(groups, name)[source]#

Return type:: tuple[ndarray, Attribute]

read_data_slice(groups, name, slices)[source]#

Return type:: tuple[ndarray, Attribute]

read_data_statistics(groups, name, channels=None)[source]#

Return type:: dict[str, float]

read_transform(group, name)[source]#

Return type:: Transform

read_image(group, name)[source]#

Return type:: Image

get_size(group)[source]#

Return type:: int

is_group_exist(group)[source]#

Return type:: bool

is_dataset_exist(group, name)[source]#

Return type:: bool

get_names(groups, index=None)[source]#

Return type:: list[str]

get_group()[source]#

Return type:: list[str]

get_infos(groups, name)[source]#

Return type:: tuple[list[int], Attribute]

get_statistics(groups)[source]#

Return type:: dict[str, dict[str, dict[str, float | list[float]]]]

konfai.utils.utils module#

Compatibility facade for KonfAI utility helpers and lightweight array utilities.

konfai.utils.utils.env_flag(name, default)[source]#

A KONFAI_* boolean switch: 0/false is off, anything else set is on, unset is default — one parser, so a switch read in two places accepts the same spellings.

Return type:: bool

konfai.utils.utils.get_module(classpath, default_classpath)[source]#

Import the module a classpath names and return it with the name to take from it.

A : separates the module from the name: everything before the last one is the module, so torch:nn:L1Loss and torch.nn:L1Loss name the same class. Without one, the name is taken from the kind’s own package, and the dots between them lead there: Dice is that package’s own, segmentation.UNet.UNet is under two of its subpackages.

Return type:: tuple[ModuleType, str]

konfai.utils.utils.get_patch_slices_from_nb_patch_per_dim(patch_size_tmp, nb_patch_per_dim, overlap)[source]#

Return type:: list[tuple[slice, ...]]

konfai.utils.utils.DEFAULT_OVERLAP_FRACTION = 0.2#

20 % of the patch.

Type:: Default overlap on tiled axes when a free-axis patch does not say otherwise

konfai.utils.utils.concretize_patch_size(patch_size, shape, multiple=None)[source]#

Resolve the per-axis patch convention onto a concrete shape: 0 = free axis -> full extent.

[0,0,0] (or None) is the whole volume; [1,0,0] is a full 2D slice; a positive entry is fixed by the user and passes through (clamped to the extent so a patch never exceeds the volume).

multiple (the model’s per-axis downsampling_factor) rounds a free axis UP to a valid input size for the network — 122 -> 128 for a factor 16 — so its encoder/decoder skips align. The rounded size may exceed the extent; the border padding (pad_to_patch) fills it and the accumulator crops it back, exactly as it does for any patch larger than its case. A factor shorter than the patch rank aligns to the TRAILING axes (a 2D model in a [1,0,0] slice regime constrains Y and X, not Z).

Return type:: list[int]

konfai.utils.utils.free_axis_rounding(multiple, axis, rank)[source]#

The rounding factor a free axis at spatial index axis (of rank axes) takes from the model’s per-axis multiple — trailing-aligned: a 2D model’s [fy, fx] in a 3-axis patch constrains Y and X and leaves Z at 1.

Return type:: int

konfai.utils.utils.size_free_axes(template, worst, multiple)[source]#

The up-front concrete patch for a free (0) axis config: the worst case with each free axis rounded up to the model’s multiple (its downsampling_factor), so the first attempt is a valid model input. None when there is nothing to size — no free axis, unknown worst case, or every free extent is already a valid multiple (the raw path already works). Shared by the prediction and training auto-patch loops so the rounding lives in one place.

Return type:: list[int] | None

konfai.utils.utils.resolve_overlap(overlap, patch_size, shape)[source]#

Resolve an overlap spec into per-axis voxels for a CONCRETE patch_size.

Accepted forms: None -> DEFAULT_OVERLAP_FRACTION of the patch per axis; an int -> absolute voxels on every axis; a float in [0,1[ or a "20%" string -> that fraction of the patch per axis; a per-axis list mixing those forms. Whatever the spec, an axis that is not tiled (single patch spans the extent) resolves to 0 – overlap only exists between patches.

Return type:: list[int]

konfai.utils.utils.PATCH_BUDGET_SAFETY_FRACTION = 0.8#: Fraction of the budget the sized patch may use; the rest absorbs intermediates the estimate misses.

konfai.utils.utils.resolve_patch(patch_size, shape, channels, dtype_bytes, budget_bytes, resident_images=1, intermediate_factor=1.0, snap=None)[source]#

Size the free axes of patch_size so ONE patch fits budget_bytes; fixed axes never move.

0 entries are free (their max = the volume’s extent); positive entries are pinned by the user. A patch’s footprint is (resident_images + intermediate_factor) * voxels * channels * dtype_bytes – resident_images is exact (counted from the config: output + targets + masks), the intermediate_factor covers the op’s working copies. When everything fits, the free axes take their full extent (the whole volume when all axes are free). Otherwise the free axes shrink ISOTROPICALLY (the patch keeps the volume’s proportions), optionally snapped down to snap multiples (a model’s valid input sizes). budget_bytes=None disables sizing (free = extent). Fixed axes alone exceeding the budget is an error: the user pinned more than the budget allows.

Return type:: list[int]

konfai.utils.utils.get_patch_slices_from_shape(patch_size, shape, overlap_tmp, multiple=None, declared_free_axis=None)[source]#

Return type:: tuple[list[tuple[slice, ...]], list[tuple[int, bool]]]

konfai.utils.utils.is_windows_absolute_path(path)[source]#

Return whether path looks like a Windows absolute path.

Return type:: bool

konfai.utils.utils.split_format_level(file_format)[source]#

Split an optional pyramid-level suffix from a format token.

Used by the OME-Zarr backend to pick a multiscale resolution directly in the dataset spec, e.g. omezarr@2 selects pyramid level 2 (coarser), independently of any transform. Returns (base_format, level) and defaults to level 0 (full resolution) when no @<int> suffix is present.

Return type:: tuple[str, int]

konfai.utils.utils.split_path_spec(value, *, default_format='mha', allowed_flags=None, supported_extensions=None)[source]#

Split a KonfAI path[:flag]:format spec without breaking Windows paths.

KonfAI accepts dataset-like strings such as:

./Dataset
./Dataset:mha
./Dataset:a:mha
C:\Data\Dataset:mha
C:\Data\Dataset:a:mha

Parsing is performed from the right so the drive separator in Windows paths is preserved.

Return type:: tuple[str, str | None, str]

Module contents#

Utility modules supporting configuration, datasets, ITK, DICOM, OME-Zarr, and runtime helpers.