GSEGUtils.lazy_disk_cache

class GSEGUtils.lazy_disk_cache.LazyDiskCache

Bases: ABC

_MEMMAP_SUFFIX = '.dat'
__init__(**settings)
Parameters:

settings (Unpack[LazyDiskCacheKw])

Return type:

None

_abc_impl = <_abc._abc_data object>
_convert_to_memmap()

Allocate (or reopen) the persistent memmap file and switch your live buffer to it, but do not flush or drop the old array.

Return type:

None

_convert_to_ndarray()

Pull the memmap entirely into RAM as a plain ndarray.

Return type:

None

abstractmethod _describe_buffer()

Return (shape, dtype, in-memory array)

Return type:

tuple[tuple[int, …], DTypeLike, ndarray[tuple[Any, …], dtype[_ScalarT]]]

abstractmethod _describe_shape_dtype()

Return (shape, dtype) without accessing the full array

Return type:

tuple[tuple[int, …], DTypeLike]

abstractmethod _drop_buffer()

Drop the in-memory array (e.g. set to None or similar)

Return type:

None

_init_from_config(config)
Parameters:

config (LazyDiskCacheConfig)

Return type:

None

abstractmethod _set_buffer(buf)

Given a memmap, restore it into your object

Parameters:

buf (ndarray[tuple[Any, ...], dtype[_ScalarT]])

Return type:

None

property automatic_offloading: bool
property cache_enabled: bool
property cache_path: Path | None
disable_caching()
Return type:

None

disable_purge()

Disable automatic deletion of the cache file when the object is garbage-collected.

enable_caching()
Return type:

None

enable_purge()

Enable automatic deletion of the cache file when the object is garbage-collected. Registers a new finalizer if needed.

static ensure_loaded(func)
load(mode='r+')

Reload the buffer from disk into memory (i.e. make self._mmap your active buffer).

Parameters:

mode (Literal['r', 'r+', 'w+', 'c'])

Return type:

None

offload()

Flush the current buffer to disk, drop the in-RAM array, and mark offloaded.

Return type:

None

property offloaded: bool
on_load()

Hook called after loading into memory. Use to cleanup or reinitialize.

Return type:

None

on_offload()

Hook called after offloading to disk. Use to prune resources.

Return type:

None

property purge_disk_on_gc: bool
class GSEGUtils.lazy_disk_cache.LazyDiskCacheKw

Bases: TypedDict

enable_caching: bool
cache_path: Path | None
purge_disk_on_gc: bool
automatic_offloading: bool
class GSEGUtils.lazy_disk_cache.LazyDiskCacheConfig

Bases: object

__init__(*args, **kwargs)
Parameters:

  • __dataclass_self__ (PydanticDataclass)

  • args (Any)

  • kwargs (Any)

Return type:

None

as_kwargs()
Return type:

LazyDiskCacheKw

automatic_offloading: bool = False
cache_path: Path | None = None
enable_caching: bool = False
extend_cache_path(new_folder)
Parameters:

new_folder (str)

Return type:

Self

classmethod from_kwargs(settings)
Parameters:

settings (LazyDiskCacheKw)

Return type:

Self

purge_disk_on_gc: bool = True
updated(**overrides)
Parameters:

overrides (LazyDiskCacheKw)

Return type:

Self

class GSEGUtils.lazy_disk_cache.DiskBackedNDArray

Bases: LazyDiskCache, NDArrayOperatorsMixin

__init__(data, **settings)
Parameters:
Return type:

None

_abc_impl = <_abc._abc_data object>
_describe_buffer()

Return (shape, dtype, in-memory array)

Return type:

tuple[tuple[int, …], DTypeLike, ndarray]

_describe_shape_dtype()

Return (shape, dtype) without accessing the full array

Return type:

tuple[tuple[int, …], DTypeLike]

_drop_buffer()

Drop the in-memory array (e.g. set to None or similar)

Return type:

None

_set_buffer(buf)

Given a memmap, restore it into your object

Parameters:

buf (ndarray[tuple[Any, ...], dtype[_ScalarT]])

Return type:

None

property data
class GSEGUtils.lazy_disk_cache.DiskBackedStore

Bases: MutableMapping[str, T], Generic

_DBNDArrayFileExt = '.pkl'
__init__(*, config=LazyDiskCacheConfig(), factory, value_type=None, validator=None)
Parameters:
Return type:

None

_abc_impl = <_abc._abc_data object>
_check_T(value)
Parameters:

value (object)

Return type:

T

_get_pickle_path(feature)
Parameters:

feature (str)

Return type:

Path

add_data_to_store(key, data, *, enable_caching_override=None, automatic_offloading_override=None, purge_disk_on_gc_override=None)
Parameters:

  • key (str)

  • data (ndarray[tuple[Any, ...], dtype[_ScalarT]])

  • enable_caching_override (bool | None)

  • automatic_offloading_override (bool | None)

  • purge_disk_on_gc_override (bool | None)

Return type:

None

property cache_dir: Path

Returns the directory where cached files are stored.

items() → a set-like object providing a view on D's items
Return type:

Iterator[tuple[str, T | None]]

keys()

Returns a list of all available image keys.

Return type:

list[str]

offload(keys=None, pickle_container=False)

Offloads selected entries to disk. When no keys are provided, every cached entry is considered. Items with cache_enabled=False are skipped. When pickle_container is True we pickle the entire container, clear the in-memory reference, and rely on lazy reload the next time the key is accessed.

Parameters:
Return type:

None

property store: dict[str, T | None]

Returns the internal store dictionary.

values() → an object providing a view on D's values
Return type:

Iterator[T | None]

_store: dict[str, T | None]
_cache_dir: Path
_enable_caching: bool
_automatic_offloading: bool
_purge_disk_on_gc: bool
_factory: Factory
_value_type: type[T] | tuple[type[T], ...] | None
_validator: Validator | None