brain_pipe.pipeline.cache.default.DefaultPipelineCache

class DefaultPipelineCache(cache_root: str, cache_folder_overrides: ~typing.Dict[~brain_pipe.pipeline.base.PipelineStep, str] | None = None, cache_key='cache', previous_cache_folder_key='previous_cache', previous_caches_key='previous_caches', serializer_fn=<function pickle_dump_wrapper>, deserializer_fn=<function pickle_load_wrapper>, filename_keys=('eeg_path', 'stimulus_path', ['stimuli', 'stimulus_path', ]'trigger_path'), separator='_-_')

Bases: PipelineCache

Default pipeline cache implementation.

__init__(cache_root: str, cache_folder_overrides: ~typing.Dict[~brain_pipe.pipeline.base.PipelineStep, str] | None = None, cache_key='cache', previous_cache_folder_key='previous_cache', previous_caches_key='previous_caches', serializer_fn=<function pickle_dump_wrapper>, deserializer_fn=<function pickle_load_wrapper>, filename_keys=('eeg_path', 'stimulus_path', ['stimuli', 'stimulus_path', ]'trigger_path'), separator='_-_')

Create a new DefaultPipelineCache instance.

Parameters:

  • cache_root (str) – Root folder for the cache.

  • cache_folder_overrides (Optional[Dict[PipelineStep, str]]) – Dictionary of cache folder overrides.

  • cache_key (str) – Cache key.

  • previous_cache_folder_key (str) – Previous cache folder key.

  • previous_caches_key (str) – Previous caches key.

  • serializer_fn (Callable) – Serializer function.

  • deserializer_fn (Callable) – Deserializer function.

  • filename_keys" (Union[str, Sequence[str]]) – Filename keys.

  • separator (str) – Separator.

Methods

__init__(cache_root[, ...])

Create a new DefaultPipelineCache instance.

find_existing_cache_from_data_dict(step, ...)

Find existing cache files for a step.

find_existing_cache_from_previous_filename(...)

Find existing cache files for a step.

get_cache_dict(path, step, step_index)

Create a dictionary containing the cache information.

get_existing_cache_paths(step, data_dict, ...)

Get all existing cache paths for a step.

get_filename(data_dict)

Extract a filename from a data dict.

get_foldername(step, step_index)

Create a folder name for a step.

get_path(step, data_dict, step_index)

Create a path for a cache file.

load(path)

Load a cache file.

load_from_data_dict(data_dict)

Load a cache file from a data_dict.

predict_filenames_from_data_dict(data_dict)

Predict possible filenames from a data dict.

predict_filenames_from_previous_filename(...)

Predict possible filenames from a previous filename.

predict_paths_from_data_dict(step, ...)

Predict the paths of the cache files that will be created by a step.

predict_paths_from_previous_filename(step, ...)

Predict the paths of the cache files that will be created by a step.

save(path, data_dict)

Save a cache file.
find_existing_cache_from_data_dict(step: PipelineStep, data_dict: Dict[str, Any], step_index: int | None) → Sequence[str]

Find existing cache files for a step.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • data_dict (Dict[str, Any]) – Dictionary containing the data.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Sequence of possibly existing cache paths

Return type:

Sequence[str]

find_existing_cache_from_previous_filename(step: PipelineStep, previous_filename: str, step_index: int | None) → Sequence[str]

Find existing cache files for a step.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • previous_filename (str) – Path to the previous cache file.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Sequence of possibly existing cache paths

Return type:

Sequence[str]

get_cache_dict(path: str, step: PipelineStep, step_index: int | None) → Dict[str, Any]

Create a dictionary containing the cache information.

Parameters:

  • path (str) – Path to the cache file

  • step (PipelineStep) – PipelineStep that is being cached

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Dictionary containing the cache information

Return type:

Dict[str, Any]

get_existing_cache_paths(step: PipelineStep, data_dict: Dict[str, Any], step_index: int | None) → Sequence[str]

Get all existing cache paths for a step.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • data_dict (Dict[str, Any]) – Dictionary containing the data.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Sequence of possibly existing cache paths

Return type:

Sequence[str]

get_filename(data_dict: Dict[str, Any]) → str

Extract a filename from a data dict.

Parameters:

data_dict (Dict[str, Any]) – Data dict.

Returns:

Filename.

Return type:

str

get_foldername(step: PipelineStep, step_index: int | None) → str

Create a folder name for a step.

Parameters:

  • step (PipelineStep) – Step to create a folder name for.

  • step_index (Optional[int]) – Step index. If None, no index is added.

Returns:

Folder name.

Return type:

str

get_path(step: PipelineStep, data_dict: Dict[str, Any], step_index: int | None) → str

Create a path for a cache file.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • data_dict (Dict[str, Any]) – Dictionary containing the data.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Path to the cache file

Return type:

str

load(path: str) → Dict[str, Any]

Load a cache file.

Parameters:

path (str) – Path to the cache file

Returns:

Dictionary containing the cache information

Return type:

Dict[str, Any]

load_from_data_dict(data_dict: Dict[str, Any]) → Dict[str, Any]

Load a cache file from a data_dict.

Parameters:

data_dict (Dict[str, Any]) – Dictionary containing the cache information

Returns:

Dictionary containing the cache information

Return type:

Dict[str, Any]

predict_filenames_from_data_dict(data_dict: Dict[str, Any]) → Sequence[str]

Predict possible filenames from a data dict.

Parameters:

data_dict (Dict[str, Any]) – Data dict to predict filenames from.

Returns:

Possible filenames.

Return type:

Sequence[str]

predict_filenames_from_previous_filename(previous_filename: str) → Sequence[Sequence[str]]

Predict possible filenames from a previous filename.

Parameters:

previous_filename (str) – Previous filename.

Returns:

Possible filenames.

Return type:

Sequence[Sequence[str]]

predict_paths_from_data_dict(step: PipelineStep, data_dict: Dict[str, Any], step_index: int | None) → Sequence[str]

Predict the paths of the cache files that will be created by a step.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • data_dict – Dictionary containing the data. The path will be predicted from this.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Sequence of predicted cache paths

Return type:

Sequence[str]

predict_paths_from_previous_filename(step: PipelineStep, previous_filename: str, step_index: int | None) → Sequence[str]

Predict the paths of the cache files that will be created by a step.

Parameters:

  • step (PipelineStep) – PipelineStep that is being cached

  • previous_filename (str) – Path to the previous cache file. The path will be predicted from this.

  • step_index (Optional[int]) – Index of the step in the pipeline. None if the step is not part of a pipeline.

Returns:

Sequence of predicted cache paths

Return type:

Sequence[str]

save(path: str, data_dict: Dict[str, Any])

Save a cache file.

Parameters:

  • path (str) – Path to save the cache file to.

  • data_dict (Dict[str, Any]) – Dictionary containing the cache information

Returns:

Dictionary containing the cache information

Return type:

Dict[str, Any]