kwneuro.cache¶

Attributes¶

T

Classes¶

`CacheSpec`	Explicit cache spec for `@cacheable` when the return type does not implement the cache protocol.
`Cache`	Context manager that activates caching for all `@cacheable`-decorated functions.

Functions¶

cacheable(→ Any)

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

Module Contents¶

kwneuro.cache.T¶

class kwneuro.cache.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.

class kwneuro.cache.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.

kwneuro.cache.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]: ...