kwneuro¶

kwneuro: Processing pipelines to extract brain microstructure from diffusion MRI

Submodules¶

Classes¶

`Cache`	Context manager that activates caching for all `@cacheable`-decorated functions.
`CacheSpec`	Explicit cache spec for `@cacheable` when the return type does not implement the cache protocol.
`StructuralImage`	A structural (e.g. T1-weighted) MRI image.

Functions¶

cacheable(→ Any)

Decorator that adds transparent caching when a Cache context is active.

Package Contents¶

class kwneuro.Cache¶

Context manager that activates caching for all @cacheable-decorated functions.

Scalar arguments and imaging data arguments are fingerprinted automatically. Changes to either trigger a cache miss on the next run. Arguments that cannot be fingerprinted issue a UserWarning. Each subject should use a distinct cache_dir to avoid races.

cache_dir: pathlib.Path¶: Directory where cached outputs and per-step sidecar files are stored. Created automatically.

force: bool | set[str] = False¶: Cache bypass control. False uses all cached outputs, True recomputes everything, a set of step names recomputes only those steps.

is_forced(step_name: str) → bool¶: Return True if this step should bypass the cache.

is_cached(step_name: str, files: list[str], scalars: dict[str, Any] | None = None, hashes: dict[str, str] | None = None) → bool¶

Return True when all output files exist, the step is not forced, and the stored fingerprint matches.

The sidecar file stores two sections: “scalars” for human-readable scalar argument values and “hashes” for sha256 content fingerprints. A mismatch in either section, or a missing sidecar when params are expected, is treated as a cache miss.

Parameters:

step_name – Name of the cache step, used to locate the sidecar file.
files – Output filenames relative to cache_dir that must all exist for a cache hit.
scalars – Scalar argument values to compare against the stored sidecar, or None.
hashes – Content fingerprints to compare against the stored sidecar, or None.

Returns: True on a cache hit, False on a miss.

status(steps: list[collections.abc.Callable[Ellipsis, Any]]) → dict[str, bool]¶

Return a mapping of each step’s qualified name to whether all its cached output files exist.

Non-decorated callables in steps are silently skipped.

Parameters:: steps – List of cacheable-decorated functions to check.

Returns: Dict mapping fn.__qualname__ to True if all cached outputs exist, False otherwise.

class kwneuro.CacheSpec¶

Bases: Generic[T]

Explicit cache spec for @cacheable when the return type does not implement the cache protocol.

Use the bare @cacheable decorator instead when the return type implements _cache_files, _cache_save, and _cache_load.

files: list[str]¶: Output filenames relative to the cache directory.

save: collections.abc.Callable[[T, pathlib.Path], None]¶: Callable that persists the result to the cache directory.

load: collections.abc.Callable[[pathlib.Path], T]¶: Callable that reconstructs the result from the cache directory.

step_name: str = ''¶: Step name used to identify this step in the params sidecar. Defaults to the decorated function name.

kwneuro.cacheable(fn_or_spec: collections.abc.Callable[Ellipsis, Any] | CacheSpec[Any]) → Any¶

Decorator that adds transparent caching when a Cache context is active.

Outside a with Cache(...) block the function runs normally with no overhead. Scalar and dataclass arguments are fingerprinted automatically; a change in either causes a cache miss. Arguments that cannot be fingerprinted trigger a UserWarning.

Can be used in two ways. As a bare decorator when the return type implements the cache protocol (_cache_files, _cache_save, _cache_load):

@cacheable
def estimate_dti(dwi: Dwi, mask: VolumeResource | None = None) -> Dti: ...

Or with an explicit CacheSpec for return types that cannot carry the protocol:

@cacheable(CacheSpec(files=[...], save=..., load=...))
def compute_csd_peaks(...) -> tuple[VolumeResource, VolumeResource]: ...

class kwneuro.StructuralImage¶

A structural (e.g. T1-weighted) MRI image.

volume: kwneuro.resource.VolumeResource¶: The structural MRI volume. Expected to be a 3D volume.

load() → StructuralImage¶: Load any on-disk resources into memory and return a StructuralImage with all resources loaded.

save(path: kwneuro.util.PathLike, basename: str) → StructuralImage¶

Save the volume to disk and return a StructuralImage with an on-disk resource.

Parameters:

path – The desired save directory.
basename – The desired file basename (without extension).

Returns: A StructuralImage with its internal resource being on-disk.

correct_bias() → StructuralImage¶

Apply N4 bias field correction using ANTsPy.

Returns: A new StructuralImage with the bias-corrected volume.

extract_brain() → kwneuro.resource.InMemoryVolumeResource¶

Extract a brain mask using HD-BET.

This is meant to be convenient rather than efficient. Using this in a loop could result in unnecessary repetition of file I/O operations. For efficiency, see kwneuro.masks.brain_extract_structural_batch().

segment_tissues(mask: kwneuro.resource.VolumeResource | None = None, method: str = 'atropos') → kwneuro.resource.InMemoryVolumeResource¶

Segment brain tissues into labeled classes.

Parameters:

mask – Optional brain mask. Used only with method="atropos"; when omitted a mask is generated automatically via ants.get_mask(). Ignored for method="deep_atropos", which handles preprocessing internally.
method –
Segmentation method. One of:
- "atropos" (default): Classical ANTsPy k-means Atropos. Returns a 3-class labeled volume (1=CSF, 2=GM, 3=WM). No extra installation required.
- "deep_atropos": Deep-learning segmentation via ANTsPyNet. Returns a 6-class labeled volume (1=CSF, 2=GM, 3=WM, 4=deep GM, 5=cerebellum, 6=brainstem). Requires pip install kwneuro[antspynet].

Returns: A labeled segmentation volume.

parcellate(method: str = 'dkt') → kwneuro.resource.InMemoryVolumeResource¶

Cortical parcellation.

Parameters:: method – Parcellation method. Currently only "dkt" is supported: Desikan-Killiany-Tourville (DKT) cortical labeling via ANTsPyNet (~84 regions). Requires pip install kwneuro[antspynet].

Returns: A DKT-labeled parcellation volume.