physioex.data.BasePhysioDataset

class physioex.data.BasePhysioDataset(root=None, channels=None, pipelines='raw', sequence_length=21, subset=None, cache_dir=None, cache_enabled=True, epoch_length_sec=None, label_transform=None, skip_corrupt=True, skip_subjects_without_channels=True, stage_map=None, trim_excess_wake=True, memmap_cache_size=1000)

Bases: Dataset

Abstract base class for lazy-loading EDF-based sleep datasets.

Subclasses must implement:

• _list_subjects() -> list[SubjectSpec]

• _read_subject_labels(spec) -> np.ndarray of int per-epoch labels

Subclasses typically override:

• DATASET_NAME (class attribute, used as cache directory name)

• CHANNEL_PREFERENCES (class attribute, per-modality preference lists)

Optional overrides:

• _read_edf_header(spec) to inject external metadata

• _read_subject_channel(spec, resolved) for non-EDF formats

Parameters:

• root (Optional[str])

• channels (Optional[List[Union[str, Dict]]])

• pipelines (Union[PreprocessingPipeline, str, Dict[str, Union[PreprocessingPipeline, str]]])

• sequence_length (int)

• subset (Optional[str])

• cache_dir (Optional[str])

• cache_enabled (bool)

• epoch_length_sec (Optional[float])

• label_transform (Optional[Callable])

• skip_corrupt (bool)

• skip_subjects_without_channels (bool)

• stage_map (Optional[Dict])

• trim_excess_wake (bool)

• memmap_cache_size (int)

__init__(root=None, channels=None, pipelines='raw', sequence_length=21, subset=None, cache_dir=None, cache_enabled=True, epoch_length_sec=None, label_transform=None, skip_corrupt=True, skip_subjects_without_channels=True, stage_map=None, trim_excess_wake=True, memmap_cache_size=1000)

Parameters:

• root (str | None)

• channels (List[str | Dict] | None)

• pipelines (PreprocessingPipeline | str | Dict[str, PreprocessingPipeline | str])

• sequence_length (int)

• subset (str | None)

• cache_dir (str | None)

• cache_enabled (bool)

• epoch_length_sec (float | None)

• label_transform (Callable | None)

• skip_corrupt (bool)

• skip_subjects_without_channels (bool)

• stage_map (Dict | None)

• trim_excess_wake (bool)

• memmap_cache_size (int)

Methods

__init__([root, channels, pipelines, ...])
available_channels()	Return all unique channel names found across the dataset with their occurrence count.
close()	Explicitly close all open memmap file handles.
get_all_subject_metadata()	Return metadata for all subjects keyed by subject_id.
get_n_subjects()	Return the number of subjects in this dataset.
get_splits([fold])	Return (train, valid, test) subject_id lists.
get_subject_events(subject_id)	Return all temporal events for a subject without loading signals.
get_subject_metadata(subject_id)	Return subject-level metadata without loading signals.
get_subjects()	Return list of all subject IDs.
memmap_cache_stats()	Return LRU memmap cache statistics.
probe([subject])	Return discovered channels + metadata for a subject (int index, string id, or None = first).
split([fold])	Return train/valid/test indices compatible with physioex/data/dataset.py:PhysioExDataset.split.

Attributes

CHANNEL_PREFERENCES
DATASET_NAME
DEFAULT_EPOCH_LENGTH_SEC

available_channels()

Return all unique channel names found across the dataset with their occurrence count. Useful for discovering what channels to request.

Returns:

dict mapping physical_channel_name -> number of subjects that have it.

Return type:

Dict[str, int]

close()

Explicitly close all open memmap file handles.

This clears the LRU memmap cache and closes all file descriptors. Normally not needed as Python’s garbage collector will handle this, but useful for explicit cleanup or when reusing the same dataset object.

Return type:

None

get_all_subject_metadata()

Return metadata for all subjects keyed by subject_id.

Return type:

Dict[str, Dict[str, Any]]

get_n_subjects()

Return the number of subjects in this dataset.

Return type:

int

get_splits(fold=0)

Return (train, valid, test) subject_id lists.

Default: random 70/15/15 split with seed 42 + fold. Subclasses can override to provide custom benchmark folds.

Parameters:

fold (int)

Return type:

Tuple[List[str], List[str], List[str]]

get_subject_events(subject_id)

Return all temporal events for a subject without loading signals.

Parameters:

subject_id (str)

Return type:

List[SleepEvent]

get_subject_metadata(subject_id)

Return subject-level metadata without loading signals.

Parameters:

subject_id (str)

Return type:

Dict[str, Any]

get_subjects()

Return list of all subject IDs.

Return type:

List[str]

memmap_cache_stats()

Return LRU memmap cache statistics.

Returns:

• size: current number of cached memmaps

• max_size: maximum cache size

• hits: number of cache hits

• misses: number of cache misses

• hit_rate: hit rate as percentage

Return type:

Dict with

probe(subject=None)

Return discovered channels + metadata for a subject (int index, string id, or None = first).

Parameters:

subject (int | str | None)

Return type:

Dict[str, Any]

split(fold=0)

Return train/valid/test indices compatible with physioex/data/dataset.py:PhysioExDataset.split.

Training: flat integer indices into the sequence-mode __getitem__. Valid/Test: list of (dataset_idx, subject_id) tuples matching the legacy API. The convention is: one dataset here -> dataset_idx=0 for all subjects.

Parameters:

fold (int)

Return type:

Tuple[ndarray, List[Tuple[int, str]], List[Tuple[int, str]]]