Scheduled service maintenance on November 22


On Friday, November 22, 2024, between 06:00 CET and 18:00 CET, GIN services will undergo planned maintenance. Extended service interruptions should be expected. We will try to keep downtimes to a minimum, but recommend that users avoid critical tasks, large data uploads, or DOI requests during this time.

We apologize for any inconvenience.

test_blackrockio.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  1. # -*- coding: utf-8 -*-
  2. """
  3. Tests of neo.io.blackrockio
  4. """
  5. # needed for python 3 compatibility
  6. from __future__ import absolute_import
  7. try:
  8. import unittest2 as unittest
  9. except ImportError:
  10. import unittest
  11. from numpy.testing import assert_equal
  12. import numpy as np
  13. import quantities as pq
  14. from neo.io.blackrockio import BlackrockIO
  15. from neo.test.iotest.common_io_test import BaseTestIO
  16. from neo.test.iotest.tools import get_test_file_full_path
  17. # check scipy
  18. try:
  19. from distutils import version
  20. import scipy.io
  21. import scipy.version
  22. except ImportError as err:
  23. HAVE_SCIPY = False
  24. SCIPY_ERR = err
  25. else:
  26. if version.LooseVersion(scipy.version.version) < '0.8':
  27. HAVE_SCIPY = False
  28. SCIPY_ERR = ImportError("your scipy version is too old to support " +
  29. "MatlabIO, you need at least 0.8. " +
  30. "You have %s" % scipy.version.version)
  31. else:
  32. HAVE_SCIPY = True
  33. SCIPY_ERR = None
  34. class CommonTests(BaseTestIO, unittest.TestCase):
  35. ioclass = BlackrockIO
  36. files_to_test = ['FileSpec2.3001']
  37. files_to_download = [
  38. 'FileSpec2.3001.nev',
  39. 'FileSpec2.3001.ns5',
  40. 'FileSpec2.3001.ccf',
  41. 'FileSpec2.3001.mat']
  42. ioclass = BlackrockIO
  43. def test_inputs_V23(self):
  44. """
  45. Test various inputs to BlackrockIO.read_block with version 2.3 file
  46. to check for parsing errors.
  47. """
  48. try:
  49. b = BlackrockIO(
  50. get_test_file_full_path(
  51. ioclass=BlackrockIO,
  52. filename='FileSpec2.3001',
  53. directory=self.local_test_dir, clean=False),
  54. verbose=False)
  55. except:
  56. self.fail()
  57. # Load data to maximum extent, one None is not given as list
  58. block = b.read_block(
  59. n_starts=[None], n_stops=None, channels=range(1, 9),
  60. nsx_to_load=5, units='all', load_events=True,
  61. load_waveforms=False)
  62. lena = len(block.segments[0].analogsignals[0])
  63. numspa = len(block.segments[0].spiketrains[0])
  64. # Load data using a negative time and a time exceeding the end of the
  65. # recording
  66. too_large_tstop = block.segments[0].analogsignals[0].t_stop + 1 * pq.s
  67. block = b.read_block(
  68. n_starts=[-100 * pq.ms], n_stops=[too_large_tstop],
  69. channels=range(1, 9), nsx_to_load=[5], units='all',
  70. load_events=False, load_waveforms=False)
  71. lenb = len(block.segments[0].analogsignals[0])
  72. numspb = len(block.segments[0].spiketrains[0])
  73. # Same length of analog signal?
  74. # Both should have read the complete data set!
  75. self.assertEqual(lena, lenb)
  76. # Same length of spike train?
  77. # Both should have read the complete data set!
  78. self.assertEqual(numspa, numspb)
  79. # n_starts and n_stops not given as list
  80. # verifies identical length of returned signals given equal durations
  81. # as input
  82. ns5_unit = block.segments[0].analogsignals[0].sampling_period
  83. block = b.read_block(
  84. n_starts=100 * ns5_unit, n_stops=200 * ns5_unit,
  85. channels=range(1, 9), nsx_to_load=5, units='all',
  86. load_events=False, load_waveforms=False)
  87. lena = len(block.segments[0].analogsignals[0])
  88. block = b.read_block(
  89. n_starts=301 * ns5_unit, n_stops=401 * ns5_unit,
  90. channels=range(1, 9), nsx_to_load=5, units='all',
  91. load_events=False, load_waveforms=False)
  92. lenb = len(block.segments[0].analogsignals[0])
  93. # Same length?
  94. self.assertEqual(lena, lenb)
  95. # Length should be 100 samples exactly
  96. self.assertEqual(lena, 100)
  97. # Load partial data types and check if this is selection is made
  98. block = b.read_block(
  99. n_starts=None, n_stops=None, channels=range(1, 9),
  100. nsx_to_load=5, units='none', load_events=False,
  101. load_waveforms=True)
  102. self.assertEqual(len(block.segments), 1)
  103. self.assertEqual(len(block.segments[0].analogsignals), 8)
  104. self.assertEqual(len(block.channel_indexes), 8)
  105. self.assertEqual(len(block.channel_indexes[0].units), 0)
  106. self.assertEqual(len(block.segments[0].events), 0)
  107. self.assertEqual(len(block.segments[0].spiketrains), 0)
  108. # NOTE: channel 6 does not contain any unit
  109. block = b.read_block(
  110. n_starts=[None, 3000 * pq.ms], n_stops=[1000 * pq.ms, None],
  111. channels=range(1, 9), nsx_to_load='none',
  112. units={1: 0, 5: 0, 6: 0}, load_events=True,
  113. load_waveforms=True)
  114. self.assertEqual(len(block.segments), 2)
  115. self.assertEqual(len(block.segments[0].analogsignals), 0)
  116. self.assertEqual(len(block.channel_indexes), 8)
  117. self.assertEqual(len(block.channel_indexes[0].units), 1)
  118. self.assertEqual(len(block.segments[0].events), 0)
  119. self.assertEqual(len(block.segments[0].spiketrains), 2)
  120. @unittest.skipUnless(HAVE_SCIPY, "requires scipy")
  121. def test_compare_blackrockio_with_matlabloader(self):
  122. """
  123. This test compares the output of ReachGraspIO.read_block() with the
  124. output generated by a Matlab implementation of a Blackrock file reader
  125. provided by the company. The output for comparison is provided in a
  126. .mat file created by the script create_data_matlab_blackrock.m.
  127. The function tests LFPs, spike times, and digital events on channels
  128. 80-83 and spike waveforms on channel 82, unit 1.
  129. For details on the file contents, refer to FileSpec2.3.txt
  130. """
  131. # Load data from Matlab generated files
  132. ml = scipy.io.loadmat(
  133. get_test_file_full_path(
  134. ioclass=BlackrockIO,
  135. filename='FileSpec2.3001.mat',
  136. directory=self.local_test_dir, clean=False))
  137. lfp_ml = ml['lfp'] # (channel x time) LFP matrix
  138. ts_ml = ml['ts'] # spike time stamps
  139. elec_ml = ml['el'] # spike electrodes
  140. unit_ml = ml['un'] # spike unit IDs
  141. wf_ml = ml['wf'] # waveform unit 1 channel 1
  142. mts_ml = ml['mts'] # marker time stamps
  143. mid_ml = ml['mid'] # marker IDs
  144. # Load data in channels 1-3 from original data files using the Neo
  145. # BlackrockIO
  146. session = BlackrockIO(
  147. get_test_file_full_path(
  148. ioclass=BlackrockIO,
  149. filename='FileSpec2.3001',
  150. directory=self.local_test_dir, clean=False),
  151. verbose=False)
  152. block = session.read_block(
  153. channels=range(1, 9), units='all', nsx_to_load='all',
  154. scaling='raw', load_waveforms=True, load_events=True)
  155. # Check if analog data on channels 1-8 are equal
  156. self.assertGreater(len(block.channel_indexes), 0)
  157. for chidx in block.channel_indexes:
  158. # Should only have one AnalogSignal per ChannelIndex
  159. self.assertEqual(len(chidx.analogsignals), 1)
  160. idx = chidx.analogsignals[0].annotations['channel_id']
  161. if idx in range(1, 9):
  162. # We ignore the last sample of the Analogsignal returned by the
  163. # Python implementation, since due to an error in the
  164. # corresponding matlab loader the last sample was ignored and
  165. # not saved to the test file
  166. assert_equal(np.squeeze(
  167. chidx.analogsignals[0].base[:-1]), lfp_ml[idx - 1, :])
  168. # Check if spikes in channels 1,3,5,7 are equal
  169. self.assertEqual(len(block.segments), 1)
  170. for st_i in block.segments[0].spiketrains:
  171. channelid = st_i.annotations['channel_id']
  172. if channelid in range(1, 7, 2):
  173. unitid = st_i.annotations['unit_id']
  174. matlab_spikes = ts_ml[np.nonzero(
  175. np.logical_and(elec_ml == channelid, unit_ml == unitid))]
  176. assert_equal(st_i.base, matlab_spikes)
  177. # Check waveforms of channel 1, unit 0
  178. if channelid == 1 and unitid == 0:
  179. assert_equal(np.squeeze(st_i.waveforms), wf_ml)
  180. # Check if digital input port events are equal
  181. self.assertGreater(len(block.segments[0].events), 0)
  182. for ea_i in block.segments[0].events:
  183. if ea_i.name == 'digital_input_port':
  184. # Get all digital event IDs in this recording
  185. marker_ids = set(ea_i.labels)
  186. for marker_id in marker_ids:
  187. python_digievents = ea_i.times.base[
  188. ea_i.labels == marker_id]
  189. matlab_digievents = mts_ml[
  190. np.nonzero(mid_ml == int(marker_id))]
  191. assert_equal(python_digievents, matlab_digievents)
  192. # Note: analog input events are not yet supported
  193. if __name__ == '__main__':
  194. unittest.main()