kwneuro.cache
=============

.. py:module:: kwneuro.cache


Attributes
----------

.. autoapisummary::

   kwneuro.cache.T


Classes
-------

.. autoapisummary::

   kwneuro.cache.CacheSpec
   kwneuro.cache.Cache


Functions
---------

.. autoapisummary::

   kwneuro.cache.cacheable


Module Contents
---------------

.. py:data:: T

.. py:class:: CacheSpec

   Bases: :py:obj:`Generic`\ [\ :py:obj:`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``.


   .. py:attribute:: files
      :type:  list[str]

      Output filenames relative to the cache directory.


   .. py:attribute:: save
      :type:  collections.abc.Callable[[T, pathlib.Path], None]

      Callable that persists the result to the cache directory.


   .. py:attribute:: load
      :type:  collections.abc.Callable[[pathlib.Path], T]

      Callable that reconstructs the result from the cache directory.


   .. py:attribute:: step_name
      :type:  str
      :value: ''


      Step name used to identify this step in the params sidecar. Defaults to the decorated function name.


.. py:class:: 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.


   .. py:attribute:: cache_dir
      :type:  pathlib.Path

      Directory where cached outputs and per-step sidecar files are stored. Created automatically.


   .. py:attribute:: force
      :type:  bool | set[str]
      :value: False


      Cache bypass control. False uses all cached outputs, True recomputes everything,
      a set of step names recomputes only those steps.


   .. py:method:: is_forced(step_name: str) -> bool

      Return True if this step should bypass the cache.



   .. py:method:: 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.

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

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



   .. py:method:: 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.

      :param steps: List of cacheable-decorated functions to check.

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



.. py:function:: 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]: ...