Data#

The physioex.data module turns raw EDF recordings into model-ready tensors through a lazy, content-addressed pipeline. There is no more preprocessed “array” layer and no Lightning data-module: datasets read raw EDF on demand, preprocess per channel on the fly, and cache the result to disk.

The public entry points are:

  • physioex.data.BasePhysioDataset — the abstract torch.utils.data.Dataset base that all concrete datasets inherit from.

  • physioex.data.MultiDataset — concatenates several datasets into one.

  • physioex.data.dict_collate_fn — the collate function that batches the dict samples for a DataLoader.

Concrete datasets are resolved by name from a string registry rather than imported directly. For the full class diagram and how the pieces fit together, see the data-layer architecture page.

Resolving a dataset by name#

physioex.data.datasets exposes a registry of the 12 concrete datasets:

from physioex.data.datasets import get_dataset, available_datasets

available_datasets()
# ['alzheimers', 'dcsm', 'hmc', 'hpap', 'mass', 'mesa', 'mros',
#  'parkinsons', 'shhs', 'sleepedf', 'stages', 'wsc']

HMCDataset = get_dataset("hmc")   # the class, not an instance

Instantiating a dataset#

Every concrete dataset shares the BasePhysioDataset constructor. The most common arguments are the data root (falls back to the PHYSIOEX_DATA environment variable when omitted), the channels to load, a preprocessing pipelines preset, and the sequence_length:

from physioex.data.datasets import get_dataset

data = get_dataset("hmc")(
    root=None,                     # None -> uses $PHYSIOEX_DATA
    channels=["EEG", "EOG", "EMG"],
    pipelines="time_domain",       # a preset name (see the Preprocessing page)
    sequence_length=21,
)

len(data)              # number of indexable epoch-sequences
data.get_n_subjects()  # number of subjects discovered
data.available_channels()

Each item is a dict (not a (signal, label) tuple). The keys are:

Key

Meaning

signals

stacked channel tensor for the sequence

channel_order

ordered list of channel names

channel_info

per-channel resolution metadata

labels

per-epoch AASM labels (-1 = unscored)

subject

subject metadata dict (incl. dataset_idx for MultiDataset)

epoch_indices

indices of the epochs in the recording

recording_length

number of epochs in the full recording

events

sleep events overlapping the sequence

 
item = data[0]
item["signals"].shape   # (sequence_length, n_channels, ...)
item["labels"].shape    # (sequence_length,)

The trailing dimensions of signals depend on the preprocessing preset: time-domain presets yield (L, C, T) samples, while spectrogram presets (seqsleepnet, time_frequency) yield (L, C, T, F).

Splitting#

BasePhysioDataset.split(fold) and get_splits(fold) produce reproducible train/validation/test subject splits (70/15/15, seeded by 42 + fold). See the API reference for the exact return types.

Merging datasets#

MultiDataset concatenates several BasePhysioDataset instances into a single flat index with coordinated splitting. All constituent datasets must share the same sequence_length.

from physioex.data.datasets import get_dataset
from physioex.data import MultiDataset

hmc = get_dataset("hmc")(channels=["EEG"], pipelines="raw", sequence_length=21)
dcsm = get_dataset("dcsm")(channels=["EEG"], pipelines="raw", sequence_length=21)

merged = MultiDataset([hmc, dcsm])

Building a DataLoader#

Because items are dicts, use dict_collate_fn as the loader’s collate_fn:

from torch.utils.data import DataLoader
from physioex.data import dict_collate_fn

loader = DataLoader(data, batch_size=32, shuffle=True, collate_fn=dict_collate_fn)
batch = next(iter(loader))
batch["signals"].shape   # (32, sequence_length, n_channels, ...)

Training

For actual training and evaluation you rarely wire the DataLoader yourself: physioex.train.trainer.Trainer.build_dataloaders detects a BasePhysioDataset/MultiDataset and attaches dict_collate_fn automatically. See the Training page and the train CLIs.

API#

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)[source]

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)

available_channels()[source]

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()[source]

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()[source]

Return metadata for all subjects keyed by subject_id.

Return type:

Dict[str, Dict[str, Any]]

get_n_subjects()[source]

Return the number of subjects in this dataset.

Return type:

int

get_splits(fold=0)[source]

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)[source]

Return all temporal events for a subject without loading signals.

Parameters:

subject_id (str)

Return type:

List[SleepEvent]

get_subject_metadata(subject_id)[source]

Return subject-level metadata without loading signals.

Parameters:

subject_id (str)

Return type:

Dict[str, Any]

get_subjects()[source]

Return list of all subject IDs.

Return type:

List[str]

memmap_cache_stats()[source]

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)[source]

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)[source]

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]]]

class physioex.data.MultiDataset(datasets, memmap_cache_size=1000)[source]

Bases: Dataset

Wraps multiple BasePhysioDataset instances into a single unified dataset.

Provides ConcatDataset-like flat indexing across all datasets, with: - Unified __getitem__ returning the same dict format as BasePhysioDataset - Cross-dataset split coordination via split(fold) - dataset_idx tracking in the returned dict’s subject metadata - Proportional memmap cache distribution across constituent datasets

Parameters:
available_channels()[source]

Return union of available channels across all datasets.

Return type:

Dict[str, int]

close()[source]

Close all memmap caches in constituent datasets.

Return type:

None

property datasets: List[BasePhysioDataset]

Return the wrapped dataset list (read-only reference).

get_n_subjects()[source]

Return total number of subjects across all datasets.

Return type:

int

get_subjects()[source]

Return all subject IDs across all datasets.

Note: IDs are not guaranteed to be unique across datasets. If disambiguation is needed, use get_subjects_with_dataset_idx.

Return type:

List[str]

get_subjects_with_dataset_idx()[source]

Return (dataset_idx, subject_id) pairs for every subject.

Return type:

List[Tuple[int, str]]

memmap_cache_stats()[source]

Aggregate memmap cache statistics across all constituent datasets.

Returns:

Dict with aggregated stats (total_hits, total_misses, hit_rate, etc.) and per-dataset breakdown.

Return type:

Dict[str, Any]

split(fold=0)[source]

Return train/valid/test indices coordinated across all datasets.

Training: flat integer indices into this MultiDataset’s __getitem__. Valid/Test: list of (dataset_idx, subject_id) tuples, where dataset_idx is the position of the originating dataset in the datasets list passed to the constructor.

Parameters:

fold (int)

Return type:

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

physioex.data.dict_collate_fn(batch)[source]

Collate a list of dict-returning dataset items.

Stacks tensor fields across the batch dimension. Preserves dict/list metadata as lists-of-dicts (not dicts-of-lists) so individual sample metadata stays accessible.

Expects each item to have at least:

  • signals: dict[str, Tensor]

  • channel_order: list[str]

  • labels: Tensor

Optional fields: epoch_indices, channel_info, subject, recording_length.

Note

Channel names are expected to be in MODALITY_INDEX format (e.g., “EEG_0”, “EOG_1”). The collate sorts channels by (modality, index) for deterministic ordering.

Parameters:

batch (List[Dict[str, Any]])

Return type:

Dict[str, Any]

See the full auto-generated API Reference for the complete physioex.data surface (readers, cache, events, modality helpers).