eegunity.module_eeg_batch package¶
Submodules¶
eegunity.module_eeg_batch.eeg_batch module¶
- class eegunity.module_eeg_batch.eeg_batch.EEGBatch(main_instance)[source]¶
Bases:
UDatasetSharedAttributesThis is a key module of UnifiedDataset class, with focus on batch processing. This EEGBatch class has the same attributes as the UnifiedDataset class.
- align_channel(output_path: str, channel_order: list, min_num_channels: int = 1, miss_bad_data: bool = False, get_data_row_params: Dict | None = None) None[source]¶
Adjust the channel order and perform interpolation on the data.
This method realigns the EEG data channels based on the provided channel_order. It utilizes get_data_row() for retrieving the data. Additional parameters can be passed to get_data_row() via get_data_row_params. For more information on available options, refer to the
get_data_row()function in this documentation.- Parameters:
output_path (str) – The path where the adjusted file will be saved.
channel_order (list) – The desired order of channels, provided as a list.
min_num_channels (int, optional) – The minimum number of channels required for alignment. Defaults to 1.
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next one if an error occurs. Defaults to False.
get_data_row_params (dict, optional) – Additional keyword arguments to be passed to
get_data_row()for data fetching. This allows fine-tuning the data retrieval process.
- Returns:
The function modifies the dataset in place by saving the adjusted data.
- Return type:
None
- Raises:
ValueError – If any invalid channels are found in the provided channel_order or if the number of matching channels is below min_num_channels.
Note
Ensure that the output path is accessible and that the provided channel order is valid before calling this method.
Examples
>>> unified_dataset.eeg_batch.align_channel('/path/to/save/', channel_order=['C3', 'C4', 'O1'], min_num_channels=3)
- auto_domain(get_data_row_params: Dict | None = None) None[source]¶
Automatically modify the ‘Domain Tag’ of each row based on ‘Sampling Rate’ and channel names.
This function processes each row in the dataset and updates the ‘Domain Tag’ by appending the ‘Sampling Rate’ and a unique encoded representation of the channel names. The channel names are retrieved using get_data_row() to ensure accuracy, and parameters can be passed to get_data_row() via get_data_row_params.
The ‘Domain Tag’ is updated in the format: f”row[‘Domain Tag’]-row[‘Sampling Rate’]-ch_enc(channel_names)”.
The function utilizes the batch_process method to apply these modifications across the dataset.
- Parameters:
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
- Returns:
The function modifies the dataset in place by updating the ‘Domain Tag’ column.
- Return type:
None
- Raises:
KeyError – If the required columns (‘Domain Tag’, ‘Sampling Rate’) are missing.
Examples
>>> unified_dataset.eeg_batch.auto_domain(get_data_row_params={'preload': True})
- batch_process(con_func: Callable, app_func: Callable, is_patch: bool, result_type: str | None = None)[source]¶
Process each row of locator based on conditions specified in con_func and apply app_func accordingly. This function handles both list and dataframe return types, ensuring the result aligns with the original locator’s rows based on the is_patch flag.
- Parameters:
con_func (Callable) – A function that takes a row of locator and returns True or False to determine if app_func should be applied to that row. The input is a single row from the locator, which you can access like a dictionary. For example, to read the file path attribute, use: file_path = row[‘File Path’]
app_func (Callable) – A function that processes a row of locator and returns the result. The input is same as con_func.
is_patch (bool) – If True, the returned list length or dataframe rows will match the locator’s row count, using placeholder elements as needed.
result_type ({'series', 'value', None}, optional) – Specifies the expected return type of app_func results. Can be “series”, “value”, or None (case insensitive). Defaults to None.
- Returns:
The processed results, either as a list or dataframe, depending on result_type and app_func return type and consistency. Returns None if result_type is None.
- Return type:
Union[None, list, pd.DataFrame]
- Raises:
ValueError – If result_type is not one of the expected values.
Note
This method is essential when designing a custom processing pipeline for the dataset. Ensure that con_func and app_func are compatible with the structure of the locator. If using is_patch, consider the implications on the data integrity.
Examples
>>> example1 >>> new_locator = unified_dataset.eeg_batch.batch_process(app_func, con_func, is_patch=True, result_type='series') >>> print(new_locator) >>> example2 >>> a_list = unified_dataset.eeg_batch.batch_process(app_func, con_func, is_patch=True, result_type='value') >>> print(a_list) >>> example3 >>> unified_dataset.eeg_batch.batch_process(app_func, con_func, is_patch=True, result_type=None)
- epoch_by_event(output_path: str, resample: int | None = None, exclude_bad: bool = True, miss_bad_data: bool = False, get_data_row_params: Dict | None = None, resample_params: Dict | None = None, epoch_params: Dict | None = None) None[source]¶
Batch process EEG data to create epochs based on events specified in the event_id column.
- Parameters:
output_path (str) – Directory to save the processed epochs.
resample (int, optional) – Resample rate for the raw data. If None, no resampling is performed.
exclude_bad (bool, optional) – Whether to exclude bad epochs. Uses simple heuristics to determine bad epochs. Default is True.
miss_bad_data (bool, optional) – Whether to skip files with processing errors. Default is False.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
resample_params (dict, optional) – Additional parameters passed to raw_data.resample().
epoch_params (dict, optional) – Additional parameters for mne.Epochs, excluding raw_data, events, and event_id.
- Returns:
The function modifies the dataset in place by creating and saving the epochs.
- Return type:
None
- Raises:
ValueError – If any parameters are inconsistent or if the specified output path is invalid.
- epoch_for_pretraining(output_path: str, seg_sec: float, resample: int | None = None, overlap: float = 0.0, exclude_bad: bool = True, baseline: Tuple[float | None, float] | None = None, miss_bad_data: bool = False, get_data_row_params: Dict | None = None, resample_params: Dict | None = None, epoch_params: Dict | None = None) None[source]¶
Processes data by creating epochs for pretraining from raw EEG data, applying optional resampling and event segmentation.
- Parameters:
output_path (str) – Path to save the preprocessed epoch data in .npy format.
seg_sec (float) – Segment length in seconds for each epoch.
resample (Optional[int], optional) – New sampling rate. If specified, raw data will be resampled.
overlap (float, optional) – Fraction of overlap between consecutive segments (0.0 means no overlap).
exclude_bad (bool, optional) – If True, drops epochs marked as bad.
baseline (Tuple[Optional[float], float], optional) – Baseline correction period, represented as a tuple (start, end). Default is (None, 0).
miss_bad_data (bool, optional) – If True, skips files with errors instead of raising an exception.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
resample_params (dict, optional) – Additional parameters passed to raw_data.resample().
epoch_params (dict, optional) – Additional parameters passed to mne.Epochs().
- Returns:
The function modifies the dataset in place by saving the processed epoch data.
- Return type:
None
- Raises:
ValueError – If the segment length is invalid or if any specified parameters are inconsistent.
Note
Ensure that the output path is accessible and that the input data is properly formatted before calling this method.
Examples
>>> unified_dataset.eeg_batch.epoch_for_pretraining('/path/to/save/', seg_sec=2.0, resample=256)
- export_h5Dataset(output_path: str, name: str = 'EEGUnity_export', verbose: bool = False, get_data_row_params: Dict | None = None) None[source]¶
Export the dataset in HDF5 format to the specified output path.
This function processes all files in the dataset, ensuring that each file is stored in a separate group with its own dataset and attributes.
- Parameters:
output_path (str) – The directory path where the exported HDF5 files will be saved. A FileNotFoundError is raised if the path does not exist.
name (str) – The name of the HDF5 file. Must be a string. The default value is ‘EEGUnity_export’. Raises a TypeError if the value provided is not a string.
verbose (bool, optional) – If True, prints the progress of the export process. Default is False.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
- Returns:
The function does not return any value.
- Return type:
None
- Raises:
FileNotFoundError – If the output path does not exist.
FileExistsError – If the HDF5 file already exists in the specified output path.
ValueError – If channel configurations or counts are inconsistent within a domain tag.
TypeError – If the name parameter is not a string.
- filter(output_path: str, filter_type: str = 'bandpass', l_freq: float | None = None, h_freq: float | None = None, notch_freq: float | None = None, auto_adjust_h_freq: bool = True, picks: str = 'all', miss_bad_data: bool = False, get_data_row_params: Dict | None = None, filter_params: Dict | None = None, notch_filter_params: Dict | None = None) None[source]¶
Apply filtering to the data, supporting low-pass, high-pass, band-pass, and notch filters.
- Parameters:
output_path (str) – Path to save the filtered file.
filter_type ({'lowpass', 'highpass', 'bandpass', 'notch'}, optional) – Type of filter to apply. Defaults to ‘bandpass’.
l_freq (float, optional) – Low cutoff frequency for the filter (used in high-pass or low-frequency band-pass filters). Defaults to None.
h_freq (float, optional) – High cutoff frequency for the filter (used in low-pass or high-frequency band-pass filters). Defaults to None.
notch_freq (float, optional) – Frequency for the notch filter. Defaults to None.
auto_adjust_h_freq (bool, optional) – Whether to automatically adjust the high cutoff frequency to fit the Nyquist frequency. Defaults to True.
picks (str, optional) – Channels to be used for filtering. Defaults to ‘all’.
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next one if an error occurs. Defaults to False.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
filter_params (dict, optional) – Additional parameters for mne_raw.filter().
notch_filter_params (dict, optional) – Additional parameters for mne_raw.notch_filter().
- Returns:
The function modifies the dataset in place.
- Return type:
None
- format_channel_names()[source]¶
Format channel names in the dataset and update the ‘Channel Names’ column.
This function processes each row in the dataset, checks the ‘Channel Names’ column, and applies a formatting function to standardize the channel names. The function utilizes the batch_process method to apply the formatting to each row of locator, and the updated channel names are then saved back to the ‘Channel Names’ column.
- Returns:
The function modifies the dataset in place by updating the ‘Channel Names’ column.
- Return type:
None
- Raises:
KeyError – If the ‘Channel Names’ column is missing from the dataset.
Note
Ensure that the dataset is properly loaded and contains the ‘Channel Names’ column before calling this method.
Examples
>>> unified_dataset.eeg_batch.format_channel_names()
- get_events(miss_bad_data: bool = False, get_data_row_params: Dict | None = None, extract_events_params: Dict | None = None) None[source]¶
Extract events and log them in the data rows.
This method processes each data row by applying the get_data_row() and extract_events() functions. Additional parameters can be passed to these functions via get_data_row_params and extract_events_params.
- Parameters:
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next one if an error occurs. Defaults to False.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
extract_events_params (dict, optional) – Additional parameters passed to extract_events() for event extraction.
- Raises:
Exception – If miss_bad_data is False, an exception is raised on processing errors.
Note
Please refer to the documentation of get_data_row() and extract_events() for detailed descriptions of the available parameters.
- get_quality(miss_bad_data: bool = False, get_data_row_params: Dict | None = None) None[source]¶
Process the data quality of EEG files by calculating quality scores for each row in the dataset.
- Parameters:
miss_bad_data (bool, optional) – If True, skips rows that contain bad data without raising an error. If False, raises an exception when encountering bad data.
get_data_row_params (dict, optional) – Additional parameters passed to the get_data_row() function. This allows fine-tuning of parameters such as unit conversion, data normalization, etc. For details, refer to the get_data_row() function documentation.
- Returns:
The function modifies the dataset in place by updating quality scores for each row.
- Return type:
None
- ica(output_path: str, miss_bad_data: bool = False, get_params: Dict | None = None, ica_params: Dict | None = None, fit_params: Dict | None = None, apply_params: Dict | None = None)[source]¶
Apply ICA (Independent Component Analysis) to the specified file in the dataset.
This method applies ICA to clean the EEG data using parameters passed through ica_params, fit_params, and apply_params. Please refer to the official documentation for mne.preprocessing.ICA, ica.fit(), and ica.apply() for the complete list of available parameters.
Documentation links: - mne.preprocessing.ICA: https://mne.tools/stable/generated/mne.preprocessing.ICA.html - ICA.fit: https://mne.tools/stable/generated/mne.preprocessing.ICA.html#fitting-ica - ICA.apply: https://mne.tools/stable/generated/mne.preprocessing.ICA.html#applying-ica
- Parameters:
output_path (str) – Path to save the processed file after applying ICA.
miss_bad_data (bool, optional) – Whether to skip bad data files and continue processing the next one. Defaults to False.
get_params (dict, optional) – Additional parameters passed to eegunity.module_eeg_parser.eeg_parser.get_data_row,
ica_params (dict, optional) – Additional parameters passed to mne.preprocessing.ICA, such as n_components, method, etc.
fit_params (dict, optional) – Additional parameters passed to ica.fit(), such as picks, decim, etc.
apply_params (dict, optional) – Additional parameters passed to ica.apply(), such as exclude, include, etc.
- Returns:
Updates the file path in the dataset locator after ICA is applied.
- Return type:
None
- Raises:
ValueError – If the output path is invalid or if the specified parameters are inconsistent.
Note
Ensure that the input data is properly formatted and that all necessary parameters are specified before calling this method.
Examples
>>> unified_dataset.eeg_batch.ica('/path/to/save/', ica_params={'n_components': 20}, fit_params={'picks': 'eeg'})
- infer_units(miss_bad_data: bool = False, get_data_row_params: Dict | None = None) None[source]¶
Infer the units of each channel and record them in the data line.
- Parameters:
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next file when an error occurs. Defaults to False.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row(). These allow for more flexible data processing during the inference process.
- Raises:
Exception – If an error occurs during file processing and miss_bad_data is set to False.
Note
This method applies a custom function to each row in the dataframe to infer the units for each channel based on the raw MNE data. The function handles errors gracefully if miss_bad_data is True.
- normalize(output_path: str, norm_type: str = 'sample-wise', miss_bad_data: bool = False, domain_mean: bool = True, get_data_row_params: Dict | None = None) None[source]¶
Normalize the data.
This method normalizes the EEG data based on the specified normalization type. It can either perform sample-wise normalization or aggregate by domain mean, depending on the provided parameters.
- Parameters:
output_path (str) – The path where the normalized file will be saved.
norm_type (str) – The type of normalization to perform. It can be: - ‘channel-wise’: Normalize each channel individually based on its mean and standard deviation. - ‘sample-wise’: Normalize all channels based on a common mean and standard deviation.
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next one if an error occurs. Defaults to False.
domain_mean (bool, optional) – If True (default), the function aggregates the results by domain tags. Each domain contains the mean and standard deviation across all related EEG channels. If False, the function calculates and stores individual mean and standard deviation for each EEG recording.
get_data_row_params (dict, optional) – Additional keyword arguments passed to
get_data_row(). This allows users to pass extra parameters required by the get_data_row function seamlessly. For details on the parameters, refer to theget_data_row()function in this documentation.
- Returns:
The function modifies the dataset in place by saving the normalized data.
- Return type:
None
- Raises:
ValueError – If the specified normalization type is invalid or if there are issues with the input data.
Note
Ensure that the output path is accessible and that the input data is properly formatted before calling this method.
Examples
>>> unified_dataset.eeg_batch.normalize('/path/to/save/', norm_type='channel-wise', domain_mean=True)
- process_mean_std(domain_mean: bool = True) None[source]¶
Process the mean and standard deviation for EEG data across different channels and optionally compute domain-level statistics.
This function calculates the mean and standard deviation for all EEG channels, both combined and individually. It can also aggregate the results by domain if domain_mean is set to True.
- Parameters:
domain_mean (bool, optional) – If True (default), the function aggregates the results by domain tags. Each domain contains the mean and standard deviation across all related EEG channels. If False, the function calculates and stores individual mean and standard deviation for each EEG recording.
- Returns:
The function updates the instance by setting the “MEAN STD” column with the calculated mean and standard deviation values. If domain_mean is True, it computes domain-aggregated statistics; otherwise, it stores per-channel results.
- Return type:
None
- Raises:
ValueError – If inconsistent channel names or numbers are found within a domain when domain_mean is True.
Note
Ensure that the EEG data is properly formatted and that all necessary channels are present before calling this method.
Examples
>>> unified_dataset.eeg_batch.process_mean_std(domain_mean=True)
- replace_paths(old_prefix, new_prefix)[source]¶
Replace the prefix of file paths in the dataset according to the provided mapping.
- Parameters:
old_prefix (str) – The old path prefix to be replaced.
new_prefix (str) – The new path prefix to replace the old one.
- Returns:
A new UnifiedDataset with updated locator.
- Return type:
object
- resample(output_path: str, miss_bad_data: bool = False, resample_params: Dict | None = None, get_data_row_params: Dict | None = None) None[source]¶
Resample the data using MNE’s resampling functionality and save the processed data.
- Parameters:
output_path (str) – The path where the resampled file will be saved.
miss_bad_data (bool, optional) – Whether to skip the current file and continue processing the next one if an error occurs. Defaults to False.
resample_params (dict, optional) – Additional parameters to be passed to the mne_raw.resample() function.
get_data_row_params (dict, optional) – Additional parameters to be passed to the get_data_row() function.
- Returns:
The function modifies the dataset in place by saving the resampled data.
- Return type:
None
- Raises:
Exception – If an error occurs during resampling and miss_bad_data is set to False, the error will be raised.
Note
Ensure that the output path is accessible and that the input data is properly formatted before calling this method.
Examples
>>> unified_dataset.eeg_batch.resample('/path/to/save/', resample_params={'sfreq': 256})
- sample_filter(channel_number: Tuple[int, int] | List[int] | None = None, sampling_rate: Tuple[float, float] | List[float] | None = None, duration: Tuple[float, float] | List[float] | None = None, completeness_check: str | None = None, domain_tag: str | None = None, file_type: str | None = None) None[source]¶
Filters the ‘locator’ dataframe based on the given criteria. This function is typically used to select the data file according to specified requirements. For advanced filtering, refer to the batch_process() method.
- Parameters:
channel_number (Union[Tuple[int, int], List[int], None], optional) – A tuple or list with (min, max) values to filter the “Number of Channels” column. If None, this criterion is ignored. Defaults to None.
sampling_rate (Union[Tuple[float, float], List[float], None], optional) – A tuple or list with (min, max) values to filter the “Sampling Rate” column. If None, this criterion is ignored. Defaults to None.
duration (Union[Tuple[float, float], List[float], None], optional) – A tuple or list with (min, max) values to filter the “Duration” column. If None, this criterion is ignored. Defaults to None.
completeness_check (str, optional) – A string that can be ‘Completed’, ‘Unavailable’, or ‘Acceptable’ to filter the “Completeness Check” column. The check is case-insensitive. If None, this criterion is ignored. Defaults to None.
domain_tag (str, optional) – A string to filter the “Domain Tag” column. If None, this criterion is ignored. Defaults to None.
file_type (str, optional) – A string to filter the “File Type” column. If None, this criterion is ignored. Defaults to None.
- Return type:
None
- Raises:
ValueError – If any of the input parameters are not in the expected format (e.g., invalid tuples or strings).
Note
This method modifies the ‘locator’ dataframe in place based on the provided filters.
Examples
>>> unified_dataset.eeg_batch.sample_filter(completeness_check='Completed')
- save_as_other(output_path: str, domain_tag: str | None = None, format: str = 'fif', preserve_events: bool = True, get_data_row_params: Dict = None) UnifiedDataset[source]¶
Save data in the specified format (‘fif’ or ‘csv’) to the given output path. If you want to save as h5DF file, please use ‘export_h5Dataset’, because h5DF file is generally used to save the whole dataset.
- Parameters:
output_path (str) – The directory path where the converted files will be saved. If the path does not exist, a FileNotFoundError is raised.
domain_tag (str, optional) – Optional filter to save only the files with a matching ‘Domain Tag’. If None, all files are processed.
format (str, optional) – The format to save the data in. Supported formats are ‘fif’ and ‘csv’. If an unsupported format is provided, a ValueError is raised. Defaults to ‘fif’.
preserve_events (bool, optional) – If True, event markers will be included in the CSV file, and metadata will be adjusted. Defaults to True.
get_data_row_params (dict, optional) – Additional parameters passed to get_data_row() for data retrieval.
- Returns:
A copied instance of the class with updated file paths and formats after the batch processing is complete.
- Return type:
instance of the same class
- Raises:
FileNotFoundError – If the output path does not exist.
ValueError – If the format is not ‘fif’ or ‘csv’.
Note
Ensure that the output_path is accessible and has the necessary write permissions.
Examples
>>> new_locator = unified_dataset.eeg_batch.save_as_other('/path/to/output', domain_tag='example', format='fif')
- set_column(col_name: str, value: list)[source]¶
Set the specified column in the locator with the given list of values.
- Parameters:
col_name (str) – The name of the column to be set.
value (list) – The list of values to set in the column. Its length must match the number of rows in the dataframe.
- Return type:
None
- Raises:
ValueError – If the length of the value list does not match the number of rows in the dataframe.
TypeError – If the input types are not as expected (e.g., col_name is not a string or value is not a list).
Note
Ensure that the provided value list contains valid entries for the specified column type.
Examples
>>> unified_dataset.eeg_batch.set_column('column_name', [1, 2, 3])
eegunity.module_eeg_batch.eeg_scores module¶
- eegunity.module_eeg_batch.eeg_scores.butter_bandpass(lowcut, highcut, fs, order=4)[source]¶
Design a Butterworth bandpass filter.
- Parameters:
low_cut (float) – The lower cutoff frequency of the filter.
high_cut (float) – The upper cutoff frequency of the filter.
sampling_freq (float) – The sampling frequency of the input signal.
order (int, optional) – The order of the Butterworth filter (default is 4).
- Returns:
b (ndarray) – The numerator (b) coefficients of the IIR filter.
a (ndarray) – The denominator (a) coefficients of the IIR filter.
Note
This function designs a bandpass filter using a Butterworth design. The cutoff frequencies are normalized by the Nyquist frequency, which is half the sampling frequency. The filter coefficients are returned as arrays suitable for use with scipy.signal.lfilter or similar filtering functions.
- eegunity.module_eeg_batch.eeg_scores.butter_bandpass_filter(data, lowcut, highcut, fs, order=4)[source]¶
Apply a bandpass filter to the data array.
- Parameters:
data (ndarray) – Data to filter, with channels as rows.
lowcut (float) – The low frequency cut-off for the filter.
highcut (float) – The high frequency cut-off for the filter.
fs (int) – The sampling frequency of the data.
order (int, optional) – The order of the filter (default is 4).
- Returns:
y – The filtered data.
- Return type:
ndarray
Note
This function uses a Butterworth bandpass filter designed with the specified cut-off frequencies and order. The filter is applied using zero-phase filtering with scipy.signal.filtfilt, which ensures that the filtered signal is not phase-shifted.
- eegunity.module_eeg_batch.eeg_scores.calculate_beta_amplitude_score(beta_data, threshold=20)[source]¶
Calculate Score 4 based on the percentage of beta wave samples in each channel that do not exceed the maximum amplitude threshold.
- Parameters:
beta_data (ndarray) – The EEG data filtered in the beta band, expected to be a 2D array where rows represent channels and columns represent amplitudes at each time point.
threshold (float, optional) – The maximum amplitude threshold for the beta waves (in microvolts, μV). Default is 20.
- Returns:
The average percentage of samples across all channels that do not exceed the threshold.
- Return type:
float
Note
This function counts the number of samples in each channel that are less than or equal to the specified threshold and calculates the percentage of such samples relative to the total number of samples in that channel. The final score is the average percentage across all channels.
- eegunity.module_eeg_batch.eeg_scores.calculate_beta_sinusoidal_score(fft_data)[source]¶
Calculate the beta sinusoidal score by analyzing the proportion of significant energy in the FFT data.
- Parameters:
fft_data (ndarray) – The FFT results of the EEG data, expected to be a 2D array where each row represents the FFT results of a channel.
- Returns:
The sinusoidal score as a percentage.
- Return type:
float
Note
This function computes the total and significant energy for each channel based on the FFT results. It applies a threshold to determine significant frequencies and calculates the score as the percentage of significant energy relative to the total energy. If the total energy is zero, the score for that channel is set to zero to avoid division errors.
- eegunity.module_eeg_batch.eeg_scores.calculate_dominant_frequency(signal, fs)[source]¶
Calculate the dominant frequency of a signal.
- Parameters:
signal (ndarray) – The signal array. This should be a 1D array representing the time-series data of the EEG signal.
fs (int) – The sampling frequency of the signal, representing the number of data points collected per second.
- Returns:
The dominant frequency of the signal, which is the frequency component that has the highest amplitude in the Fourier Transform of the signal.
- Return type:
float
Note
This function computes the Fast Fourier Transform (FFT) of the input signal and identifies the frequency with the highest amplitude. The dominant frequency is determined from the positive frequency components of the FFT.
- eegunity.module_eeg_batch.eeg_scores.calculate_eeg_quality_scores(mne_io_raw)[source]¶
Calculate EEG scores for the given data.
- Parameters:
mne_io_raw (mne.io.Raw) – MNE Raw object containing EEG data.
- Returns:
List of EEG scores corresponding to different metrics.
- Return type:
list
Note
This function calculates various EEG quality scores based on different metrics. The scores are calculated using filtered data from the original EEG signals. If fewer channels are available, only selected scores are computed. Otherwise, additional metrics are included for a more comprehensive assessment.
- eegunity.module_eeg_batch.eeg_scores.calculate_general_amplitude_score(data)[source]¶
Calculate a general amplitude score based on the proportion of signal amplitudes that fall within a specific range.
- Parameters:
data (ndarray) – 2D array of EEG data where rows represent channels and columns represent amplitudes at each time point.
- Returns:
The average score across all channels, normalized to 100.
- Return type:
float
Note
This function calculates the general amplitude score for each channel by counting the number of amplitudes within a specific range (-100 to 100) and dividing it by the total number of amplitudes. The scores are then averaged across all channels to obtain the final general amplitude score.
- eegunity.module_eeg_batch.eeg_scores.calculate_highest_amplitude_score(data, channel_indices)[source]¶
Calculate the highest amplitude score for specific channels within the Alpha band.
- Parameters:
data (ndarray) – The EEG data in the Alpha band, expected to be a 2D array where rows represent channels and columns represent amplitudes at each time point.
channel_indices (list of int) – List of indices for the channels of interest.
- Returns:
The calculated highest amplitude score, normalized to 100.
- Return type:
float
Note
This function first filters the data for the specified channels of interest. It then calculates the maximum amplitude for each channel. The amplitudes are sorted in descending order, and scoring is applied based on the percentile of each channel’s amplitude. The final score is the average of these scores, normalized to 100.
- eegunity.module_eeg_batch.eeg_scores.calculate_symmetry_score(data, channels_left, channels_right, fs)[source]¶
Calculate the symmetry score between two sets of channels.
- Parameters:
data (ndarray) – The EEG data, expected to be a 2D array where rows represent channels and columns represent amplitudes at each time point.
channels_left (list of int) – List of indices for the left channels.
channels_right (list of int) – List of indices for the right channels.
fs (int) – The sampling frequency of the data.
- Returns:
The symmetry score between the two sets of channels, expressed as a percentage.
- Return type:
float
Note
This function filters the data for the specified left and right channels, calculates the dominant frequencies for each set of channels, and computes the correlation score between the dominant frequencies. The symmetry score is normalized to a range of 0 to 100.
- eegunity.module_eeg_batch.eeg_scores.calculate_theta_amplitude_score(data, threshold=30)[source]¶
Calculate the percentage of data points not exceeding a specified amplitude threshold in the theta frequency band.
- Parameters:
data (ndarray) – The EEG data, expected to be a 2D array where rows represent channels and columns represent amplitudes at each time point.
threshold (float, optional) – The amplitude threshold for considering a data point as not exceeding. Default is 30.
- Returns:
The average percentage of data points across all channels that do not exceed the threshold.
- Return type:
float
Note
This function checks for data points in the theta frequency band that are below the specified threshold and calculates the percentage of such points for each channel. It returns the average percentage across all channels. If the input data is not already filtered for the theta band, appropriate filtering should be applied before using this function.
- eegunity.module_eeg_batch.eeg_scores.classify_channels(channels)[source]¶
Classify the given channels into different groups based on their location and type.
- Parameters:
channels (list) – List of channel names.
- Returns:
list – List of channel indices that belong to Score 2.
list – List of left-side channel indices.
list – List of right-side channel indices.
Note
This function classifies channels based on common EEG electrode naming conventions. Channels are grouped into frontal, temporal, parietal, occipital, and auricular categories. It distinguishes between left-side and right-side channels based on the suffix of their names, where odd-numbered suffixes typically denote left-side channels and even-numbered suffixes denote right-side channels. Midline and auricular channels are included in the Score 2 classification.
- eegunity.module_eeg_batch.eeg_scores.plot_radar_chart(scores, score_names, title='EEG Scores')[source]¶
Plot a radar chart for the given scores.
- Parameters:
scores (list of float) – List of scores to plot. Each score represents a metric for evaluating EEG quality.
score_names (list of str) – Names corresponding to the scores. These names describe the metrics for evaluating EEG quality.
title (str, optional) – Title for the radar chart (default is ‘EEG Scores’).
- Return type:
None
Note
This function creates a radar chart using the provided scores and score names. The radar chart is displayed using matplotlib.