"""
In this module you will find useful functions and classes to apply on-line
Neurofeedback models. Each model is based on different features to be used
as target to train
@author: Diego Marcos-Martínez
"""
# Built-in imports
from abc import ABC, abstractmethod
import concurrent
# External imports
import numpy as np
# from scipy import signal
# from tqdm import tqdm
# Medusa imports
import medusa as mds
from medusa import components
from medusa import meeg
from medusa.spatial_filtering import LaplacianFilter
[docs]class FilterBankPreprocessing(components.ProcessingMethod):
"""
Common preprocessing applied in Neurofeedback applications.
It is composed by a frequency IIR filter followed by a Laplacian spatial filter.
Functions are adapted to filter the signal in more than one frequency range, if necessary.
"""
[docs] def __init__(self, filter_bank=None, montage=None, target_channels=None, n_cha_lp=None):
super().__init__(fit_transform_signal=['signal'],
transform_signal=['signal'])
if filter_bank is None:
filter_bank = [{'order': 4,
'cutoff': (8.0, 12.0),
'btype': 'bandpass',
'filt_method':'sosfiltfilt',
'action': 'increase'}]
if montage is None:
montage = meeg.EEGChannelSet()
montage.set_standard_montage(l_cha=['FZ', 'CZ', 'PZ', 'OZ'],
standard='10-20')
if len(target_channels) == 0:
target_channels = ['CZ']
# Error check
if not filter_bank:
raise ValueError('[FilterBankPreprocessing] Filter bank parameter '
'"filter_bank" must be a list containing all '
'necessary information to perform the filtering!')
for filter in filter_bank:
if not isinstance(filter, dict):
raise ValueError('[FilterBankPreprocessing] Each filter must '
'be a dict()!')
if 'order' not in filter or \
'cutoff' not in filter or \
'filt_method' not in filter or \
'btype' not in filter:
raise ValueError('[FilterBankPreprocessing] Each filter must '
'be a dict() containing the following keys: '
'"order", "cutoff", "filt_method" and "btype"!')
if not montage:
raise ValueError('[FilterBankPreprocessing] Filter bank parameter'
'"montage" must be a dict containing all'
'labels of channels and montage standard key')
if not target_channels:
raise ValueError('[FilterBankPreprocessing] Filter bank parameter'
'"target_channels" must be a list containing all'
'labels of channels to extract NFT features')
# Parameters
self.filter_bank = filter_bank
self.l_cha = montage.l_cha
self.target_channels = target_channels
self.n_cha_lp = n_cha_lp
self.montage = montage
# Variables
self.filter_bank_iir_filters = None
self.offset_removal = None
self.laplacian_filter = None
[docs] def fit(self, fs):
"""
Fits the IIR filter and Laplacian spatial filter.
Parameters
----------
fs: float
Sampling rate in Hz.
"""
# Fit Spectral Filters
self.filter_bank_iir_filters = []
self.offset_removal = mds.IIRFilter(order=2,
cutoff=[1, 40],
btype='bandpass',
filt_method=self.filter_bank[0]['filt_method'])
self.offset_removal.fit(fs, len(self.l_cha))
for filter in self.filter_bank:
iir = mds.IIRFilter(order=filter['order'],
cutoff=filter['cutoff'],
btype=filter['btype'],
filt_method=filter['filt_method'])
iir.fit(fs, len(self.target_channels))
self.filter_bank_iir_filters.append(iir)
# Fit Laplacian Filter
self.laplacian_filter = LaplacianFilter(self.montage, mode='auto')
self.laplacian_filter.fit_lp(n_cha_lp=self.n_cha_lp, l_cha_to_filter=self.target_channels)
[docs]def mean_power(signal):
"""
This function computes the classical NF feature: the mean power across channels.
It also returns the standard
deviation of the band power if chose
Parameters
----------
signal: list or numpy.ndarray
EEG signal already filtered. Shape of [n_samples x n_channels]
"""
return np.mean(np.log(np.mean(np.power(signal, 2), axis=1)), axis=1)
[docs]def std_power(signal):
"""
This function computes the standard deviation of the power of the signal
Parameters
----------
signal: list or numpy.ndarray
EEG signal already filtered. Shape of [n_samples x n_channels]
"""
return np.mean(np.std(np.log(np.power(signal, 2)), axis=1), axis=1)
[docs]class PowerBasedNFTModel(components.Algorithm):
[docs] def __init__(self, fs, filter_bank, l_baseline_t, update_feature_window, montage,
target_channels, n_cha_lp, **kwargs):
"""
Skeleton class for basic Neurofeedback training models. This class inherits from
components.Algorithm. Therefore, it can be used to create standalone
algorithms that can be used in compatible apps from medusa-platform
for online experiments. See components.Algorithm to know more about this
functionality. Calibration and Training methods, as are common to all models, are added
in this skeleton class
"""
super().__init__(calibration=['baseline_parameters'],
training=['feedback_value'])
"""
Class constructor
"""
# Settings
self.fs = fs
self.filter_bank = filter_bank
self.l_baseline_t = l_baseline_t
self.update_feature_window = update_feature_window
self.montage = montage
self.target_channels = target_channels
self.n_cha_lp = n_cha_lp
# Init variables
self.baseline_power = None
self.mode = None
if not self.check_cutoff_settings():
raise Exception('The number of frequency bands selected does not '
'match the Neurofeedback mode.')
self.add_method('prep_method', FilterBankPreprocessing(filter_bank=self.filter_bank, montage=self.montage,
target_channels=self.target_channels,
n_cha_lp=self.n_cha_lp))
self.add_method('feat_ext_method',
FeatureExtraction(fs=fs, l_baseline_t=l_baseline_t,
update_feature_window=update_feature_window))
[docs] def calibration(self, eeg, **kwargs):
"""
Function that receives the EEG signal, filters it and extract the baseline parameter adapted to
the Neurofeedback training mode
Parameters
----------
eeg: numpy.ndarray
EEG signal to process and extract baseline parameter
Returns
-------
baseline_power: float
Value of baseline parameter to display it at Platform
"""
raise NotImplemented
[docs] def training(self, eeg):
"""
Function that receives the EEG signal, filters it and extract the feature adapted to
the Neurofeedback training mode
Parameters
----------
eeg: numpy.ndarray
EEG signal to process and extract baseline parameter
Returns
-------
baseline_power: float
Value of baseline parameter to display it at Platform
"""
raise NotImplemented
[docs] def check_cutoff_settings(self):
"""
Function that receives the EEG signal, filters it and extract the feature adapted to
the Neurofeedback training mode
Parameters
----------
eeg: numpy.ndarray
EEG signal to process and extract baseline parameter
Returns
-------
baseline_power: float
Value of baseline parameter to display it at Platform
"""
raise NotImplemented
[docs]class SingleBandNFT(PowerBasedNFTModel):
"""
The simplest model of Neurofeedback training. The feedback value consist of the
power of the band selected as target
"""
[docs] def __init__(self, fs, filter_bank, l_baseline_t, update_feature_window, montage, target_channels, n_cha_lp):
super().__init__(fs=fs, filter_bank=filter_bank, l_baseline_t=l_baseline_t,
update_feature_window=update_feature_window, montage=montage,
target_channels=target_channels, n_cha_lp=n_cha_lp)
self.get_inst('feat_ext_method').mode = 'single mode'
[docs] def check_cutoff_settings(self):
if len(self.filter_bank) > 1:
return False
else:
return True
[docs] def calibration(self, eeg, **kwargs):
filtered_signal = self.get_inst('prep_method').fit_transform_signal(signal=eeg,
fs=self.fs)
self.baseline_power = self.get_inst('feat_ext_method').set_baseline(signal=filtered_signal)
# return self.baseline_power
[docs] def training(self, eeg, **kwargs):
filtered_signal = self.get_inst('prep_method').transform_signal(signal=eeg)
feedback_value = self.get_inst('feat_ext_method').band_power(signal=filtered_signal)
return feedback_value
[docs]class RatioBandNFT(PowerBasedNFTModel):
"""
This Neurofeedback model is intended to use the ratio between the power of two frequency bands as feedback value.
Thus, the baseline power parameter is the value of this ratio at calibration stage
"""
[docs] def __init__(self, fs, filter_bank, l_baseline_t, update_feature_window, montage, target_channels, n_cha_lp):
super().__init__(fs=fs, filter_bank=filter_bank, l_baseline_t=l_baseline_t,
update_feature_window=update_feature_window, montage=montage,
target_channels=target_channels, n_cha_lp=n_cha_lp)
self.get_inst('feat_ext_method').mode = 'ratio mode'
[docs] def check_cutoff_settings(self):
if len(self.filter_bank) != 2:
return False
else:
return True
[docs] def calibration(self, eeg, **kwargs):
filtered_signals = self.get_inst('prep_method').fit_transform_signal(signal=eeg,
fs=self.fs)
self.baseline_power = self.get_inst('feat_ext_method').set_baseline(signal=filtered_signals)
[docs] def training(self, eeg, **kwargs):
filtered_signals = self.get_inst('prep_method').transform_signal(signal=eeg)
feedback_value = self.get_inst('feat_ext_method').band_power(signal=filtered_signals)
return feedback_value
[docs]class RestrictionBandNFT(PowerBasedNFTModel):
"""
This Neurofeedback model, as SingleBandNFT, is aimed to enhance the power of a target band. However, it also tries
to keep down the power of other selected frequency bands. Thereby, this training mode ensure that the user is
upregulating only the desired band. That is, this model is a more specific version of SingleBandNFT
"""
[docs] def __init__(self, fs, filter_bank, l_baseline_t, update_feature_window, montage, target_channels, n_cha_lp):
super().__init__(fs=fs, filter_bank=filter_bank, l_baseline_t=l_baseline_t,
update_feature_window=update_feature_window, montage=montage,
target_channels=target_channels, n_cha_lp=n_cha_lp)
self.get_inst('feat_ext_method').mode = 'ban mode'
[docs] def check_cutoff_settings(self):
if len(self.filter_bank) < 2:
return False
else:
return True
[docs] def calibration(self, eeg, **kwargs):
filtered_signals = self.get_inst('prep_method').fit_transform_signal(signal=eeg,
fs=self.fs)
self.baseline_power = self.get_inst('feat_ext_method').set_baseline(signal=filtered_signals)
[docs] def training(self, eeg):
filtered_signals = self.get_inst('prep_method').transform_signal(signal=eeg)
feedback_value = self.get_inst('feat_ext_method').ban_bands(signals=filtered_signals)
return feedback_value
[docs]class NeurofeedbackData(components.ExperimentData):
"""Experiment info class for Neurofeedback training experiments. It records
the important events that take place during a Neurofeedback run, allowing offline analysis."""
[docs] def __init__(self,run_onsets,run_durations,run_success,run_pauses,run_restarts, medusa_nft_app_settings):
self.run_onsets = run_onsets
self.run_durations = run_durations
self.run_success = run_success
self.run_pauses = run_pauses
self.run_restarts = run_restarts
self.medusa_nft_app_settings = medusa_nft_app_settings
[docs] def to_serializable_obj(self):
rec_dict = self.__dict__
for key in rec_dict.keys():
if type(rec_dict[key]) == np.ndarray:
rec_dict[key] = rec_dict[key].tolist()
return rec_dict
[docs] @classmethod
def from_serializable_obj(cls, dict_data):
return cls(**dict_data)