<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import glob <EOL> import os . path <EOL> import sys <EOL> import tempfile <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> import numpy as np <EOL> import quantities as pq <EOL> import neo <EOL> from neo . test . iotest . common_io_test import BaseTestIO <EOL> from neo . test . tools import assert_arrays_almost_equal <EOL> from neo . io . klustakwikio import KlustaKwikIO , HAVE_MLAB <EOL> @ unittest . skipUnless ( HAVE_MLAB , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . version_info [ <NUM_LIT:0> ] > <NUM_LIT:2> , "<STR_LIT>" ) <EOL> class testFilenameParser ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . dirname = os . path . join ( tempfile . gettempdir ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if not os . path . exists ( self . dirname ) : <EOL> raise unittest . SkipTest ( '<STR_LIT>' + <EOL> self . dirname ) <EOL> def test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) ) <EOL> if not BaseTestIO . use_network : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> fetfiles = kio . _fp . read_filenames ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( fetfiles ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( os . path . abspath ( fetfiles [ <NUM_LIT:0> ] ) , <EOL> os . path . abspath ( os . path . join ( self . dirname , <EOL> '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( os . path . abspath ( fetfiles [ <NUM_LIT:1> ] ) , <EOL> os . path . abspath ( os . path . join ( self . dirname , <EOL> '<STR_LIT>' ) ) ) <EOL> def test2 ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def test3 ( self ) : <EOL> """<STR_LIT>""" <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) ) <EOL> if not BaseTestIO . use_network : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> clufiles = kio . _fp . read_filenames ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( clufiles ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( os . path . abspath ( clufiles [ <NUM_LIT:1> ] ) , <EOL> os . path . abspath ( os . path . join ( self . dirname , <EOL> '<STR_LIT>' ) ) ) <EOL> @ unittest . skipUnless ( HAVE_MLAB , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . version_info [ <NUM_LIT:0> ] > <NUM_LIT:2> , "<STR_LIT>" ) <EOL> class testRead ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . dirname = os . path . join ( tempfile . gettempdir ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if not os . path . exists ( self . dirname ) : <EOL> raise unittest . SkipTest ( '<STR_LIT>' + <EOL> self . dirname ) <EOL> def test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) , <EOL> sampling_rate = <NUM_LIT> ) <EOL> block = kio . read ( ) [ <NUM_LIT:0> ] <EOL> seg = block . segments [ <NUM_LIT:0> ] <EOL> self . assertEqual ( len ( seg . spiketrains ) , <NUM_LIT:4> ) <EOL> for st in seg . spiketrains : <EOL> self . assertEqual ( st . units , np . array ( <NUM_LIT:1.0> ) * pq . s ) <EOL> self . assertEqual ( st . t_start , <NUM_LIT:0.0> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertTrue ( np . all ( seg . spiketrains [ <NUM_LIT:0> ] . times == np . array ( [ <NUM_LIT> , <EOL> <NUM_LIT> ] ) ) ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:1> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:1> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:1> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:1> ] . t_start , <NUM_LIT:0.0> ) <EOL> self . assertTrue ( np . all ( seg . spiketrains [ <NUM_LIT:1> ] . times == np . array ( [ <NUM_LIT> ] ) ) ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:2> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:2> ] . annotations [ '<STR_LIT>' ] , - <NUM_LIT:1> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:2> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:2> ] . t_start , <NUM_LIT:0.0> ) <EOL> self . assertTrue ( np . all ( seg . spiketrains [ <NUM_LIT:2> ] . times == np . array ( [ <NUM_LIT> ] ) ) ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:3> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:3> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:3> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:3> ] . t_start , <NUM_LIT:0.0> ) <EOL> self . assertTrue ( np . all ( seg . spiketrains [ <NUM_LIT:3> ] . times == np . array ( [ <NUM_LIT> , <EOL> <NUM_LIT> ] ) ) ) <EOL> def test2 ( self ) : <EOL> """<STR_LIT>""" <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) , <EOL> sampling_rate = <NUM_LIT> ) <EOL> block = kio . read ( ) [ <NUM_LIT:0> ] <EOL> seg = block . segments [ <NUM_LIT:0> ] <EOL> self . assertEqual ( len ( seg . spiketrains ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . name , '<STR_LIT>' ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . annotations [ '<STR_LIT>' ] , <NUM_LIT:5> ) <EOL> self . assertEqual ( seg . spiketrains [ <NUM_LIT:0> ] . t_start , <NUM_LIT:0.0> ) <EOL> self . assertTrue ( np . all ( seg . spiketrains [ <NUM_LIT:0> ] . times == np . array ( [ <NUM_LIT> , <EOL> <NUM_LIT> , <EOL> <NUM_LIT> ] ) ) ) <EOL> @ unittest . skipUnless ( HAVE_MLAB , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . version_info [ <NUM_LIT:0> ] > <NUM_LIT:2> , "<STR_LIT>" ) <EOL> class testWrite ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . dirname = os . path . join ( tempfile . gettempdir ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if not os . path . exists ( self . dirname ) : <EOL> raise unittest . SkipTest ( '<STR_LIT>' + <EOL> self . dirname ) <EOL> def test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> block = neo . Block ( ) <EOL> segment = neo . Segment ( ) <EOL> segment2 = neo . Segment ( ) <EOL> block . segments . append ( segment ) <EOL> block . segments . append ( segment2 ) <EOL> st1 = neo . SpikeTrain ( times = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st1 . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> st1 . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> segment . spiketrains . append ( st1 ) <EOL> st1B = neo . SpikeTrain ( times = [ <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st1B . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> segment2 . spiketrains . append ( st1B ) <EOL> st2 = neo . SpikeTrain ( times = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st2 . annotations [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> st2 . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> segment . spiketrains . append ( st2 ) <EOL> st3 = neo . SpikeTrain ( times = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st3 . annotations [ '<STR_LIT>' ] = - <NUM_LIT:1> <EOL> st3 . annotations [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> segment . spiketrains . append ( st3 ) <EOL> st4 = neo . SpikeTrain ( times = [ <NUM_LIT> , <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st4 . annotations [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> segment . spiketrains . append ( st4 ) <EOL> delete_test_session ( ) <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) , <EOL> sampling_rate = <NUM_LIT> ) <EOL> kio . write_block ( block ) <EOL> for fn in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . assertTrue ( os . path . exists ( os . path . join ( self . dirname , <EOL> '<STR_LIT>' + fn ) ) ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:6> , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:11> , <NUM_LIT> ] ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> ] ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:0> , <NUM_LIT:50> , <NUM_LIT> , <NUM_LIT:100> ] ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> ] ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:0> , <NUM_LIT:5> , <NUM_LIT:9> ] ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> delete_test_session ( ) <EOL> @ unittest . skipUnless ( HAVE_MLAB , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . version_info [ <NUM_LIT:0> ] > <NUM_LIT:2> , "<STR_LIT>" ) <EOL> class testWriteWithFeatures ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . dirname = os . path . join ( tempfile . gettempdir ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if not os . path . exists ( self . dirname ) : <EOL> raise unittest . SkipTest ( '<STR_LIT>' + <EOL> self . dirname ) <EOL> def test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> block = neo . Block ( ) <EOL> segment = neo . Segment ( ) <EOL> segment2 = neo . Segment ( ) <EOL> block . segments . append ( segment ) <EOL> block . segments . append ( segment2 ) <EOL> st1 = neo . SpikeTrain ( times = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] , units = '<STR_LIT:s>' , t_stop = <NUM_LIT:1.> ) <EOL> st1 . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> st1 . annotations [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> wff = np . array ( [ <EOL> [ <NUM_LIT> , <NUM_LIT> ] , <EOL> [ - <NUM_LIT> , <NUM_LIT> ] , <EOL> [ <NUM_LIT> , - <NUM_LIT> ] ] ) <EOL> st1 . annotations [ '<STR_LIT>' ] = wff <EOL> segment . spiketrains . append ( st1 ) <EOL> if not os . path . exists ( self . dirname ) : <EOL> os . mkdir ( self . dirname ) <EOL> delete_test_session ( self . dirname ) <EOL> kio = KlustaKwikIO ( filename = os . path . join ( self . dirname , '<STR_LIT>' ) , <EOL> sampling_rate = <NUM_LIT> ) <EOL> kio . write_block ( block ) <EOL> for fn in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . assertTrue ( os . path . exists ( os . path . join ( self . dirname , <EOL> '<STR_LIT>' + fn ) ) ) <EOL> fi = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) <EOL> self . assertEqual ( fi . readline ( ) , '<STR_LIT>' ) <EOL> data = fi . readlines ( ) <EOL> new_wff = [ ] <EOL> new_times = [ ] <EOL> for line in data : <EOL> line_split = line . split ( ) <EOL> new_wff . append ( [ float ( val ) for val in line_split [ : - <NUM_LIT:1> ] ] ) <EOL> new_times . append ( int ( line_split [ - <NUM_LIT:1> ] ) ) <EOL> self . assertEqual ( new_times , [ <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:6> ] ) <EOL> assert_arrays_almost_equal ( wff , np . array ( new_wff ) , <NUM_LIT> ) <EOL> data = file ( os . path . join ( self . dirname , '<STR_LIT>' ) ) . readlines ( ) <EOL> data = [ int ( d ) for d in data ] <EOL> self . assertEqual ( data , [ <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> block = kio . read_block ( ) <EOL> train = block . segments [ <NUM_LIT:0> ] . spiketrains [ <NUM_LIT:0> ] <EOL> assert_arrays_almost_equal ( wff , train . annotations [ '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> delete_test_session ( self . dirname ) <EOL> @ unittest . skipUnless ( HAVE_MLAB , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . version_info [ <NUM_LIT:0> ] > <NUM_LIT:2> , "<STR_LIT>" ) <EOL> class CommonTests ( BaseTestIO , unittest . TestCase ) : <EOL> ioclass = KlustaKwikIO <EOL> files_to_test = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> files_to_download = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> def delete_test_session ( dirname = None ) : <EOL> """<STR_LIT>""" <EOL> if dirname is None : <EOL> dirname = os . path . join ( os . path . dirname ( __file__ ) , <EOL> '<STR_LIT>' ) <EOL> for fi in glob . glob ( os . path . join ( dirname , '<STR_LIT:*>' ) ) : <EOL> os . remove ( fi ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
