physioex.data.ChannelCache#

class physioex.data.ChannelCache(cache_root=None)[source]#

Bases: object

Parameters:

cache_root (Optional[str])

__init__(cache_root=None)[source]#
Parameters:

cache_root (str | None)

Methods

__init__([cache_root])

atomic_save_array(path, data, meta)

Write data to path atomically, plus {stem}.meta.json sidecar.

clear([dataset, subject, pipeline_hash])

Remove cached data.

dataset_root(dataset)

events_path(dataset, subject_id)

exists(path)

header_path(dataset, subject_id)

labels_meta_path(dataset, subject_id)

labels_path(dataset, subject_id)

load_json(path)

load_memmap(data_path)

Open a cached .npy as a read-only memmap.

save_json(path, obj)

Atomic JSON write.

signal_dir(dataset, subject_id, physical, ...)

signal_meta_path(dataset, subject_id, ...)

signal_path(dataset, subject_id, physical, ...)

subject_meta_path(dataset, subject_id)

Attributes

SCHEMA_VERSION

atomic_save_array(path, data, meta)[source]#

Write data to path atomically, plus {stem}.meta.json sidecar.

Algorithm: write to a temp file in the same directory, flush, then os.replace (POSIX-atomic on same filesystem, safe on NFSv3+).

Parameters:
Return type:

None

clear(dataset=None, subject=None, pipeline_hash=None)[source]#

Remove cached data. Omit args to clear broader scopes.

  • clear() -> removes entire cache_root / SCHEMA_VERSION

  • clear(dataset) -> removes that dataset’s subtree

  • clear(dataset, subject) -> removes subject’s subtree within dataset

  • clear(dataset, subject, pipeline_hash) -> removes that pipeline hash across all channels

Parameters:

  • dataset (str | None)

  • subject (str | None)

  • pipeline_hash (str | None)

Return type:

None

load_memmap(data_path)[source]#

Open a cached .npy as a read-only memmap. Returns (memmap, meta).

For non-native numpy dtypes (e.g. ml_dtypes.bfloat16), np.load returns a void type (|V2). We detect this via the sidecar metadata and reconstruct the memmap with the correct dtype and shape, computing the .npy header offset from the file.

Parameters:

data_path (str | Path)

Return type:

Tuple[ndarray, Dict]

save_json(path, obj)[source]#

Atomic JSON write.

Parameters:
Return type:

None