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

addAttributes(src: Dataset | Group, attrName: str, attrValue)[source]#
addDataset(grp: Group, dsName: str, arr: array, chunks: tuple | None = None, **kwargs)[source]#
addGroup(grpName: str)[source]#
property name#
save()[source]#

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

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

A pipeline that applies a list of functions sequentially to an input.

The Pipeline class allows users to define a sequence of transformations (functions) that are applied to an input one after the other.

functions#

A list of functions to be applied in sequence.

Type:

list

Example usage (EEG processing):
>>> 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)
>>> # Load sample data
>>> # raw = mne.io.read_raw_fif(mne.datasets.sample.data_path() + '/MEG/sample/sample_audvis_raw.fif', preload=True)
>>> # Define processing functions for the pipeline
>>> 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]#

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#