<s> """<STR_LIT>""" <EOL> import numpy as np <EOL> from pyriemann . classification import MDM <EOL> from pyriemann . utils . mean import mean_covariance <EOL> from sklearn . base import BaseEstimator , TransformerMixin <EOL> from multiprocessing import Pool <EOL> from functools import partial <EOL> from mne . event import _find_stim_steps <EOL> def create_sequence ( events ) : <EOL> """<STR_LIT>""" <EOL> sequence = np . zeros ( ( events . shape [ <NUM_LIT:1> ] , <NUM_LIT:1> ) ) <EOL> handStart = np . int64 ( _find_stim_steps ( np . atleast_2d ( events [ <NUM_LIT:0> ] ) , <NUM_LIT:0> ) [ : : <NUM_LIT:2> , <NUM_LIT:0> ] ) <EOL> lift_on = np . int64 ( _find_stim_steps ( np . atleast_2d ( events [ <NUM_LIT:1> ] ) , <NUM_LIT:0> ) [ : : <NUM_LIT:2> , <NUM_LIT:0> ] ) <EOL> lift_off = np . int64 ( _find_stim_steps ( np . atleast_2d ( events [ <NUM_LIT:3> ] ) , <NUM_LIT:0> ) [ <NUM_LIT:1> : : <NUM_LIT:2> , <NUM_LIT:0> ] ) <EOL> replace_on = np . int64 ( _find_stim_steps ( np . atleast_2d ( events [ <NUM_LIT:4> ] ) , <NUM_LIT:0> ) <EOL> [ : : <NUM_LIT:2> , <NUM_LIT:0> ] ) <EOL> replace_off = np . int64 ( _find_stim_steps ( np . atleast_2d ( events [ <NUM_LIT:5> ] ) , <NUM_LIT:0> ) <EOL> [ <NUM_LIT:1> : : <NUM_LIT:2> , <NUM_LIT:0> ] ) <EOL> for i in range ( len ( handStart ) ) : <EOL> j = <NUM_LIT:1> <EOL> sequence [ ( handStart [ i ] - <NUM_LIT> ) : handStart [ i ] ] = j <EOL> j += <NUM_LIT:1> <EOL> sequence [ handStart [ i ] : lift_on [ i ] ] = j <EOL> j += <NUM_LIT:1> <EOL> sequence [ lift_on [ i ] : lift_off [ i ] ] = j <EOL> j += <NUM_LIT:1> <EOL> sequence [ lift_off [ i ] : replace_on [ i ] ] = j <EOL> j += <NUM_LIT:1> <EOL> sequence [ replace_on [ i ] : replace_off [ i ] ] = j <EOL> j += <NUM_LIT:1> <EOL> sequence [ replace_off [ i ] : ( replace_off [ i ] + <NUM_LIT> ) ] = j <EOL> j += <NUM_LIT:1> <EOL> return sequence <EOL> class DistanceCalculatorAlex ( BaseEstimator , TransformerMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , metric_mean = '<STR_LIT>' , metric_dist = [ '<STR_LIT>' ] , <EOL> n_jobs = <NUM_LIT:7> , subsample = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> self . metric_mean = metric_mean <EOL> self . metric_dist = metric_dist <EOL> self . n_jobs = n_jobs <EOL> self . subsample = subsample <EOL> def fit ( self , X , y ) : <EOL> """<STR_LIT>""" <EOL> self . mdm = MDM ( metric = self . metric_mean , n_jobs = self . n_jobs ) <EOL> labels = np . squeeze ( create_sequence ( y . T ) [ : : self . subsample ] ) <EOL> self . mdm . fit ( X , labels ) <EOL> return self <EOL> def transform ( self , X , y = None ) : <EOL> """<STR_LIT>""" <EOL> feattr = [ ] <EOL> for metric in self . metric_dist : <EOL> self . mdm . metric_dist = metric <EOL> feat = self . mdm . transform ( X ) <EOL> feat = feat [ : , <NUM_LIT:1> : ] - np . atleast_2d ( feat [ : , <NUM_LIT:0> ] ) . T <EOL> feattr . append ( feat ) <EOL> feattr = np . concatenate ( feattr , axis = <NUM_LIT:1> ) <EOL> feattr [ np . isnan ( feattr ) ] = <NUM_LIT:0> <EOL> return feattr <EOL> def fit_transform ( self , X , y ) : <EOL> """<STR_LIT>""" <EOL> self . fit ( X , y ) <EOL> return self . transform ( X ) <EOL> class DistanceCalculatorRafal ( BaseEstimator , TransformerMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , metric_mean = '<STR_LIT>' , metric_dist = [ '<STR_LIT>' ] , <EOL> n_jobs = <NUM_LIT:12> , subsample = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> self . metric_mean = metric_mean <EOL> self . metric_dist = metric_dist <EOL> self . n_jobs = n_jobs <EOL> self . subsample = subsample <EOL> def fit ( self , X , y ) : <EOL> """<STR_LIT>""" <EOL> self . mdm = MDM ( metric = self . metric_mean , n_jobs = self . n_jobs ) <EOL> labels = y [ : : self . subsample ] <EOL> pCalcMeans = partial ( mean_covariance , metric = self . metric_mean ) <EOL> pool = Pool ( processes = <NUM_LIT:6> ) <EOL> mc1 = pool . map ( pCalcMeans , [ X [ labels [ : , i ] == <NUM_LIT:1> ] for i in range ( <NUM_LIT:6> ) ] ) <EOL> pool . close ( ) <EOL> pool = Pool ( processes = <NUM_LIT:6> ) <EOL> mc0 = pool . map ( pCalcMeans , [ X [ labels [ : , i ] == <NUM_LIT:0> ] for i in range ( <NUM_LIT:6> ) ] ) <EOL> pool . close ( ) <EOL> self . mdm . covmeans = mc1 + mc0 <EOL> return self <EOL> def transform ( self , X , y = None ) : <EOL> """<STR_LIT>""" <EOL> feattr = [ ] <EOL> for metric in self . metric_dist : <EOL> self . mdm . metric_dist = metric <EOL> feat = self . mdm . transform ( X ) <EOL> feat = feat [ : , <NUM_LIT:0> : <NUM_LIT:6> ] - feat [ : , <NUM_LIT:6> : ] <EOL> feattr . append ( feat ) <EOL> feattr = np . concatenate ( feattr , axis = <NUM_LIT:1> ) <EOL> feattr [ np . isnan ( feattr ) ] = <NUM_LIT:0> <EOL> return feattr <EOL> def fit_transform ( self , X , y ) : <EOL> """<STR_LIT>""" <EOL> self . fit ( X , y ) <EOL> return self . transform ( X ) </s>
