eegunity.utils package#

Submodules#

eegunity.utils.channel_align_raw module#

eegunity.utils.channel_align_raw.channel_align_raw(mne_raw, channel_order, min_matched_channel=1)[source]#

Aligns and orders the channels of an MNE Raw object according to a specified channel order.

This function ensures that the channels in the raw MNE object are aligned and ordered according to the specified channel_order. If some channels from channel_order are missing in the raw data, they will be added with zero values and later interpolated.

Parameters:

  • mne_raw (mne.io.Raw) – The raw EEG/MEG data in an MNE Raw object.

  • channel_order (list of str) – The desired order of channels.

  • min_matched_channel (int, optional) – The minimum required number of matched channels, by default 1.

Returns:

The modified raw object with channels aligned and missing channels interpolated.

Return type:

mne.io.Raw

Raises:

ValueError – If the number of matched channels is less than min_matched_channel.

Notes

  • The function picks and reorders the matched channels to match channel_order.

  • If some channels from channel_order are missing in mne_raw, they are added as zero data channels and interpolated.

  • The missing channels are first marked as ‘bad’ before interpolation.

Examples

>>> import mne
>>> raw = mne.io.read_raw_fif('sample_raw.fif', preload=True)
>>> desired_order = ['Fp1', 'Fp2', 'F3', 'F4', 'C3', 'C4', 'P3', 'P4', 'O1', 'O2']
>>> aligned_raw = channel_align_raw(raw, desired_order, min_matched_channel=5)

eegunity.utils.con_udatasets module#

eegunity.utils.con_udatasets.con_udatasets(datasets)[source]#

Concatenates the locator DataFrames of the given UnifiedDataset objects, and returns a new UnifiedDataset with the concatenated locator.

The function checks if all elements in the input list are instances of the ‘UnifiedDataset’ class without directly importing it. It then calls the get_locator() method of each dataset, concatenates them, and sets the new locator in a copied version of the first dataset using set_locator().

Parameters:

datasets (list) – A list of UnifiedDataset instances to concatenate their locators.

Returns:

A new UnifiedDataset with the concatenated locator.

Return type:

UnifiedDataset

Raises:

ValueError – If any element in the list is not an instance of ‘UnifiedDataset’.

eegunity.utils.h5 module#

class eegunity.utils.h5.h5Dataset(path: Path, name: str)[source]#

Bases: object

Handle HDF5 file operations in a format compatible with h5py.

This class is adapted from: 935963004/LaBraM.

addAttributes(src: Dataset | Group, attrName: str, attrValue)[source]#

Add an attribute to a dataset or group.

Parameters:

  • src (h5py.Dataset or h5py.Group) – The target object to which the attribute will be added.

  • attrName (str) – The name of the attribute.

  • attrValue (any) – The value of the attribute.

addDataset(grp: Group, dsName: str, arr: array, chunks: tuple | None = None, **kwargs)[source]#

Add a dataset to a specified group.

Parameters:

  • grp (h5py.Group) – The group to which the dataset will be added.

  • dsName (str) – The name of the dataset.

  • arr (np.array) – The data to store in the dataset.

  • chunks (tuple, optional) – The chunk shape to use when storing the dataset.

  • **kwargs – Additional keyword arguments passed to create_dataset.

Returns:

The created dataset object.

Return type:

h5py.Dataset

addGroup(grpName: str)[source]#

Add a new group to the HDF5 file.

Parameters:

grpName (str) – The name of the group to create.

Returns:

The created group object.

Return type:

h5py.Group

property name#

Get the name of the HDF5 dataset.

Returns:

The name of the HDF5 file.

Return type:

str

save()[source]#

Close the HDF5 file.

eegunity.utils.handle_errors module#

eegunity.utils.handle_errors.handle_errors(miss_bad_data: bool, error_list: list | None = None)[source]#

Decorator to handle errors in function execution based on the miss_bad_data flag.

Parameters:

  • miss_bad_data (bool) – If True, errors are caught and logged instead of raising exceptions.

  • error_list (list, optional) – If provided, errors will be added to this list. Default is None (do not store errors).

Return type:

Decorated function that handles errors as specified.

eegunity.utils.log_processing module#

eegunity.utils.log_processing.log_processing(func)[source]#

Decorator that logs the processing of a data row.

This decorator prints a message indicating which row is being processed before calling the original function.

Parameters:

func (callable) – The function to decorate. It must accept a ‘row’ as its first argument.

Returns:

The wrapped function with added logging behavior.

Return type:

callable

eegunity.utils.normalize module#

eegunity.utils.normalize.normalize_mne(mne_raw: Raw) → Raw[source]#

Normalize each channel of the given MNE Raw object so that its mean is 0 and its standard deviation is 1.

This function processes the data from the MNE Raw object and normalizes each channel independently. The mean of each channel will be set to 0, and the standard deviation will be set to 1, effectively standardizing the data across channels.

Parameters:#

mne_rawmne.io.Raw

An instance of the MNE Raw object containing EEG/MEG data to be normalized.

Returns:#

mne.io.Raw

The input MNE Raw object with its data normalized per channel.

Notes:#

The normalization is done in place, meaning the original data in mne_raw is modified.

Example:#

>>> raw = mne.io.read_raw_fif('sample_data.fif')
>>> raw_normalized = normalize_mne(raw)
>>> print(raw_normalized.get_data())

eegunity.utils.pipeline module#

class eegunity.utils.pipeline.Pipeline(functions)[source]#

Bases: object

Apply a list of functions sequentially to an input.

The Pipeline class enables users to define and apply a sequence of transformations (functions) to input data.

functions#

A list of functions to apply in order.

Type:

list of callable

Examples

EEG processing pipeline using MNE: >>> import mne >>> def bandpass_filter(raw, l_freq, h_freq): … return raw.filter(l_freq=l_freq, h_freq=h_freq) >>> def notch_filter(raw, freqs): … return raw.notch_filter(freqs=freqs) >>> def resample(raw, sfreq): … return raw.resample(sfreq=sfreq) >>> # Define processing functions >>> functions = [ … lambda raw: bandpass_filter(raw, 0.1, 75), … lambda raw: notch_filter(raw, freqs=50), … lambda raw: resample(raw, sfreq=200) … ] >>> # Initialize and apply the pipeline >>> pipeline = Pipeline(functions) >>> processed_raw = pipeline.forward(raw) >>> print(processed_raw.info[‘sfreq’])

forward(X)[source]#

Apply all functions in the pipeline to the input data.

Parameters:

X (any) – The input data to be transformed.

Returns:

The transformed data after applying all functions.

Return type:

any

eegunity.utils.split_hdf5_file module#

eegunity.utils.split_hdf5_file.split_hdf5_file(input_path, max_file_size=10737418240, output_dir='.')[source]#

Split an HDF5 file into multiple parts if its total size exceeds the given limit.

The minimal splitting unit is a top-level group. If the file size surpasses the specified max_file_size, this function creates multiple output HDF5 files and distributes the top-level groups among them without splitting any single group. Output files are named based on the input file’s base name, with suffixes like _s1.hdf5, _s2.hdf5, etc.

Parameters:

  • input_path (str) – Path to the input HDF5 file.

  • max_file_size (int, optional) – Maximum size in bytes for each output HDF5 file (default is 10GB).

  • output_dir (str, optional) – Directory where output files will be saved. Defaults to the current directory.

Returns:

A list of paths to the generated HDF5 files.

Return type:

list of str

Module contents#