<s> import copy <EOL> import os <EOL> from math import ceil <EOL> import warnings <EOL> import numpy as np <EOL> from scipy import linalg , sparse <EOL> from scipy . sparse import coo_matrix <EOL> from . filter import resample <EOL> from . evoked import _get_peak <EOL> from . parallel import parallel_func <EOL> from . surface import ( read_surface , _get_ico_surface , read_morph_map , <EOL> _compute_nearest , mesh_edges ) <EOL> from . source_space import ( _ensure_src , _get_morph_src_reordering , <EOL> _ensure_src_subject ) <EOL> from . utils import ( get_subjects_dir , _check_subject , logger , verbose , <EOL> _time_mask , warn as warn_ ) <EOL> from . viz import plot_source_estimates <EOL> from . fixes import in1d , sparse_block_diag <EOL> from . io . base import ToDataFrameMixin , TimeMixin <EOL> from . externals . six . moves import zip <EOL> from . externals . six import string_types <EOL> from . externals . h5io import read_hdf5 , write_hdf5 <EOL> def _read_stc ( filename ) : <EOL> """<STR_LIT>""" <EOL> fid = open ( filename , '<STR_LIT:rb>' ) <EOL> stc = dict ( ) <EOL> fid . seek ( <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> file_length = fid . tell ( ) <EOL> fid . seek ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> stc [ '<STR_LIT>' ] = float ( np . fromfile ( fid , dtype = "<STR_LIT>" , count = <NUM_LIT:1> ) ) <EOL> stc [ '<STR_LIT>' ] /= <NUM_LIT> <EOL> stc [ '<STR_LIT>' ] = float ( np . fromfile ( fid , dtype = "<STR_LIT>" , count = <NUM_LIT:1> ) ) <EOL> stc [ '<STR_LIT>' ] /= <NUM_LIT> <EOL> vertices_n = int ( np . fromfile ( fid , dtype = "<STR_LIT>" , count = <NUM_LIT:1> ) ) <EOL> stc [ '<STR_LIT>' ] = np . fromfile ( fid , dtype = "<STR_LIT>" , count = vertices_n ) <EOL> data_n = int ( np . fromfile ( fid , dtype = "<STR_LIT>" , count = <NUM_LIT:1> ) ) <EOL> if ( vertices_n and <EOL> ( ( file_length / <NUM_LIT:4> - <NUM_LIT:4> - vertices_n ) % ( data_n * vertices_n ) ) != <NUM_LIT:0> ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> stc [ '<STR_LIT:data>' ] = np . fromfile ( fid , dtype = "<STR_LIT>" , count = vertices_n * data_n ) <EOL> stc [ '<STR_LIT:data>' ] = stc [ '<STR_LIT:data>' ] . reshape ( [ data_n , vertices_n ] ) . T <EOL> fid . close ( ) <EOL> return stc <EOL> def _write_stc ( filename , tmin , tstep , vertices , data ) : <EOL> """<STR_LIT>""" <EOL> fid = open ( filename , '<STR_LIT:wb>' ) <EOL> fid . write ( np . array ( <NUM_LIT:1000> * tmin , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . write ( np . array ( <NUM_LIT:1000> * tstep , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . write ( np . array ( vertices . shape [ <NUM_LIT:0> ] , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . write ( np . array ( vertices , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . write ( np . array ( data . shape [ <NUM_LIT:1> ] , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . write ( np . array ( data . T , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . close ( ) <EOL> def _read_3 ( fid ) : <EOL> """<STR_LIT>""" <EOL> data = np . fromfile ( fid , dtype = np . uint8 , count = <NUM_LIT:3> ) . astype ( np . int32 ) <EOL> out = np . left_shift ( data [ <NUM_LIT:0> ] , <NUM_LIT:16> ) + np . left_shift ( data [ <NUM_LIT:1> ] , <NUM_LIT:8> ) + data [ <NUM_LIT:2> ] <EOL> return out <EOL> def _read_w ( filename ) : <EOL> """<STR_LIT>""" <EOL> with open ( filename , '<STR_LIT:rb>' , buffering = <NUM_LIT:0> ) as fid : <EOL> fid . read ( <NUM_LIT:2> ) <EOL> vertices_n = int ( _read_3 ( fid ) ) <EOL> vertices = np . zeros ( ( vertices_n ) , dtype = np . int32 ) <EOL> data = np . zeros ( ( vertices_n ) , dtype = np . float32 ) <EOL> for i in range ( vertices_n ) : <EOL> vertices [ i ] = _read_3 ( fid ) <EOL> data [ i ] = np . fromfile ( fid , dtype = '<STR_LIT>' , count = <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> w = dict ( ) <EOL> w [ '<STR_LIT>' ] = vertices <EOL> w [ '<STR_LIT:data>' ] = data <EOL> return w <EOL> def _write_3 ( fid , val ) : <EOL> """<STR_LIT>""" <EOL> f_bytes = np . zeros ( ( <NUM_LIT:3> ) , dtype = np . uint8 ) <EOL> f_bytes [ <NUM_LIT:0> ] = ( val >> <NUM_LIT:16> ) & <NUM_LIT:255> <EOL> f_bytes [ <NUM_LIT:1> ] = ( val >> <NUM_LIT:8> ) & <NUM_LIT:255> <EOL> f_bytes [ <NUM_LIT:2> ] = val & <NUM_LIT:255> <EOL> fid . write ( f_bytes . tostring ( ) ) <EOL> def _write_w ( filename , vertices , data ) : <EOL> """<STR_LIT>""" <EOL> assert ( len ( vertices ) == len ( data ) ) <EOL> fid = open ( filename , '<STR_LIT:wb>' ) <EOL> fid . write ( np . zeros ( ( <NUM_LIT:2> ) , dtype = np . uint8 ) . tostring ( ) ) <EOL> vertices_n = len ( vertices ) <EOL> _write_3 ( fid , vertices_n ) <EOL> for i in range ( vertices_n ) : <EOL> _write_3 ( fid , vertices [ i ] ) <EOL> fid . write ( np . array ( float ( data [ i ] ) , dtype = '<STR_LIT>' ) . tostring ( ) ) <EOL> fid . close ( ) <EOL> def read_source_estimate ( fname , subject = None ) : <EOL> """<STR_LIT>""" <EOL> fname_arg = fname <EOL> ftype = None <EOL> if os . path . exists ( fname ) : <EOL> if fname . endswith ( '<STR_LIT>' ) or fname . endswith ( '<STR_LIT>' ) or fname . endswith ( '<STR_LIT>' ) or fname . endswith ( '<STR_LIT>' ) : <EOL> ftype = '<STR_LIT>' <EOL> elif fname . endswith ( '<STR_LIT>' ) : <EOL> ftype = '<STR_LIT>' <EOL> if fname . endswith ( ( '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> fname = fname [ : - <NUM_LIT:7> ] <EOL> else : <EOL> err = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % fname ) <EOL> raise IOError ( err ) <EOL> elif fname . endswith ( '<STR_LIT>' ) : <EOL> ftype = '<STR_LIT:w>' <EOL> if fname . endswith ( ( '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> fname = fname [ : - <NUM_LIT:5> ] <EOL> else : <EOL> err = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % fname ) <EOL> raise IOError ( err ) <EOL> elif fname . endswith ( '<STR_LIT>' ) : <EOL> ftype = '<STR_LIT>' <EOL> fname = fname [ : - <NUM_LIT:7> ] <EOL> else : <EOL> raise RuntimeError ( '<STR_LIT>' % fname_arg ) <EOL> if ftype is not '<STR_LIT>' : <EOL> stc_exist = [ os . path . exists ( f ) <EOL> for f in [ fname + '<STR_LIT>' , fname + '<STR_LIT>' ] ] <EOL> w_exist = [ os . path . exists ( f ) <EOL> for f in [ fname + '<STR_LIT>' , fname + '<STR_LIT>' ] ] <EOL> h5_exist = os . path . exists ( fname + '<STR_LIT>' ) <EOL> if all ( stc_exist ) and ( ftype is not '<STR_LIT:w>' ) : <EOL> ftype = '<STR_LIT>' <EOL> elif all ( w_exist ) : <EOL> ftype = '<STR_LIT:w>' <EOL> elif h5_exist : <EOL> ftype = '<STR_LIT>' <EOL> elif any ( stc_exist ) or any ( w_exist ) : <EOL> raise IOError ( "<STR_LIT>" % fname_arg ) <EOL> else : <EOL> raise IOError ( "<STR_LIT>" <EOL> % fname_arg ) <EOL> if ftype == '<STR_LIT>' : <EOL> if fname . endswith ( '<STR_LIT>' ) : <EOL> kwargs = _read_stc ( fname ) <EOL> elif fname . endswith ( '<STR_LIT>' ) : <EOL> kwargs = _read_w ( fname ) <EOL> kwargs [ '<STR_LIT:data>' ] = kwargs [ '<STR_LIT:data>' ] [ : , np . newaxis ] <EOL> kwargs [ '<STR_LIT>' ] = <NUM_LIT:0.0> <EOL> kwargs [ '<STR_LIT>' ] = <NUM_LIT:0.0> <EOL> else : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> elif ftype == '<STR_LIT>' : <EOL> lh = _read_stc ( fname + '<STR_LIT>' ) <EOL> rh = _read_stc ( fname + '<STR_LIT>' ) <EOL> assert lh [ '<STR_LIT>' ] == rh [ '<STR_LIT>' ] <EOL> assert lh [ '<STR_LIT>' ] == rh [ '<STR_LIT>' ] <EOL> kwargs = lh . copy ( ) <EOL> kwargs [ '<STR_LIT:data>' ] = np . r_ [ lh [ '<STR_LIT:data>' ] , rh [ '<STR_LIT:data>' ] ] <EOL> kwargs [ '<STR_LIT>' ] = [ lh [ '<STR_LIT>' ] , rh [ '<STR_LIT>' ] ] <EOL> elif ftype == '<STR_LIT:w>' : <EOL> lh = _read_w ( fname + '<STR_LIT>' ) <EOL> rh = _read_w ( fname + '<STR_LIT>' ) <EOL> kwargs = lh . copy ( ) <EOL> kwargs [ '<STR_LIT:data>' ] = np . atleast_2d ( np . r_ [ lh [ '<STR_LIT:data>' ] , rh [ '<STR_LIT:data>' ] ] ) . T <EOL> kwargs [ '<STR_LIT>' ] = [ lh [ '<STR_LIT>' ] , rh [ '<STR_LIT>' ] ] <EOL> kwargs [ '<STR_LIT>' ] = <NUM_LIT:0.0> <EOL> kwargs [ '<STR_LIT>' ] = <NUM_LIT:1.0> <EOL> elif ftype == '<STR_LIT>' : <EOL> kwargs = read_hdf5 ( fname + '<STR_LIT>' , title = '<STR_LIT>' ) <EOL> if ftype != '<STR_LIT>' : <EOL> vertices = kwargs [ '<STR_LIT>' ] <EOL> if any ( np . any ( np . diff ( v . astype ( int ) ) <= <NUM_LIT:0> ) for v in vertices ) : <EOL> sidx = [ np . argsort ( verts ) for verts in vertices ] <EOL> vertices = [ verts [ idx ] for verts , idx in zip ( vertices , sidx ) ] <EOL> data = kwargs [ '<STR_LIT:data>' ] [ np . r_ [ sidx [ <NUM_LIT:0> ] , len ( sidx [ <NUM_LIT:0> ] ) + sidx [ <NUM_LIT:1> ] ] ] <EOL> kwargs [ '<STR_LIT>' ] = vertices <EOL> kwargs [ '<STR_LIT:data>' ] = data <EOL> if '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = subject <EOL> if subject is not None and subject != kwargs [ '<STR_LIT>' ] : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( subject , kwargs [ '<STR_LIT>' ] ) ) <EOL> if ftype == '<STR_LIT>' : <EOL> stc = VolSourceEstimate ( ** kwargs ) <EOL> else : <EOL> stc = SourceEstimate ( ** kwargs ) <EOL> return stc <EOL> def _make_stc ( data , vertices , tmin = None , tstep = None , subject = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( vertices , list ) and len ( vertices ) == <NUM_LIT:2> : <EOL> stc = SourceEstimate ( data , vertices = vertices , tmin = tmin , tstep = tstep , <EOL> subject = subject ) <EOL> elif isinstance ( vertices , np . ndarray ) or isinstance ( vertices , list ) and len ( vertices ) == <NUM_LIT:1> : <EOL> stc = VolSourceEstimate ( data , vertices = vertices , tmin = tmin , <EOL> tstep = tstep , subject = subject ) <EOL> elif isinstance ( vertices , list ) and len ( vertices ) > <NUM_LIT:2> : <EOL> stc = MixedSourceEstimate ( data , vertices = vertices , tmin = tmin , <EOL> tstep = tstep , subject = subject ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return stc <EOL> def _verify_source_estimate_compat ( a , b ) : <EOL> """<STR_LIT>""" <EOL> compat = False <EOL> if len ( a . vertices ) == len ( b . vertices ) : <EOL> if all ( np . array_equal ( av , vv ) <EOL> for av , vv in zip ( a . vertices , b . vertices ) ) : <EOL> compat = True <EOL> if not compat : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if a . subject != b . subject : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( a . subject , b . subject ) ) <EOL> class _BaseSourceEstimate ( ToDataFrameMixin , TimeMixin ) : <EOL> """<STR_LIT>""" <EOL> @ verbose <EOL> def __init__ ( self , data , vertices = None , tmin = None , tstep = None , <EOL> subject = None , verbose = None ) : <EOL> kernel , sens_data = None , None <EOL> if isinstance ( data , tuple ) : <EOL> if len ( data ) != <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> kernel , sens_data = data <EOL> data = None <EOL> if kernel . shape [ <NUM_LIT:1> ] != sens_data . shape [ <NUM_LIT:0> ] : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if isinstance ( vertices , list ) : <EOL> if not all ( isinstance ( v , np . ndarray ) for v in vertices ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if any ( np . any ( np . diff ( v . astype ( int ) ) <= <NUM_LIT:0> ) for v in vertices ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> n_src = sum ( [ len ( v ) for v in vertices ] ) <EOL> if len ( vertices ) == <NUM_LIT:1> : <EOL> vertices = vertices [ <NUM_LIT:0> ] <EOL> elif isinstance ( vertices , np . ndarray ) : <EOL> n_src = len ( vertices ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if data is not None and data . shape [ <NUM_LIT:0> ] != n_src : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( n_src , data . shape [ <NUM_LIT:0> ] ) ) <EOL> self . _data = data <EOL> self . tmin = tmin <EOL> self . tstep = tstep <EOL> self . vertices = vertices <EOL> self . verbose = verbose <EOL> self . _kernel = kernel <EOL> self . _sens_data = sens_data <EOL> self . _kernel_removed = False <EOL> self . times = None <EOL> self . _update_times ( ) <EOL> self . subject = _check_subject ( None , subject , False ) <EOL> @ property <EOL> def sfreq ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:1.> / self . tstep <EOL> def _remove_kernel_sens_data_ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _kernel is not None or self . _sens_data is not None : <EOL> self . _kernel_removed = True <EOL> self . _data = np . dot ( self . _kernel , self . _sens_data ) <EOL> self . _kernel = None <EOL> self . _sens_data = None <EOL> def crop ( self , tmin = None , tmax = None ) : <EOL> """<STR_LIT>""" <EOL> mask = _time_mask ( self . times , tmin , tmax , sfreq = self . sfreq ) <EOL> self . tmin = self . times [ np . where ( mask ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ] <EOL> if self . _kernel is not None and self . _sens_data is not None : <EOL> self . _sens_data = self . _sens_data [ : , mask ] <EOL> else : <EOL> self . _data = self . _data [ : , mask ] <EOL> self . _update_times ( ) <EOL> return self <EOL> @ verbose <EOL> def resample ( self , sfreq , npad = None , window = '<STR_LIT>' , n_jobs = <NUM_LIT:1> , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if npad is None : <EOL> npad = <NUM_LIT:100> <EOL> warn_ ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> DeprecationWarning ) <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> o_sfreq = <NUM_LIT:1.0> / self . tstep <EOL> self . _data = resample ( self . _data , sfreq , o_sfreq , npad , n_jobs = n_jobs ) <EOL> self . tstep = <NUM_LIT:1.0> / sfreq <EOL> self . _update_times ( ) <EOL> @ property <EOL> def data ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _data is None : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> return self . _data <EOL> @ property <EOL> def shape ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _data is not None : <EOL> return self . _data . shape <EOL> return ( self . _kernel . shape [ <NUM_LIT:0> ] , self . _sens_data . shape [ <NUM_LIT:1> ] ) <EOL> def _update_times ( self ) : <EOL> """<STR_LIT>""" <EOL> self . times = self . tmin + ( self . tstep * np . arange ( self . shape [ <NUM_LIT:1> ] ) ) <EOL> def __add__ ( self , a ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc += a <EOL> return stc <EOL> def __iadd__ ( self , a ) : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> if isinstance ( a , _BaseSourceEstimate ) : <EOL> _verify_source_estimate_compat ( self , a ) <EOL> self . _data += a . data <EOL> else : <EOL> self . _data += a <EOL> return self <EOL> def mean ( self ) : <EOL> """<STR_LIT>""" <EOL> data = self . data <EOL> tmax = self . tmin + self . tstep * data . shape [ <NUM_LIT:1> ] <EOL> tmin = ( self . tmin + tmax ) / <NUM_LIT> <EOL> tstep = tmax - self . tmin <EOL> mean_stc = SourceEstimate ( self . data . mean ( axis = <NUM_LIT:1> ) [ : , np . newaxis ] , <EOL> vertices = self . vertices , tmin = tmin , <EOL> tstep = tstep , subject = self . subject ) <EOL> return mean_stc <EOL> def __sub__ ( self , a ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc -= a <EOL> return stc <EOL> def __isub__ ( self , a ) : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> if isinstance ( a , _BaseSourceEstimate ) : <EOL> _verify_source_estimate_compat ( self , a ) <EOL> self . _data -= a . data <EOL> else : <EOL> self . _data -= a <EOL> return self <EOL> def __truediv__ ( self , a ) : <EOL> return self . __div__ ( a ) <EOL> def __div__ ( self , a ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc /= a <EOL> return stc <EOL> def __itruediv__ ( self , a ) : <EOL> return self . __idiv__ ( a ) <EOL> def __idiv__ ( self , a ) : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> if isinstance ( a , _BaseSourceEstimate ) : <EOL> _verify_source_estimate_compat ( self , a ) <EOL> self . _data /= a . data <EOL> else : <EOL> self . _data /= a <EOL> return self <EOL> def __mul__ ( self , a ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc *= a <EOL> return stc <EOL> def __imul__ ( self , a ) : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> if isinstance ( a , _BaseSourceEstimate ) : <EOL> _verify_source_estimate_compat ( self , a ) <EOL> self . _data *= a . data <EOL> else : <EOL> self . _data *= a <EOL> return self <EOL> def __pow__ ( self , a ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc **= a <EOL> return stc <EOL> def __ipow__ ( self , a ) : <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> self . _data **= a <EOL> return self <EOL> def __radd__ ( self , a ) : <EOL> return self + a <EOL> def __rsub__ ( self , a ) : <EOL> return self - a <EOL> def __rmul__ ( self , a ) : <EOL> return self * a <EOL> def __rdiv__ ( self , a ) : <EOL> return self / a <EOL> def __neg__ ( self ) : <EOL> stc = copy . deepcopy ( self ) <EOL> stc . _remove_kernel_sens_data_ ( ) <EOL> stc . _data *= - <NUM_LIT:1> <EOL> return stc <EOL> def __pos__ ( self ) : <EOL> return self <EOL> def sqrt ( self ) : <EOL> """<STR_LIT>""" <EOL> return self ** ( <NUM_LIT:0.5> ) <EOL> def copy ( self ) : <EOL> """<STR_LIT>""" <EOL> return copy . deepcopy ( self ) <EOL> def bin ( self , width , tstart = None , tstop = None , func = np . mean ) : <EOL> """<STR_LIT>""" <EOL> if tstart is None : <EOL> tstart = self . tmin <EOL> if tstop is None : <EOL> tstop = self . times [ - <NUM_LIT:1> ] <EOL> times = np . arange ( tstart , tstop + self . tstep , width ) <EOL> nv , _ = self . shape <EOL> nt = len ( times ) - <NUM_LIT:1> <EOL> data = np . empty ( ( nv , nt ) , dtype = self . data . dtype ) <EOL> for i in range ( nt ) : <EOL> idx = ( self . times >= times [ i ] ) & ( self . times < times [ i + <NUM_LIT:1> ] ) <EOL> data [ : , i ] = func ( self . data [ : , idx ] , axis = <NUM_LIT:1> ) <EOL> tmin = times [ <NUM_LIT:0> ] + width / <NUM_LIT> <EOL> stc = _make_stc ( data , vertices = self . vertices , <EOL> tmin = tmin , tstep = width , subject = self . subject ) <EOL> return stc <EOL> def transform_data ( self , func , idx = None , tmin_idx = None , tmax_idx = None ) : <EOL> """<STR_LIT>""" <EOL> if idx is None : <EOL> idx = slice ( None , None ) <EOL> if self . _kernel is None and self . _sens_data is None : <EOL> if self . _kernel_removed : <EOL> warn_ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> data_t = func ( self . data [ idx , tmin_idx : tmax_idx ] ) <EOL> if isinstance ( data_t , tuple ) : <EOL> data_t = data_t [ <NUM_LIT:0> ] <EOL> else : <EOL> sens_data_t = func ( self . _sens_data [ : , tmin_idx : tmax_idx ] ) <EOL> if isinstance ( sens_data_t , tuple ) : <EOL> sens_data_t = sens_data_t [ <NUM_LIT:0> ] <EOL> data_shape = sens_data_t . shape <EOL> if len ( data_shape ) > <NUM_LIT:2> : <EOL> sens_data_t = sens_data_t . reshape ( data_shape [ <NUM_LIT:0> ] , <EOL> np . prod ( data_shape [ <NUM_LIT:1> : ] ) ) <EOL> data_t = np . dot ( self . _kernel [ idx , : ] , sens_data_t ) <EOL> if len ( data_shape ) > <NUM_LIT:2> : <EOL> data_t = data_t . reshape ( data_t . shape [ <NUM_LIT:0> ] , * data_shape [ <NUM_LIT:1> : ] ) <EOL> return data_t <EOL> def transform ( self , func , idx = None , tmin = None , tmax = None , copy = False ) : <EOL> """<STR_LIT>""" <EOL> times = <NUM_LIT> * self . times <EOL> t_idx = np . where ( _time_mask ( times , tmin , tmax , sfreq = self . sfreq ) ) [ <NUM_LIT:0> ] <EOL> if tmin is None : <EOL> tmin_idx = None <EOL> else : <EOL> tmin_idx = t_idx [ <NUM_LIT:0> ] <EOL> if tmax is None : <EOL> tmax_idx = None <EOL> else : <EOL> tmax_idx = t_idx [ - <NUM_LIT:1> ] <EOL> data_t = self . transform_data ( func , idx = idx , tmin_idx = tmin_idx , <EOL> tmax_idx = tmax_idx ) <EOL> if idx is not None : <EOL> idx_lh = idx [ idx < len ( self . lh_vertno ) ] <EOL> idx_rh = idx [ idx >= len ( self . lh_vertno ) ] - len ( self . lh_vertno ) <EOL> verts_lh = self . lh_vertno [ idx_lh ] <EOL> verts_rh = self . rh_vertno [ idx_rh ] <EOL> else : <EOL> verts_lh = self . lh_vertno <EOL> verts_rh = self . rh_vertno <EOL> verts = [ verts_lh , verts_rh ] <EOL> tmin_idx = <NUM_LIT:0> if tmin_idx is None else tmin_idx <EOL> tmax_idx = - <NUM_LIT:1> if tmax_idx is None else tmax_idx <EOL> tmin = self . times [ tmin_idx ] <EOL> times = np . arange ( self . times [ tmin_idx ] , <EOL> self . times [ tmax_idx ] + self . tstep / <NUM_LIT:2> , self . tstep ) <EOL> if data_t . ndim > <NUM_LIT:2> : <EOL> if copy : <EOL> stcs = [ SourceEstimate ( data_t [ : , : , a ] , verts , tmin , <EOL> self . tstep , self . subject ) <EOL> for a in range ( data_t . shape [ - <NUM_LIT:1> ] ) ] <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> stcs = self if not copy else self . copy ( ) <EOL> stcs . _data , stcs . vertices = data_t , verts <EOL> stcs . tmin , stcs . times = tmin , times <EOL> return stcs <EOL> class SourceEstimate ( _BaseSourceEstimate ) : <EOL> """<STR_LIT>""" <EOL> @ verbose <EOL> def __init__ ( self , data , vertices = None , tmin = None , tstep = None , <EOL> subject = None , verbose = None ) : <EOL> if not ( isinstance ( vertices , list ) and len ( vertices ) == <NUM_LIT:2> ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _BaseSourceEstimate . __init__ ( self , data , vertices = vertices , tmin = tmin , <EOL> tstep = tstep , subject = subject , <EOL> verbose = verbose ) <EOL> @ verbose <EOL> def save ( self , fname , ftype = '<STR_LIT>' , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if ftype not in ( '<STR_LIT>' , '<STR_LIT:w>' , '<STR_LIT>' ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> % ftype ) <EOL> lh_data = self . data [ : len ( self . lh_vertno ) ] <EOL> rh_data = self . data [ - len ( self . rh_vertno ) : ] <EOL> if ftype == '<STR_LIT>' : <EOL> logger . info ( '<STR_LIT>' ) <EOL> _write_stc ( fname + '<STR_LIT>' , tmin = self . tmin , tstep = self . tstep , <EOL> vertices = self . lh_vertno , data = lh_data ) <EOL> _write_stc ( fname + '<STR_LIT>' , tmin = self . tmin , tstep = self . tstep , <EOL> vertices = self . rh_vertno , data = rh_data ) <EOL> elif ftype == '<STR_LIT:w>' : <EOL> if self . shape [ <NUM_LIT:1> ] != <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> _write_w ( fname + '<STR_LIT>' , vertices = self . lh_vertno , <EOL> data = lh_data [ : , <NUM_LIT:0> ] ) <EOL> _write_w ( fname + '<STR_LIT>' , vertices = self . rh_vertno , <EOL> data = rh_data [ : , <NUM_LIT:0> ] ) <EOL> elif ftype == '<STR_LIT>' : <EOL> write_hdf5 ( fname + '<STR_LIT>' , <EOL> dict ( vertices = self . vertices , data = self . data , <EOL> tmin = self . tmin , tstep = self . tstep , <EOL> subject = self . subject ) , title = '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> def __repr__ ( self ) : <EOL> if isinstance ( self . vertices , list ) : <EOL> nv = sum ( [ len ( v ) for v in self . vertices ] ) <EOL> else : <EOL> nv = self . vertices . size <EOL> s = "<STR_LIT>" % nv <EOL> if self . subject is not None : <EOL> s += "<STR_LIT>" % self . subject <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . tmin ) <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . times [ - <NUM_LIT:1> ] ) <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . tstep ) <EOL> s += "<STR_LIT>" % self . shape <EOL> return "<STR_LIT>" % s <EOL> @ property <EOL> def lh_data ( self ) : <EOL> return self . data [ : len ( self . lh_vertno ) ] <EOL> @ property <EOL> def rh_data ( self ) : <EOL> return self . data [ len ( self . lh_vertno ) : ] <EOL> @ property <EOL> def lh_vertno ( self ) : <EOL> return self . vertices [ <NUM_LIT:0> ] <EOL> @ property <EOL> def rh_vertno ( self ) : <EOL> return self . vertices [ <NUM_LIT:1> ] <EOL> def _hemilabel_stc ( self , label ) : <EOL> if label . hemi == '<STR_LIT>' : <EOL> stc_vertices = self . vertices [ <NUM_LIT:0> ] <EOL> else : <EOL> stc_vertices = self . vertices [ <NUM_LIT:1> ] <EOL> idx = np . nonzero ( in1d ( stc_vertices , label . vertices ) ) [ <NUM_LIT:0> ] <EOL> vertices = stc_vertices [ idx ] <EOL> if label . hemi == '<STR_LIT>' : <EOL> values = self . data [ idx + len ( self . vertices [ <NUM_LIT:0> ] ) ] <EOL> else : <EOL> values = self . data [ idx ] <EOL> return vertices , values <EOL> def in_label ( self , label ) : <EOL> """<STR_LIT>""" <EOL> if label . subject is not None and self . subject is not None and label . subject != self . subject : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( label . subject , <EOL> self . subject ) ) <EOL> if label . hemi == '<STR_LIT>' : <EOL> lh_vert , lh_val = self . _hemilabel_stc ( label . lh ) <EOL> rh_vert , rh_val = self . _hemilabel_stc ( label . rh ) <EOL> vertices = [ lh_vert , rh_vert ] <EOL> values = np . vstack ( ( lh_val , rh_val ) ) <EOL> elif label . hemi == '<STR_LIT>' : <EOL> lh_vert , values = self . _hemilabel_stc ( label ) <EOL> vertices = [ lh_vert , np . array ( [ ] , int ) ] <EOL> elif label . hemi == '<STR_LIT>' : <EOL> rh_vert , values = self . _hemilabel_stc ( label ) <EOL> vertices = [ np . array ( [ ] , int ) , rh_vert ] <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % label ) <EOL> if sum ( [ len ( v ) for v in vertices ] ) == <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> label_stc = SourceEstimate ( values , vertices = vertices , <EOL> tmin = self . tmin , tstep = self . tstep , <EOL> subject = self . subject ) <EOL> return label_stc <EOL> def expand ( self , vertices ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( vertices , list ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if not len ( self . vertices ) == len ( vertices ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _remove_kernel_sens_data_ ( ) <EOL> inserters = list ( ) <EOL> offsets = [ <NUM_LIT:0> ] <EOL> for vi , ( v_old , v_new ) in enumerate ( zip ( self . vertices , vertices ) ) : <EOL> v_new = np . setdiff1d ( v_new , v_old ) <EOL> inds = np . searchsorted ( v_old , v_new ) <EOL> inserters += [ inds . copy ( ) ] <EOL> offsets += [ len ( v_old ) ] <EOL> self . vertices [ vi ] = np . insert ( v_old , inds , v_new ) <EOL> inds = [ ii + offset for ii , offset in zip ( inserters , offsets [ : - <NUM_LIT:1> ] ) ] <EOL> inds = np . concatenate ( inds ) <EOL> new_data = np . zeros ( ( len ( inds ) , self . _data . shape [ <NUM_LIT:1> ] ) ) <EOL> self . _data = np . insert ( self . _data , inds , new_data , axis = <NUM_LIT:0> ) <EOL> return self <EOL> @ verbose <EOL> def extract_label_time_course ( self , labels , src , mode = '<STR_LIT>' , <EOL> allow_empty = False , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> label_tc = extract_label_time_course ( self , labels , src , mode = mode , <EOL> return_generator = False , <EOL> allow_empty = allow_empty , <EOL> verbose = verbose ) <EOL> return label_tc <EOL> def center_of_mass ( self , subject = None , hemi = None , restrict_vertices = False , <EOL> subjects_dir = None ) : <EOL> """<STR_LIT>""" <EOL> subject = _check_subject ( self . subject , subject ) <EOL> values = np . sum ( self . data , axis = <NUM_LIT:1> ) <EOL> vert_inds = [ np . arange ( len ( self . vertices [ <NUM_LIT:0> ] ) ) , <EOL> np . arange ( len ( self . vertices [ <NUM_LIT:1> ] ) ) + len ( self . vertices [ <NUM_LIT:0> ] ) ] <EOL> if hemi is None : <EOL> hemi = np . where ( np . array ( [ np . sum ( values [ vi ] ) <EOL> for vi in vert_inds ] ) ) [ <NUM_LIT:0> ] <EOL> if not len ( hemi ) == <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> hemi = hemi [ <NUM_LIT:0> ] <EOL> if hemi not in [ <NUM_LIT:0> , <NUM_LIT:1> ] : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> subjects_dir = get_subjects_dir ( subjects_dir , raise_error = True ) <EOL> values = values [ vert_inds [ hemi ] ] <EOL> hemis = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> surf = os . path . join ( subjects_dir , subject , '<STR_LIT>' , <EOL> hemis [ hemi ] + '<STR_LIT>' ) <EOL> if isinstance ( surf , string_types ) : <EOL> surf = read_surface ( surf ) <EOL> if restrict_vertices is False : <EOL> restrict_vertices = np . arange ( surf [ <NUM_LIT:0> ] . shape [ <NUM_LIT:0> ] ) <EOL> elif restrict_vertices is True : <EOL> restrict_vertices = self . vertices [ hemi ] <EOL> if np . any ( self . data < <NUM_LIT:0> ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> pos = surf [ <NUM_LIT:0> ] [ self . vertices [ hemi ] , : ] . T <EOL> c_o_m = np . sum ( pos * values , axis = <NUM_LIT:1> ) / np . sum ( values ) <EOL> vertex = np . argmin ( np . sqrt ( np . mean ( ( surf [ <NUM_LIT:0> ] [ restrict_vertices , : ] - <EOL> c_o_m ) ** <NUM_LIT:2> , axis = <NUM_LIT:1> ) ) ) <EOL> vertex = restrict_vertices [ vertex ] <EOL> masses = np . sum ( self . data , axis = <NUM_LIT:0> ) . astype ( float ) <EOL> t_ind = np . sum ( masses * np . arange ( self . shape [ <NUM_LIT:1> ] ) ) / np . sum ( masses ) <EOL> t = self . tmin + self . tstep * t_ind <EOL> return vertex , hemi , t <EOL> def plot ( self , subject = None , surface = '<STR_LIT>' , hemi = '<STR_LIT>' , <EOL> colormap = '<STR_LIT>' , time_label = '<STR_LIT>' , <EOL> smoothing_steps = <NUM_LIT:10> , transparent = None , alpha = <NUM_LIT:1.0> , <EOL> time_viewer = False , config_opts = None , subjects_dir = None , <EOL> figure = None , views = '<STR_LIT>' , colorbar = True , clim = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> brain = plot_source_estimates ( self , subject , surface = surface , <EOL> hemi = hemi , colormap = colormap , <EOL> time_label = time_label , <EOL> smoothing_steps = smoothing_steps , <EOL> transparent = transparent , alpha = alpha , <EOL> time_viewer = time_viewer , <EOL> config_opts = config_opts , <EOL> subjects_dir = subjects_dir , figure = figure , <EOL> views = views , colorbar = colorbar , <EOL> clim = clim ) <EOL> return brain <EOL> @ verbose <EOL> def to_original_src ( self , src_orig , subject_orig = None , <EOL> subjects_dir = None , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if self . subject is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> src_orig = _ensure_src ( src_orig , kind = '<STR_LIT>' ) <EOL> subject_orig = _ensure_src_subject ( src_orig , subject_orig ) <EOL> data_idx , vertices = _get_morph_src_reordering ( <EOL> self . vertices , src_orig , subject_orig , self . subject , subjects_dir ) <EOL> return SourceEstimate ( self . _data [ data_idx ] , vertices , <EOL> self . tmin , self . tstep , subject_orig ) <EOL> @ verbose <EOL> def morph ( self , subject_to , grade = <NUM_LIT:5> , smooth = None , subjects_dir = None , <EOL> buffer_size = <NUM_LIT:64> , n_jobs = <NUM_LIT:1> , subject_from = None , sparse = False , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> subject_from = _check_subject ( self . subject , subject_from ) <EOL> if sparse : <EOL> if grade is not None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return _morph_sparse ( self , subject_from , subject_to , subjects_dir ) <EOL> else : <EOL> return morph_data ( subject_from , subject_to , self , grade , smooth , <EOL> subjects_dir , buffer_size , n_jobs , verbose ) <EOL> def morph_precomputed ( self , subject_to , vertices_to , morph_mat , <EOL> subject_from = None ) : <EOL> """<STR_LIT>""" <EOL> subject_from = _check_subject ( self . subject , subject_from ) <EOL> return morph_data_precomputed ( subject_from , subject_to , self , <EOL> vertices_to , morph_mat ) <EOL> def get_peak ( self , hemi = None , tmin = None , tmax = None , mode = '<STR_LIT>' , <EOL> vert_as_index = False , time_as_index = False ) : <EOL> """<STR_LIT>""" <EOL> data = { '<STR_LIT>' : self . lh_data , '<STR_LIT>' : self . rh_data , None : self . data } [ hemi ] <EOL> vertno = { '<STR_LIT>' : self . lh_vertno , '<STR_LIT>' : self . rh_vertno , <EOL> None : np . concatenate ( self . vertices ) } [ hemi ] <EOL> vert_idx , time_idx = _get_peak ( data , self . times , tmin , tmax , mode ) <EOL> return ( vert_idx if vert_as_index else vertno [ vert_idx ] , <EOL> time_idx if time_as_index else self . times [ time_idx ] ) <EOL> class VolSourceEstimate ( _BaseSourceEstimate ) : <EOL> """<STR_LIT>""" <EOL> @ verbose <EOL> def __init__ ( self , data , vertices = None , tmin = None , tstep = None , <EOL> subject = None , verbose = None ) : <EOL> if not ( isinstance ( vertices , np . ndarray ) or <EOL> isinstance ( vertices , list ) and len ( vertices ) == <NUM_LIT:1> ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _BaseSourceEstimate . __init__ ( self , data , vertices = vertices , tmin = tmin , <EOL> tstep = tstep , subject = subject , <EOL> verbose = verbose ) <EOL> @ verbose <EOL> def save ( self , fname , ftype = '<STR_LIT>' , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if ftype not in [ '<STR_LIT>' , '<STR_LIT:w>' ] : <EOL> raise ValueError ( '<STR_LIT>' % ftype ) <EOL> if ftype == '<STR_LIT>' : <EOL> logger . info ( '<STR_LIT>' ) <EOL> if not ( fname . endswith ( '<STR_LIT>' ) or fname . endswith ( '<STR_LIT>' ) ) : <EOL> fname += '<STR_LIT>' <EOL> _write_stc ( fname , tmin = self . tmin , tstep = self . tstep , <EOL> vertices = self . vertices , data = self . data ) <EOL> elif ftype == '<STR_LIT:w>' : <EOL> logger . info ( '<STR_LIT>' ) <EOL> if not ( fname . endswith ( '<STR_LIT>' ) or fname . endswith ( '<STR_LIT>' ) ) : <EOL> fname += '<STR_LIT>' <EOL> _write_w ( fname , vertices = self . vertices , data = self . data ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> def save_as_volume ( self , fname , src , dest = '<STR_LIT>' , mri_resolution = False ) : <EOL> """<STR_LIT>""" <EOL> save_stc_as_volume ( fname , self , src , dest = dest , <EOL> mri_resolution = mri_resolution ) <EOL> def as_volume ( self , src , dest = '<STR_LIT>' , mri_resolution = False ) : <EOL> """<STR_LIT>""" <EOL> return save_stc_as_volume ( None , self , src , dest = dest , <EOL> mri_resolution = mri_resolution ) <EOL> def __repr__ ( self ) : <EOL> if isinstance ( self . vertices , list ) : <EOL> nv = sum ( [ len ( v ) for v in self . vertices ] ) <EOL> else : <EOL> nv = self . vertices . size <EOL> s = "<STR_LIT>" % nv <EOL> if self . subject is not None : <EOL> s += "<STR_LIT>" % self . subject <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . tmin ) <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . times [ - <NUM_LIT:1> ] ) <EOL> s += "<STR_LIT>" % ( <NUM_LIT> * self . tstep ) <EOL> s += "<STR_LIT>" % self . shape <EOL> return "<STR_LIT>" % s <EOL> def get_peak ( self , tmin = None , tmax = None , mode = '<STR_LIT>' , <EOL> vert_as_index = False , time_as_index = False ) : <EOL> """<STR_LIT>""" <EOL> vert_idx , time_idx = _get_peak ( self . data , self . times , tmin , tmax , <EOL> mode ) <EOL> return ( vert_idx if vert_as_index else self . vertices [ vert_idx ] , <EOL> time_idx if time_as_index else self . times [ time_idx ] ) <EOL> class MixedSourceEstimate ( _BaseSourceEstimate ) : <EOL> """<STR_LIT>""" <EOL> @ verbose <EOL> def __init__ ( self , data , vertices = None , tmin = None , tstep = None , <EOL> subject = None , verbose = None ) : <EOL> if not isinstance ( vertices , list ) or len ( vertices ) < <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _BaseSourceEstimate . __init__ ( self , data , vertices = vertices , tmin = tmin , <EOL> tstep = tstep , subject = subject , <EOL> verbose = verbose ) <EOL> def plot_surface ( self , src , subject = None , surface = '<STR_LIT>' , hemi = '<STR_LIT>' , <EOL> colormap = '<STR_LIT>' , time_label = '<STR_LIT>' , <EOL> smoothing_steps = <NUM_LIT:10> , <EOL> transparent = None , alpha = <NUM_LIT:1.0> , time_viewer = False , <EOL> config_opts = { } , subjects_dir = None , figure = None , <EOL> views = '<STR_LIT>' , colorbar = True , clim = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> surf = _ensure_src ( src , kind = '<STR_LIT>' ) <EOL> data = self . data [ : surf [ <NUM_LIT:0> ] [ '<STR_LIT>' ] + surf [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ] <EOL> vertices = [ s [ '<STR_LIT>' ] for s in surf ] <EOL> stc = SourceEstimate ( data , vertices , self . tmin , self . tstep , <EOL> self . subject , self . verbose ) <EOL> return plot_source_estimates ( stc , subject , surface = surface , hemi = hemi , <EOL> colormap = colormap , time_label = time_label , <EOL> smoothing_steps = smoothing_steps , <EOL> transparent = transparent , alpha = alpha , <EOL> time_viewer = time_viewer , <EOL> config_opts = config_opts , <EOL> subjects_dir = subjects_dir , figure = figure , <EOL> views = views , colorbar = colorbar , clim = clim ) <EOL> @ verbose <EOL> def _morph_buffer ( data , idx_use , e , smooth , n_vertices , nearest , maps , <EOL> warn = True , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> n_iter = <NUM_LIT> <EOL> if smooth is not None : <EOL> if smooth <= <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> smooth -= <NUM_LIT:1> <EOL> e = e . tocsr ( ) <EOL> if sparse . issparse ( data ) : <EOL> use_sparse = True <EOL> if not isinstance ( data , sparse . csr_matrix ) : <EOL> data = data . tocsr ( ) <EOL> else : <EOL> use_sparse = False <EOL> done = False <EOL> for k in range ( n_iter + <NUM_LIT:1> ) : <EOL> mult = np . zeros ( e . shape [ <NUM_LIT:1> ] ) <EOL> mult [ idx_use ] = <NUM_LIT:1> <EOL> idx_use_data = idx_use <EOL> data_sum = e * mult <EOL> idx_use = np . where ( data_sum ) [ <NUM_LIT:0> ] <EOL> idx_out = idx_use <EOL> if smooth is None : <EOL> if k == n_iter or len ( idx_use ) >= n_vertices : <EOL> idx_out = None <EOL> done = True <EOL> elif k == smooth : <EOL> idx_out = None <EOL> done = True <EOL> data = _morph_mult ( data , e , use_sparse , idx_use_data , idx_out ) <EOL> if done is True : <EOL> break <EOL> if use_sparse : <EOL> data . data /= data_sum [ idx_use ] . repeat ( np . diff ( data . indptr ) ) <EOL> else : <EOL> data /= data_sum [ idx_use ] [ : , None ] <EOL> if use_sparse : <EOL> data_sum [ data_sum == <NUM_LIT:0> ] = <NUM_LIT:1> <EOL> data . data /= data_sum . repeat ( np . diff ( data . indptr ) ) <EOL> else : <EOL> data [ idx_use , : ] /= data_sum [ idx_use ] [ : , None ] <EOL> if len ( idx_use ) != len ( data_sum ) and warn : <EOL> warn_ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( len ( data_sum ) - len ( idx_use ) , len ( data_sum ) ) ) <EOL> logger . info ( '<STR_LIT>' % ( k + <NUM_LIT:1> ) ) <EOL> data_morphed = maps [ nearest , : ] * data <EOL> return data_morphed <EOL> def _morph_mult ( data , e , use_sparse , idx_use_data , idx_use_out = None ) : <EOL> """<STR_LIT>""" <EOL> if len ( idx_use_data ) < e . shape [ <NUM_LIT:1> ] : <EOL> if use_sparse : <EOL> data = e [ : , idx_use_data ] * data <EOL> else : <EOL> col , row = np . meshgrid ( np . arange ( data . shape [ <NUM_LIT:1> ] ) , idx_use_data ) <EOL> d_sparse = sparse . csr_matrix ( ( data . ravel ( ) , <EOL> ( row . ravel ( ) , col . ravel ( ) ) ) , <EOL> shape = ( e . shape [ <NUM_LIT:1> ] , data . shape [ <NUM_LIT:1> ] ) ) <EOL> data = e * d_sparse <EOL> data = np . asarray ( data . todense ( ) ) <EOL> else : <EOL> data = e * data <EOL> if idx_use_out is not None : <EOL> data = data [ idx_use_out ] <EOL> return data <EOL> def _get_subject_sphere_tris ( subject , subjects_dir ) : <EOL> spheres = [ os . path . join ( subjects_dir , subject , '<STR_LIT>' , <EOL> xh + '<STR_LIT>' ) for xh in [ '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> tris = [ read_surface ( s ) [ <NUM_LIT:1> ] for s in spheres ] <EOL> return tris <EOL> def _sparse_argmax_nnz_row ( csr_mat ) : <EOL> """<STR_LIT>""" <EOL> n_rows = csr_mat . shape [ <NUM_LIT:0> ] <EOL> idx = np . empty ( n_rows , dtype = np . int ) <EOL> for k in range ( n_rows ) : <EOL> row = csr_mat [ k ] . tocoo ( ) <EOL> idx [ k ] = row . col [ np . argmax ( row . data ) ] <EOL> return idx <EOL> def _morph_sparse ( stc , subject_from , subject_to , subjects_dir = None ) : <EOL> """<STR_LIT>""" <EOL> maps = read_morph_map ( subject_to , subject_from , subjects_dir ) <EOL> stc_morph = stc . copy ( ) <EOL> stc_morph . subject = subject_to <EOL> cnt = <NUM_LIT:0> <EOL> for k , hemi in enumerate ( [ '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> if stc . vertices [ k ] . size > <NUM_LIT:0> : <EOL> map_hemi = maps [ k ] <EOL> vertno_k = _sparse_argmax_nnz_row ( map_hemi [ stc . vertices [ k ] ] ) <EOL> order = np . argsort ( vertno_k ) <EOL> n_active_hemi = len ( vertno_k ) <EOL> data_hemi = stc_morph . _data [ cnt : cnt + n_active_hemi ] <EOL> stc_morph . _data [ cnt : cnt + n_active_hemi ] = data_hemi [ order ] <EOL> stc_morph . vertices [ k ] = vertno_k [ order ] <EOL> cnt += n_active_hemi <EOL> else : <EOL> stc_morph . vertices [ k ] = np . array ( [ ] , int ) <EOL> return stc_morph <EOL> @ verbose <EOL> def morph_data ( subject_from , subject_to , stc_from , grade = <NUM_LIT:5> , smooth = None , <EOL> subjects_dir = None , buffer_size = <NUM_LIT:64> , n_jobs = <NUM_LIT:1> , warn = True , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( stc_from , SourceEstimate ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> subjects_dir = get_subjects_dir ( subjects_dir , raise_error = True ) <EOL> nearest = grade_to_vertices ( subject_to , grade , subjects_dir , n_jobs ) <EOL> tris = _get_subject_sphere_tris ( subject_from , subjects_dir ) <EOL> maps = read_morph_map ( subject_from , subject_to , subjects_dir ) <EOL> data = [ stc_from . lh_data , stc_from . rh_data ] <EOL> data_morphed = [ None , None ] <EOL> n_chunks = ceil ( stc_from . data . shape [ <NUM_LIT:1> ] / float ( buffer_size ) ) <EOL> parallel , my_morph_buffer , _ = parallel_func ( _morph_buffer , n_jobs ) <EOL> for hemi in [ <NUM_LIT:0> , <NUM_LIT:1> ] : <EOL> e = mesh_edges ( tris [ hemi ] ) <EOL> e . data [ e . data == <NUM_LIT:2> ] = <NUM_LIT:1> <EOL> n_vertices = e . shape [ <NUM_LIT:0> ] <EOL> e = e + sparse . eye ( n_vertices , n_vertices ) <EOL> idx_use = stc_from . vertices [ hemi ] <EOL> if len ( idx_use ) == <NUM_LIT:0> : <EOL> continue <EOL> data_morphed [ hemi ] = np . concatenate ( <EOL> parallel ( my_morph_buffer ( data_buffer , idx_use , e , smooth , <EOL> n_vertices , nearest [ hemi ] , maps [ hemi ] , <EOL> warn = warn ) <EOL> for data_buffer <EOL> in np . array_split ( data [ hemi ] , n_chunks , axis = <NUM_LIT:1> ) ) , axis = <NUM_LIT:1> ) <EOL> vertices = [ nearest [ <NUM_LIT:0> ] , nearest [ <NUM_LIT:1> ] ] <EOL> if data_morphed [ <NUM_LIT:0> ] is None : <EOL> if data_morphed [ <NUM_LIT:1> ] is None : <EOL> data = np . r_ [ [ ] , [ ] ] <EOL> vertices = [ np . array ( [ ] , int ) , np . array ( [ ] , int ) ] <EOL> else : <EOL> data = data_morphed [ <NUM_LIT:1> ] <EOL> vertices = [ np . array ( [ ] , int ) , vertices [ <NUM_LIT:1> ] ] <EOL> elif data_morphed [ <NUM_LIT:1> ] is None : <EOL> data = data_morphed [ <NUM_LIT:0> ] <EOL> vertices = [ vertices [ <NUM_LIT:0> ] , np . array ( [ ] , int ) ] <EOL> else : <EOL> data = np . r_ [ data_morphed [ <NUM_LIT:0> ] , data_morphed [ <NUM_LIT:1> ] ] <EOL> stc_to = SourceEstimate ( data , vertices , stc_from . tmin , stc_from . tstep , <EOL> subject = subject_to , verbose = stc_from . verbose ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> return stc_to <EOL> @ verbose <EOL> def compute_morph_matrix ( subject_from , subject_to , vertices_from , vertices_to , <EOL> smooth = None , subjects_dir = None , warn = True , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( '<STR_LIT>' ) <EOL> subjects_dir = get_subjects_dir ( subjects_dir , raise_error = True ) <EOL> tris = _get_subject_sphere_tris ( subject_from , subjects_dir ) <EOL> maps = read_morph_map ( subject_from , subject_to , subjects_dir ) <EOL> morpher = [ None ] * <NUM_LIT:2> <EOL> for hemi in [ <NUM_LIT:0> , <NUM_LIT:1> ] : <EOL> e = mesh_edges ( tris [ hemi ] ) <EOL> e . data [ e . data == <NUM_LIT:2> ] = <NUM_LIT:1> <EOL> n_vertices = e . shape [ <NUM_LIT:0> ] <EOL> e = e + sparse . eye ( n_vertices , n_vertices ) <EOL> idx_use = vertices_from [ hemi ] <EOL> if len ( idx_use ) == <NUM_LIT:0> : <EOL> morpher [ hemi ] = [ ] <EOL> continue <EOL> m = sparse . eye ( len ( idx_use ) , len ( idx_use ) , format = '<STR_LIT>' ) <EOL> morpher [ hemi ] = _morph_buffer ( m , idx_use , e , smooth , n_vertices , <EOL> vertices_to [ hemi ] , maps [ hemi ] , warn = warn ) <EOL> if isinstance ( morpher [ <NUM_LIT:0> ] , list ) : <EOL> morpher = morpher [ <NUM_LIT:1> ] <EOL> elif isinstance ( morpher [ <NUM_LIT:1> ] , list ) : <EOL> morpher = morpher [ <NUM_LIT:0> ] <EOL> else : <EOL> morpher = sparse_block_diag ( morpher , format = '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> return morpher <EOL> @ verbose <EOL> def grade_to_vertices ( subject , grade , subjects_dir = None , n_jobs = <NUM_LIT:1> , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if subject == '<STR_LIT>' and grade == <NUM_LIT:5> : <EOL> return [ np . arange ( <NUM_LIT> ) , np . arange ( <NUM_LIT> ) ] <EOL> subjects_dir = get_subjects_dir ( subjects_dir , raise_error = True ) <EOL> spheres_to = [ os . path . join ( subjects_dir , subject , '<STR_LIT>' , <EOL> xh + '<STR_LIT>' ) for xh in [ '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> lhs , rhs = [ read_surface ( s ) [ <NUM_LIT:0> ] for s in spheres_to ] <EOL> if grade is not None : <EOL> if isinstance ( grade , list ) : <EOL> if not len ( grade ) == <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> vertices = grade <EOL> else : <EOL> ico = _get_ico_tris ( grade , return_surf = True ) <EOL> lhs /= np . sqrt ( np . sum ( lhs ** <NUM_LIT:2> , axis = <NUM_LIT:1> ) ) [ : , None ] <EOL> rhs /= np . sqrt ( np . sum ( rhs ** <NUM_LIT:2> , axis = <NUM_LIT:1> ) ) [ : , None ] <EOL> parallel , my_compute_nearest , _ = parallel_func ( _compute_nearest , n_jobs ) <EOL> lhs , rhs , rr = [ a . astype ( np . float32 ) <EOL> for a in [ lhs , rhs , ico [ '<STR_LIT>' ] ] ] <EOL> vertices = parallel ( my_compute_nearest ( xhs , rr ) <EOL> for xhs in [ lhs , rhs ] ) <EOL> vertices = [ np . sort ( verts ) for verts in vertices ] <EOL> else : <EOL> vertices = [ np . arange ( lhs . shape [ <NUM_LIT:0> ] ) , np . arange ( rhs . shape [ <NUM_LIT:0> ] ) ] <EOL> return vertices <EOL> def morph_data_precomputed ( subject_from , subject_to , stc_from , vertices_to , <EOL> morph_mat ) : <EOL> """<STR_LIT>""" <EOL> if not sparse . issparse ( morph_mat ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not isinstance ( vertices_to , list ) or not len ( vertices_to ) == <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not sum ( len ( v ) for v in vertices_to ) == morph_mat . shape [ <NUM_LIT:0> ] : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if not stc_from . data . shape [ <NUM_LIT:0> ] == morph_mat . shape [ <NUM_LIT:1> ] : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if stc_from . subject is not None and stc_from . subject != subject_from : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> data = morph_mat * stc_from . data <EOL> stc_to = SourceEstimate ( data , vertices_to , stc_from . tmin , stc_from . tstep , <EOL> verbose = stc_from . verbose , subject = subject_to ) <EOL> return stc_to <EOL> @ verbose <EOL> def spatio_temporal_src_connectivity ( src , n_times , dist = None , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if dist is None : <EOL> if src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] is None : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> used_verts = [ np . unique ( s [ '<STR_LIT>' ] ) for s in src ] <EOL> lh_tris = np . searchsorted ( used_verts [ <NUM_LIT:0> ] , src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> rh_tris = np . searchsorted ( used_verts [ <NUM_LIT:1> ] , src [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> tris = np . concatenate ( ( lh_tris , rh_tris + np . max ( lh_tris ) + <NUM_LIT:1> ) ) <EOL> connectivity = spatio_temporal_tris_connectivity ( tris , n_times ) <EOL> masks = [ in1d ( u , s [ '<STR_LIT>' ] ) for s , u in zip ( src , used_verts ) ] <EOL> if sum ( u . size for u in used_verts ) != connectivity . shape [ <NUM_LIT:0> ] / n_times : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if [ np . sum ( m ) for m in masks ] != [ len ( s [ '<STR_LIT>' ] ) for s in src ] : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> masks = np . concatenate ( masks ) <EOL> missing = <NUM_LIT:100> * float ( len ( masks ) - np . sum ( masks ) ) / len ( masks ) <EOL> if missing : <EOL> warn_ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % missing ) <EOL> masks = np . tile ( masks , n_times ) <EOL> masks = np . where ( masks ) [ <NUM_LIT:0> ] <EOL> connectivity = connectivity . tocsr ( ) <EOL> connectivity = connectivity [ masks ] <EOL> connectivity = connectivity [ : , masks ] <EOL> connectivity = connectivity . tocoo ( ) <EOL> return connectivity <EOL> else : <EOL> return spatio_temporal_dist_connectivity ( src , n_times , dist ) <EOL> @ verbose <EOL> def grade_to_tris ( grade , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> a = _get_ico_tris ( grade , None , False ) <EOL> tris = np . concatenate ( ( a , a + ( np . max ( a ) + <NUM_LIT:1> ) ) ) <EOL> return tris <EOL> @ verbose <EOL> def spatio_temporal_tris_connectivity ( tris , n_times , remap_vertices = False , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if remap_vertices : <EOL> logger . info ( '<STR_LIT>' ) <EOL> tris = np . searchsorted ( np . unique ( tris ) , tris ) <EOL> edges = mesh_edges ( tris ) . tocoo ( ) <EOL> return _get_connectivity_from_edges ( edges , n_times ) <EOL> @ verbose <EOL> def spatio_temporal_dist_connectivity ( src , n_times , dist , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] is None : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> edges = sparse_block_diag ( [ s [ '<STR_LIT>' ] [ s [ '<STR_LIT>' ] , : ] [ : , s [ '<STR_LIT>' ] ] <EOL> for s in src ] ) <EOL> edges . data [ : ] = np . less_equal ( edges . data , dist ) <EOL> edges = edges . tocsr ( ) <EOL> edges . eliminate_zeros ( ) <EOL> edges = edges . tocoo ( ) <EOL> return _get_connectivity_from_edges ( edges , n_times ) <EOL> @ verbose <EOL> def spatial_src_connectivity ( src , dist = None , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> return spatio_temporal_src_connectivity ( src , <NUM_LIT:1> , dist ) <EOL> @ verbose <EOL> def spatial_tris_connectivity ( tris , remap_vertices = False , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> return spatio_temporal_tris_connectivity ( tris , <NUM_LIT:1> , remap_vertices ) <EOL> def spatial_dist_connectivity ( src , dist , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> return spatio_temporal_dist_connectivity ( src , <NUM_LIT:1> , dist ) <EOL> def spatial_inter_hemi_connectivity ( src , dist , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> from scipy . spatial . distance import cdist <EOL> src = _ensure_src ( src , kind = '<STR_LIT>' ) <EOL> conn = cdist ( src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] , <EOL> src [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ src [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ] ) <EOL> conn = sparse . csr_matrix ( conn <= dist , dtype = int ) <EOL> empties = [ sparse . csr_matrix ( ( nv , nv ) , dtype = int ) for nv in conn . shape ] <EOL> conn = sparse . vstack ( [ sparse . hstack ( [ empties [ <NUM_LIT:0> ] , conn ] ) , <EOL> sparse . hstack ( [ conn . T , empties [ <NUM_LIT:1> ] ] ) ] ) <EOL> return conn <EOL> @ verbose <EOL> def _get_connectivity_from_edges ( edges , n_times , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> n_vertices = edges . shape [ <NUM_LIT:0> ] <EOL> logger . info ( "<STR_LIT>" % n_vertices ) <EOL> nnz = edges . col . size <EOL> aux = n_vertices * np . arange ( n_times ) [ : , None ] * np . ones ( ( <NUM_LIT:1> , nnz ) , np . int ) <EOL> col = ( edges . col [ None , : ] + aux ) . ravel ( ) <EOL> row = ( edges . row [ None , : ] + aux ) . ravel ( ) <EOL> if n_times > <NUM_LIT:1> : <EOL> o = ( n_vertices * np . arange ( n_times - <NUM_LIT:1> ) [ : , None ] + <EOL> np . arange ( n_vertices ) [ None , : ] ) . ravel ( ) <EOL> d = ( n_vertices * np . arange ( <NUM_LIT:1> , n_times ) [ : , None ] + <EOL> np . arange ( n_vertices ) [ None , : ] ) . ravel ( ) <EOL> row = np . concatenate ( ( row , o , d ) ) <EOL> col = np . concatenate ( ( col , d , o ) ) <EOL> data = np . ones ( edges . data . size * n_times + <NUM_LIT:2> * n_vertices * ( n_times - <NUM_LIT:1> ) , <EOL> dtype = np . int ) <EOL> connectivity = coo_matrix ( ( data , ( row , col ) ) , <EOL> shape = ( n_times * n_vertices , ) * <NUM_LIT:2> ) <EOL> return connectivity <EOL> @ verbose <EOL> def _get_ico_tris ( grade , verbose = None , return_surf = False ) : <EOL> """<STR_LIT>""" <EOL> ico = _get_ico_surface ( grade ) <EOL> if not return_surf : <EOL> return ico [ '<STR_LIT>' ] <EOL> else : <EOL> return ico <EOL> def save_stc_as_volume ( fname , stc , src , dest = '<STR_LIT>' , mri_resolution = False ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( stc , VolSourceEstimate ) : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> n_times = stc . data . shape [ <NUM_LIT:1> ] <EOL> shape = src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> shape3d = ( shape [ <NUM_LIT:2> ] , shape [ <NUM_LIT:1> ] , shape [ <NUM_LIT:0> ] ) <EOL> shape = ( n_times , shape [ <NUM_LIT:2> ] , shape [ <NUM_LIT:1> ] , shape [ <NUM_LIT:0> ] ) <EOL> vol = np . zeros ( shape ) <EOL> mask3d = src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . reshape ( shape3d ) . astype ( np . bool ) <EOL> if mri_resolution : <EOL> mri_shape3d = ( src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> mri_shape = ( n_times , src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> mri_vol = np . zeros ( mri_shape ) <EOL> interpolator = src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> for k , v in enumerate ( vol ) : <EOL> v [ mask3d ] = stc . data [ : , k ] <EOL> if mri_resolution : <EOL> mri_vol [ k ] = ( interpolator * v . ravel ( ) ) . reshape ( mri_shape3d ) <EOL> if mri_resolution : <EOL> vol = mri_vol <EOL> vol = vol . T <EOL> if mri_resolution : <EOL> affine = src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] . copy ( ) <EOL> else : <EOL> affine = src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] . copy ( ) <EOL> if dest == '<STR_LIT>' : <EOL> affine = np . dot ( src [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] , affine ) <EOL> affine [ : <NUM_LIT:3> ] *= <NUM_LIT> <EOL> try : <EOL> import nibabel as nib <EOL> except ImportError : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> header = nib . nifti1 . Nifti1Header ( ) <EOL> header . set_xyzt_units ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> header [ '<STR_LIT>' ] [ <NUM_LIT:4> ] = <NUM_LIT> * stc . tstep <EOL> with warnings . catch_warnings ( record = True ) : <EOL> img = nib . Nifti1Image ( vol , affine , header = header ) <EOL> if fname is not None : <EOL> nib . save ( img , fname ) <EOL> return img <EOL> def _get_label_flip ( labels , label_vertidx , src ) : <EOL> """<STR_LIT>""" <EOL> from . label import label_sign_flip <EOL> label_flip = list ( ) <EOL> for label , vertidx in zip ( labels , label_vertidx ) : <EOL> if label . hemi == '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if vertidx is not None : <EOL> flip = label_sign_flip ( label , src ) [ : , None ] <EOL> else : <EOL> flip = None <EOL> label_flip . append ( flip ) <EOL> return label_flip <EOL> @ verbose <EOL> def _gen_extract_label_time_course ( stcs , labels , src , mode = '<STR_LIT>' , <EOL> allow_empty = False , verbose = None ) : <EOL> """<STR_LIT>""" <EOL> n_labels = len ( labels ) <EOL> vertno = [ s [ '<STR_LIT>' ] for s in src ] <EOL> nvert = [ len ( vn ) for vn in vertno ] <EOL> label_vertidx = list ( ) <EOL> for label in labels : <EOL> if label . hemi == '<STR_LIT>' : <EOL> sub_labels = [ label . lh , label . rh ] <EOL> else : <EOL> sub_labels = [ label ] <EOL> this_vertidx = list ( ) <EOL> for slabel in sub_labels : <EOL> if slabel . hemi == '<STR_LIT>' : <EOL> this_vertno = np . intersect1d ( vertno [ <NUM_LIT:0> ] , slabel . vertices ) <EOL> vertidx = np . searchsorted ( vertno [ <NUM_LIT:0> ] , this_vertno ) <EOL> elif slabel . hemi == '<STR_LIT>' : <EOL> this_vertno = np . intersect1d ( vertno [ <NUM_LIT:1> ] , slabel . vertices ) <EOL> vertidx = nvert [ <NUM_LIT:0> ] + np . searchsorted ( vertno [ <NUM_LIT:1> ] , this_vertno ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % label . name ) <EOL> this_vertidx . append ( vertidx ) <EOL> this_vertidx = np . concatenate ( this_vertidx ) <EOL> if len ( this_vertidx ) == <NUM_LIT:0> : <EOL> msg = ( '<STR_LIT>' <EOL> % label . name ) <EOL> if not allow_empty : <EOL> raise ValueError ( msg ) <EOL> else : <EOL> warn_ ( msg + '<STR_LIT>' ) <EOL> this_vertidx = None <EOL> label_vertidx . append ( this_vertidx ) <EOL> if mode == '<STR_LIT>' : <EOL> pass <EOL> elif mode == '<STR_LIT>' : <EOL> label_flip = _get_label_flip ( labels , label_vertidx , src ) <EOL> elif mode == '<STR_LIT>' : <EOL> label_flip = _get_label_flip ( labels , label_vertidx , src ) <EOL> elif mode == '<STR_LIT>' : <EOL> pass <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % mode ) <EOL> for stc in stcs : <EOL> if len ( stc . vertices [ <NUM_LIT:0> ] ) != nvert [ <NUM_LIT:0> ] or len ( stc . vertices [ <NUM_LIT:1> ] ) != nvert [ <NUM_LIT:1> ] : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if any ( np . any ( svn != vn ) for svn , vn in zip ( stc . vertices , vertno ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' <EOL> % ( n_labels , mode ) ) <EOL> label_tc = np . zeros ( ( n_labels , stc . data . shape [ <NUM_LIT:1> ] ) , <EOL> dtype = stc . data . dtype ) <EOL> if mode == '<STR_LIT>' : <EOL> for i , vertidx in enumerate ( label_vertidx ) : <EOL> if vertidx is not None : <EOL> label_tc [ i ] = np . mean ( stc . data [ vertidx , : ] , axis = <NUM_LIT:0> ) <EOL> elif mode == '<STR_LIT>' : <EOL> for i , ( vertidx , flip ) in enumerate ( zip ( label_vertidx , <EOL> label_flip ) ) : <EOL> if vertidx is not None : <EOL> label_tc [ i ] = np . mean ( flip * stc . data [ vertidx , : ] , axis = <NUM_LIT:0> ) <EOL> elif mode == '<STR_LIT>' : <EOL> for i , ( vertidx , flip ) in enumerate ( zip ( label_vertidx , <EOL> label_flip ) ) : <EOL> if vertidx is not None : <EOL> U , s , V = linalg . svd ( stc . data [ vertidx , : ] , <EOL> full_matrices = False ) <EOL> sign = np . sign ( np . dot ( U [ : , <NUM_LIT:0> ] , flip ) ) <EOL> scale = linalg . norm ( s ) / np . sqrt ( len ( vertidx ) ) <EOL> label_tc [ i ] = sign * scale * V [ <NUM_LIT:0> ] <EOL> elif mode == '<STR_LIT>' : <EOL> for i , vertidx in enumerate ( label_vertidx ) : <EOL> if vertidx is not None : <EOL> label_tc [ i ] = np . max ( np . abs ( stc . data [ vertidx , : ] ) , axis = <NUM_LIT:0> ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % mode ) <EOL> yield label_tc <EOL> @ verbose <EOL> def extract_label_time_course ( stcs , labels , src , mode = '<STR_LIT>' , <EOL> allow_empty = False , return_generator = False , <EOL> verbose = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( stcs , SourceEstimate ) : <EOL> stcs = [ stcs ] <EOL> return_several = False <EOL> return_generator = False <EOL> else : <EOL> return_several = True <EOL> if not isinstance ( labels , list ) : <EOL> labels = [ labels ] <EOL> label_tc = _gen_extract_label_time_course ( stcs , labels , src , mode = mode , <EOL> allow_empty = allow_empty ) <EOL> if not return_generator : <EOL> label_tc = list ( label_tc ) <EOL> if not return_several : <EOL> label_tc = label_tc [ <NUM_LIT:0> ] <EOL> return label_tc </s>
