<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import numpy as np <EOL> import quantities as pq <EOL> import os <EOL> try : <EOL> import neuroshare as ns <EOL> except ImportError as err : <EOL> print ( err ) <EOL> else : <EOL> pass <EOL> from neo . io . baseio import BaseIO <EOL> from neo . core import Segment , AnalogSignal , SpikeTrain , EventArray , EpochArray <EOL> class NeuroshareapiIO ( BaseIO ) : <EOL> is_readable = True <EOL> is_writable = False <EOL> supported_objects = [ Segment , AnalogSignal , SpikeTrain , EventArray , EpochArray ] <EOL> has_header = False <EOL> is_streameable = False <EOL> readable_objects = [ Segment , AnalogSignal , SpikeTrain , EventArray , EpochArray ] <EOL> writeable_objects = [ ] <EOL> read_params = { <EOL> Segment : [ <EOL> ( "<STR_LIT>" , { "<STR_LIT:value>" : <NUM_LIT:0.> , "<STR_LIT:label>" : "<STR_LIT>" } ) , <EOL> ( "<STR_LIT>" , { "<STR_LIT:value>" : <NUM_LIT:0.> , "<STR_LIT:label>" : "<STR_LIT>" } ) , <EOL> ] , <EOL> } <EOL> write_params = None <EOL> name = "<STR_LIT>" <EOL> extensions = [ ] <EOL> mode = "<STR_LIT:file>" <EOL> def __init__ ( self , filename = None , dllpath = None ) : <EOL> """<STR_LIT>""" <EOL> BaseIO . __init__ ( self ) <EOL> self . filename = filename <EOL> eventID = <NUM_LIT:1> <EOL> analogID = <NUM_LIT:2> <EOL> epochID = <NUM_LIT:3> <EOL> if self . filename != None : <EOL> if dllpath is not None : <EOL> name = os . path . splitext ( os . path . basename ( dllpath ) ) [ <NUM_LIT:0> ] <EOL> library = ns . Library ( name , dllpath ) <EOL> else : <EOL> library = None <EOL> self . fd = ns . File ( self . filename , library = library ) <EOL> self . metadata = self . fd . metadata_raw <EOL> self . metadata [ "<STR_LIT>" ] = <NUM_LIT:1.> / self . metadata [ "<STR_LIT>" ] <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = list ( ) <EOL> self . metadata [ "<STR_LIT>" ] = <NUM_LIT:0> <EOL> for entity in self . fd . entities : <EOL> if entity . entity_type == analogID and entity . label [ <NUM_LIT:0> : <NUM_LIT:4> ] != "<STR_LIT>" : <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . label [ - <NUM_LIT:4> : ] ) <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . id ) <EOL> self . metadata [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> if entity . entity_type == eventID and entity . label [ <NUM_LIT:0> : <NUM_LIT:4> ] == "<STR_LIT>" : <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . label [ <NUM_LIT:0> : <NUM_LIT:4> ] + entity . label [ - <NUM_LIT:4> : ] ) <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . id ) <EOL> self . metadata [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> if entity . entity_type == eventID and entity . label [ <NUM_LIT:0> : <NUM_LIT:4> ] == "<STR_LIT>" : <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . label [ - <NUM_LIT:5> : ] ) <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . id ) <EOL> self . metadata [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> if entity . entity_type == epochID and entity . label [ <NUM_LIT:0> : <NUM_LIT:4> ] == "<STR_LIT>" : <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . label [ - <NUM_LIT:4> : ] ) <EOL> self . metadata [ "<STR_LIT>" ] . append ( entity . id ) <EOL> self . metadata [ "<STR_LIT>" ] += <NUM_LIT:1> <EOL> def read_segment ( self , <EOL> lazy = False , <EOL> cascade = True , <EOL> t_start = <NUM_LIT:0.> , <EOL> segment_duration = <NUM_LIT:0.> , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> if segment_duration == <NUM_LIT:0.> : <EOL> segment_duration = float ( self . metadata [ "<STR_LIT>" ] ) <EOL> if segment_duration >= float ( self . metadata [ "<STR_LIT>" ] ) : <EOL> segment_duration = float ( self . metadata [ "<STR_LIT>" ] ) <EOL> if segment_duration + t_start > float ( self . metadata [ "<STR_LIT>" ] ) : <EOL> segment_duration = float ( self . metadata [ "<STR_LIT>" ] ) - t_start <EOL> seg = Segment ( name = "<STR_LIT>" ) <EOL> if cascade : <EOL> if self . metadata [ "<STR_LIT>" ] == <NUM_LIT:0> : <EOL> print ( "<STR_LIT>" ) <EOL> else : <EOL> for i in range ( self . metadata [ "<STR_LIT>" ] ) : <EOL> ana = self . read_analogsignal ( lazy = lazy , cascade = cascade , <EOL> channel_index = self . metadata [ "<STR_LIT>" ] [ i ] , <EOL> segment_duration = segment_duration , t_start = t_start ) <EOL> seg . analogsignals += [ ana ] <EOL> for i in range ( self . metadata [ "<STR_LIT>" ] ) : <EOL> eva = self . read_eventarray ( lazy = lazy , <EOL> cascade = cascade , <EOL> channel_index = self . metadata [ "<STR_LIT>" ] [ i ] , <EOL> segment_duration = segment_duration , <EOL> t_start = t_start , ) <EOL> seg . eventarrays += [ eva ] <EOL> for i in range ( self . metadata [ "<STR_LIT>" ] ) : <EOL> epa = self . read_epocharray ( lazy = lazy , <EOL> cascade = cascade , <EOL> channel_index = self . metadata [ "<STR_LIT>" ] [ i ] , <EOL> segment_duration = segment_duration , <EOL> t_start = t_start , ) <EOL> seg . epocharrays += [ epa ] <EOL> for i in range ( self . metadata [ "<STR_LIT>" ] ) : <EOL> sptr = self . read_spiketrain ( lazy = lazy , cascade = cascade , <EOL> channel_index = self . metadata [ "<STR_LIT>" ] [ i ] , <EOL> segment_duration = segment_duration , <EOL> t_start = t_start ) <EOL> seg . spiketrains += [ sptr ] <EOL> seg . create_many_to_one_relationship ( ) <EOL> return seg <EOL> """<STR_LIT>""" <EOL> def read_analogsignal ( self , <EOL> lazy = False , <EOL> cascade = True , <EOL> channel_index = <NUM_LIT:0> , <EOL> segment_duration = <NUM_LIT:0.> , <EOL> t_start = <NUM_LIT:0.> , <EOL> ) : <EOL> if segment_duration == <NUM_LIT:0.> : <EOL> segment_duration = float ( self . metadata [ "<STR_LIT>" ] ) <EOL> if segment_duration >= float ( self . metadata [ "<STR_LIT>" ] ) : <EOL> segment_duration = float ( self . metadata [ "<STR_LIT>" ] ) <EOL> if lazy : <EOL> anasig = AnalogSignal ( [ ] , units = "<STR_LIT>" , sampling_rate = self . metadata [ "<STR_LIT>" ] * pq . Hz , <EOL> t_start = t_start * pq . s , <EOL> ) <EOL> tvect = np . arange ( t_start , t_start + segment_duration , <NUM_LIT:1.> / self . metadata [ "<STR_LIT>" ] ) <EOL> anasig . lazy_shape = tvect . shape <EOL> else : <EOL> sig = self . fd . get_entity ( channel_index ) <EOL> sigUnits = sig . units <EOL> chanName = sig . label [ - <NUM_LIT:4> : ] <EOL> startat = int ( t_start * self . metadata [ "<STR_LIT>" ] ) <EOL> bins = int ( ( segment_duration + t_start ) * self . metadata [ "<STR_LIT>" ] ) <EOL> if startat + bins > sig . item_count : <EOL> bins = sig . item_count - startat <EOL> sig , _ , _ = sig . get_data ( index = startat , count = bins ) <EOL> anasig = AnalogSignal ( sig , units = sigUnits , sampling_rate = self . metadata [ "<STR_LIT>" ] * pq . Hz , <EOL> t_start = t_start * pq . s , <EOL> t_stop = ( t_start + segment_duration ) * pq . s , <EOL> channel_index = channel_index ) <EOL> anasig . annotate ( info = "<STR_LIT>" % chanName ) <EOL> return anasig <EOL> def read_spiketrain ( self , <EOL> lazy = False , <EOL> cascade = True , <EOL> channel_index = <NUM_LIT:0> , <EOL> segment_duration = <NUM_LIT:0.> , <EOL> t_start = <NUM_LIT:0.> ) : <EOL> """<STR_LIT>""" <EOL> sr = self . metadata [ "<STR_LIT>" ] <EOL> times = list ( ) <EOL> if lazy : <EOL> spiketr = SpikeTrain ( times , units = pq . s , <EOL> t_stop = t_start + segment_duration , <EOL> t_start = t_start * pq . s , lazy_shape = <NUM_LIT> ) <EOL> else : <EOL> tempSpks = self . fd . get_entity ( channel_index ) <EOL> startat = tempSpks . get_index_by_time ( t_start , <NUM_LIT:0> ) <EOL> endat = tempSpks . get_index_by_time ( float ( segment_duration + t_start ) , - <NUM_LIT:1> ) <EOL> numIndx = endat - startat <EOL> waveforms = np . array ( np . zeros ( [ numIndx , tempSpks . max_sample_count ] ) ) <EOL> for i in range ( startat , endat , <NUM_LIT:1> ) : <EOL> tempCuts , timeStamp , duration , unit = tempSpks . get_data ( i ) <EOL> waveforms [ i ] = tempCuts [ <NUM_LIT:0> ] <EOL> times . append ( timeStamp ) <EOL> spiketr = SpikeTrain ( times , units = pq . s , <EOL> t_stop = t_start + segment_duration , <EOL> t_start = t_start * pq . s , <EOL> name = "<STR_LIT>" + tempSpks . label [ - <NUM_LIT:3> : ] , <EOL> waveforms = waveforms * pq . volt , <EOL> sampling_rate = sr * pq . Hz , <EOL> file_origin = self . filename , <EOL> annotate = ( "<STR_LIT>" + str ( channel_index ) ) ) <EOL> return spiketr <EOL> def read_eventarray ( self , lazy = False , cascade = True , <EOL> channel_index = <NUM_LIT:0> , <EOL> t_start = <NUM_LIT:0.> , <EOL> segment_duration = <NUM_LIT:0.> ) : <EOL> """<STR_LIT>""" <EOL> if lazy : <EOL> eva = EventArray ( file_origin = self . filename ) <EOL> else : <EOL> tempNames = list ( ) <EOL> tempTimeStamp = list ( ) <EOL> trigEntity = self . fd . get_entity ( channel_index ) <EOL> startat = trigEntity . get_index_by_time ( t_start , <NUM_LIT:0> ) <EOL> endat = trigEntity . get_index_by_time ( float ( segment_duration + t_start ) , - <NUM_LIT:1> ) <EOL> for i in range ( startat , endat + <NUM_LIT:1> , <NUM_LIT:1> ) : <EOL> tempNames . append ( trigEntity . label [ - <NUM_LIT:8> : ] ) <EOL> tempData , onOrOff = trigEntity . get_data ( i ) <EOL> if onOrOff == <NUM_LIT:1> : <EOL> tempTimeStamp . append ( tempData ) <EOL> eva = EventArray ( labels = np . array ( tempNames , dtype = "<STR_LIT:S>" ) , <EOL> times = np . array ( tempTimeStamp ) * pq . s , <EOL> file_origin = self . filename , <EOL> description = "<STR_LIT>" ) <EOL> return eva <EOL> def read_epocharray ( self , lazy = False , cascade = True , <EOL> channel_index = <NUM_LIT:0> , <EOL> t_start = <NUM_LIT:0.> , <EOL> segment_duration = <NUM_LIT:0.> ) : <EOL> """<STR_LIT>""" <EOL> if lazy : <EOL> epa = EpochArray ( file_origin = self . filename , <EOL> times = None , durations = None , labels = None ) <EOL> else : <EOL> tempNames = list ( ) <EOL> tempTimeStamp = list ( ) <EOL> durations = list ( ) <EOL> digEntity = self . fd . get_entity ( channel_index ) <EOL> startat = digEntity . get_index_by_time ( t_start , <NUM_LIT:0> ) <EOL> endat = digEntity . get_index_by_time ( float ( segment_duration + t_start ) , - <NUM_LIT:1> ) <EOL> for i in range ( startat , endat + <NUM_LIT:1> , <NUM_LIT:1> ) : <EOL> if i % <NUM_LIT:2> == <NUM_LIT:1> : <EOL> tempNames . append ( digEntity . label [ - <NUM_LIT:8> : ] ) <EOL> tempData , onOrOff = digEntity . get_data ( i - <NUM_LIT:1> ) <EOL> tempTimeStamp . append ( tempData ) <EOL> tempData1 , onOrOff = digEntity . get_data ( i ) <EOL> durations . append ( tempData1 - tempData ) <EOL> epa = EpochArray ( file_origin = self . filename , <EOL> times = np . array ( tempTimeStamp ) * pq . s , <EOL> durations = np . array ( durations ) * pq . s , <EOL> labels = np . array ( tempNames , dtype = "<STR_LIT:S>" ) , <EOL> description = "<STR_LIT>" ) <EOL> return epa </s>
