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.
`Dwi`	A diffusion weighted image.
`StructuralImage`	A structural (e.g. T1-weighted) MRI image.

Functions¶

`cacheable`(→ Any)	Decorator that adds transparent caching when a Cache context is active.
`read_dwi_fsl`(→ kwneuro.dwi.Dwi)	Create a DWI from a NIfTI volume and FSL b-value/b-vector files.
`read_structural`(→ kwneuro.structural.StructuralImage)	Create a structural image from a lazy NIfTI volume resource.
`read_volume`(→ kwneuro.io.NiftiVolumeResource)	Create a lazy volume resource from a NIfTI path.
`write_dwi_fsl`(→ kwneuro.dwi.Dwi)	Write a DWI to NIfTI plus FSL b-value/b-vector files.
`write_structural`(→ kwneuro.structural.StructuralImage)	Write a structural image to a NIfTI path and return the on-disk image.
`write_volume`(→ kwneuro.io.NiftiVolumeResource)	Write a volume resource to a NIfTI path and return the on-disk resource.

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.Dwi¶

A diffusion weighted image.

volume: kwneuro.resource.VolumeResource¶: The DWI image volume. It is assumed to be a 4D volume, with the first three dimensions being spatial and the final dimension indexing the diffusion weightings.

bval: kwneuro.resource.BvalResource¶: The DWI b-values

bvec: kwneuro.resource.BvecResource¶: The DWI b-vectors

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

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

Save all resources to disk and return a Dwi with all resources being on-disk.

Parameters:

path – The desired save directory.
basename – The desired file basenames, i.e. without an extension.

Returns: A Dwi with its internal resources being on-disk.

get_gtab() → dipy.core.gradients.GradientTable¶: Get the GradientTable for this DWI.

compute_mean_b0() → kwneuro.resource.InMemoryVolumeResource¶: Compute the mean of the b=0 images of a DWI.

static concatenate(dwis: list[Dwi]) → Dwi¶

Concatenate a list of Dwis into a single (loaded) DWI.

The affine and metadata of the first Dwi in the list will be used to concatenate volumes.

denoise() → Dwi¶: Denoise using Patch2Self from DIPY.

extract_brain() → kwneuro.resource.InMemoryVolumeResource¶: Extract brain mask. 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_dwi_batch().

estimate_dti(mask: kwneuro.resource.VolumeResource | None = None) → kwneuro.dti.Dti¶: Estimate diffusion tensor image from this DWI

estimate_noddi(mask: kwneuro.resource.VolumeResource | None = None, dpar: float = 0.0017, n_kernel_dirs: int = 500, regenerate_kernels: bool = True) → kwneuro.noddi.Noddi¶: Estimate NODDI model parameters from this DWI. See kwneuro.noddi.Noddi.estimate_noddi() for details.

kwneuro.read_dwi_fsl(volume: kwneuro.util.PathLike, bval: kwneuro.util.PathLike | None = None, bvec: kwneuro.util.PathLike | None = None) → kwneuro.dwi.Dwi¶

Create a DWI from a NIfTI volume and FSL b-value/b-vector files.

When bval or bvec are omitted, they are inferred from the DWI volume path by replacing a .nii or .nii.gz suffix with .bval and .bvec. This covers common BIDS-style DWI sidecars when the NIfTI path is passed directly.

kwneuro.read_structural(path: kwneuro.util.PathLike) → kwneuro.structural.StructuralImage¶: Create a structural image from a lazy NIfTI volume resource.

kwneuro.read_volume(path: kwneuro.util.PathLike) → kwneuro.io.NiftiVolumeResource¶: Create a lazy volume resource from a NIfTI path.

kwneuro.write_dwi_fsl(dwi: kwneuro.dwi.Dwi, volume: kwneuro.util.PathLike, bval: kwneuro.util.PathLike | None = None, bvec: kwneuro.util.PathLike | None = None) → kwneuro.dwi.Dwi¶

Write a DWI to NIfTI plus FSL b-value/b-vector files.

When bval or bvec are omitted, sidecar paths are inferred from the output volume path in the same way as read_dwi_fsl().

kwneuro.write_structural(structural: kwneuro.structural.StructuralImage, path: kwneuro.util.PathLike) → kwneuro.structural.StructuralImage¶: Write a structural image to a NIfTI path and return the on-disk image.

kwneuro.write_volume(volume: kwneuro.resource.VolumeResource, path: kwneuro.util.PathLike) → kwneuro.io.NiftiVolumeResource¶: Write a volume resource to a NIfTI path and return the on-disk resource.

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.