<s> import operator , re , logging , os . path <EOL> import numpy as np <EOL> import eegtools . io <EOL> from shared import Recording , data_source <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> sessions = range ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> subjects = sessions <EOL> LICENSE = '''<STR_LIT>''' <EOL> URL_TEMPLATE = '<STR_LIT>' '<STR_LIT>' <EOL> EVENTS = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:4> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:5> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:6> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:7> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:8> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:9> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:10> , '<STR_LIT>' ) , <EOL> ] <EOL> B1 = dict ( T0 = <NUM_LIT:1> ) <EOL> B2 = dict ( T0 = <NUM_LIT:2> ) <EOL> T1 = dict ( T0 = <NUM_LIT:0> , T1 = <NUM_LIT:3> , T2 = <NUM_LIT:4> ) <EOL> T2 = dict ( T0 = <NUM_LIT:0> , T1 = <NUM_LIT:5> , T2 = <NUM_LIT:6> ) <EOL> T3 = dict ( T0 = <NUM_LIT:0> , T1 = <NUM_LIT:7> , T2 = <NUM_LIT:8> ) <EOL> T4 = dict ( T0 = <NUM_LIT:0> , T1 = <NUM_LIT:9> , T2 = <NUM_LIT:10> ) <EOL> TASKS = [ B1 , B2 ] + [ T1 , T2 , T3 , T4 ] * <NUM_LIT:3> <EOL> KNOWN_LABS = '<STR_LIT>' . split ( ) <EOL> log = logging . getLogger ( __name__ ) <EOL> def gen_urls ( subject , url_template = URL_TEMPLATE ) : <EOL> '''<STR_LIT>''' <EOL> runs = np . arange ( <NUM_LIT> ) + <NUM_LIT:1> <EOL> return [ url_template % dict ( subject = subject , run = run ) <EOL> for run in runs ] , runs <EOL> def load_schalk_run ( edf , run ) : <EOL> '''<STR_LIT>''' <EOL> task_dic = TASKS [ run - <NUM_LIT:1> ] <EOL> log . debug ( '<STR_LIT>' % ( run , task_dic ) ) <EOL> d = eegtools . io . load_edf ( edf ) <EOL> fs , events = d . sample_rate , [ ] <EOL> for ( start , duration , texts ) in d . annotations : <EOL> for label in texts : <EOL> events . append ( <EOL> ( task_dic [ str ( label ) ] , start * fs , ( start + duration ) * fs ) ) <EOL> return d , np . asarray ( events , int ) . T . reshape ( <NUM_LIT:3> , - <NUM_LIT:1> ) <EOL> def clean_chan_lab ( feat_lab ) : <EOL> '''<STR_LIT>''' <EOL> lookup = dict ( ( l . lower ( ) , l ) for l in KNOWN_LABS ) <EOL> return [ lookup [ l . strip ( '<STR_LIT:.>' ) . lower ( ) ] for l in feat_lab ] <EOL> def concatenate_events ( events , block_lens ) : <EOL> '''<STR_LIT>''' <EOL> def shift_events ( events , offset ) : <EOL> events = events . copy ( ) <EOL> events [ <NUM_LIT:1> : ] += offset <EOL> return events <EOL> offset = np . cumsum ( [ <NUM_LIT:0> ] + block_lens [ : - <NUM_LIT:1> ] ) <EOL> return np . hstack ( [ shift_events ( be , o ) for ( be , o ) in zip ( events , offset ) ] ) <EOL> def block_dt ( n , sample_rate ) : <EOL> '''<STR_LIT>''' <EOL> dt = np . ones ( n ) * <NUM_LIT:1.> / sample_rate <EOL> dt [ <NUM_LIT:0> ] = np . nan <EOL> return dt <EOL> def load ( subject , ds = data_source ( ) , url_template = URL_TEMPLATE ) : <EOL> '''<STR_LIT>''' <EOL> urls , runs = gen_urls ( subject , url_template ) <EOL> log . debug ( '<STR_LIT>' % ( urls , ) ) <EOL> rec = [ load_schalk_run ( ds . open ( u ) , r ) for ( u , r ) in zip ( urls , runs ) ] <EOL> runs , events = zip ( * rec ) <EOL> X = np . hstack ( [ r . X . astype ( np . float32 ) for r in runs ] ) <EOL> dt = np . hstack ( [ block_dt ( r . X . shape [ <NUM_LIT:1> ] , r . sample_rate ) for r in runs ] ) [ <NUM_LIT:1> : ] <EOL> chan_lab = clean_chan_lab ( runs [ <NUM_LIT:0> ] . chan_lab ) <EOL> folds = np . hstack ( [ <EOL> np . ones ( e . shape [ <NUM_LIT:1> ] , int ) * i for ( i , e ) in enumerate ( events ) ] ) <EOL> E = concatenate_events ( events , [ r . X . shape [ <NUM_LIT:1> ] for r in runs ] ) <EOL> return Recording ( X = X , dt = dt , chan_lab = chan_lab , events = E , folds = folds , <EOL> event_lab = EVENTS , rec_id = '<STR_LIT>' % subject , <EOL> license = LICENSE ) </s>
