123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230 |
- # -*- coding: utf-8 -*-
- """
- Tests of neo.io.blackrockio
- """
- # needed for python 3 compatibility
- from __future__ import absolute_import
- try:
- import unittest2 as unittest
- except ImportError:
- import unittest
- from numpy.testing import assert_equal
- import numpy as np
- import quantities as pq
- from neo.io.blackrockio import BlackrockIO
- from neo.test.iotest.common_io_test import BaseTestIO
- from neo.test.iotest.tools import get_test_file_full_path
- # check scipy
- try:
- from distutils import version
- import scipy.io
- import scipy.version
- except ImportError as err:
- HAVE_SCIPY = False
- SCIPY_ERR = err
- else:
- if version.LooseVersion(scipy.version.version) < '0.8':
- HAVE_SCIPY = False
- SCIPY_ERR = ImportError("your scipy version is too old to support " +
- "MatlabIO, you need at least 0.8. " +
- "You have %s" % scipy.version.version)
- else:
- HAVE_SCIPY = True
- SCIPY_ERR = None
- class CommonTests(BaseTestIO, unittest.TestCase):
- ioclass = BlackrockIO
- files_to_test = ['FileSpec2.3001']
- files_to_download = [
- 'FileSpec2.3001.nev',
- 'FileSpec2.3001.ns5',
- 'FileSpec2.3001.ccf',
- 'FileSpec2.3001.mat']
- ioclass = BlackrockIO
- def test_inputs_V23(self):
- """
- Test various inputs to BlackrockIO.read_block with version 2.3 file
- to check for parsing errors.
- """
- try:
- b = BlackrockIO(
- get_test_file_full_path(
- ioclass=BlackrockIO,
- filename='FileSpec2.3001',
- directory=self.local_test_dir, clean=False),
- verbose=False)
- except:
- self.fail()
- # Load data to maximum extent, one None is not given as list
- block = b.read_block(
- n_starts=[None], n_stops=None, channels=range(1, 9),
- nsx_to_load=5, units='all', load_events=True,
- load_waveforms=False)
- lena = len(block.segments[0].analogsignals[0])
- numspa = len(block.segments[0].spiketrains[0])
- # Load data using a negative time and a time exceeding the end of the
- # recording
- too_large_tstop = block.segments[0].analogsignals[0].t_stop + 1 * pq.s
- block = b.read_block(
- n_starts=[-100 * pq.ms], n_stops=[too_large_tstop],
- channels=range(1, 9), nsx_to_load=[5], units='all',
- load_events=False, load_waveforms=False)
- lenb = len(block.segments[0].analogsignals[0])
- numspb = len(block.segments[0].spiketrains[0])
- # Same length of analog signal?
- # Both should have read the complete data set!
- self.assertEqual(lena, lenb)
- # Same length of spike train?
- # Both should have read the complete data set!
- self.assertEqual(numspa, numspb)
- # n_starts and n_stops not given as list
- # verifies identical length of returned signals given equal durations
- # as input
- ns5_unit = block.segments[0].analogsignals[0].sampling_period
- block = b.read_block(
- n_starts=100 * ns5_unit, n_stops=200 * ns5_unit,
- channels=range(1, 9), nsx_to_load=5, units='all',
- load_events=False, load_waveforms=False)
- lena = len(block.segments[0].analogsignals[0])
- block = b.read_block(
- n_starts=301 * ns5_unit, n_stops=401 * ns5_unit,
- channels=range(1, 9), nsx_to_load=5, units='all',
- load_events=False, load_waveforms=False)
- lenb = len(block.segments[0].analogsignals[0])
- # Same length?
- self.assertEqual(lena, lenb)
- # Length should be 100 samples exactly
- self.assertEqual(lena, 100)
- # Load partial data types and check if this is selection is made
- block = b.read_block(
- n_starts=None, n_stops=None, channels=range(1, 9),
- nsx_to_load=5, units='none', load_events=False,
- load_waveforms=True)
- self.assertEqual(len(block.segments), 1)
- self.assertEqual(len(block.segments[0].analogsignals), 8)
- self.assertEqual(len(block.channel_indexes), 8)
- self.assertEqual(len(block.channel_indexes[0].units), 0)
- self.assertEqual(len(block.segments[0].events), 0)
- self.assertEqual(len(block.segments[0].spiketrains), 0)
- # NOTE: channel 6 does not contain any unit
- block = b.read_block(
- n_starts=[None, 3000 * pq.ms], n_stops=[1000 * pq.ms, None],
- channels=range(1, 9), nsx_to_load='none',
- units={1: 0, 5: 0, 6: 0}, load_events=True,
- load_waveforms=True)
- self.assertEqual(len(block.segments), 2)
- self.assertEqual(len(block.segments[0].analogsignals), 0)
- self.assertEqual(len(block.channel_indexes), 8)
- self.assertEqual(len(block.channel_indexes[0].units), 1)
- self.assertEqual(len(block.segments[0].events), 0)
- self.assertEqual(len(block.segments[0].spiketrains), 2)
- @unittest.skipUnless(HAVE_SCIPY, "requires scipy")
- def test_compare_blackrockio_with_matlabloader(self):
- """
- This test compares the output of ReachGraspIO.read_block() with the
- output generated by a Matlab implementation of a Blackrock file reader
- provided by the company. The output for comparison is provided in a
- .mat file created by the script create_data_matlab_blackrock.m.
- The function tests LFPs, spike times, and digital events on channels
- 80-83 and spike waveforms on channel 82, unit 1.
- For details on the file contents, refer to FileSpec2.3.txt
- """
- # Load data from Matlab generated files
- ml = scipy.io.loadmat(
- get_test_file_full_path(
- ioclass=BlackrockIO,
- filename='FileSpec2.3001.mat',
- directory=self.local_test_dir, clean=False))
- lfp_ml = ml['lfp'] # (channel x time) LFP matrix
- ts_ml = ml['ts'] # spike time stamps
- elec_ml = ml['el'] # spike electrodes
- unit_ml = ml['un'] # spike unit IDs
- wf_ml = ml['wf'] # waveform unit 1 channel 1
- mts_ml = ml['mts'] # marker time stamps
- mid_ml = ml['mid'] # marker IDs
- # Load data in channels 1-3 from original data files using the Neo
- # BlackrockIO
- session = BlackrockIO(
- get_test_file_full_path(
- ioclass=BlackrockIO,
- filename='FileSpec2.3001',
- directory=self.local_test_dir, clean=False),
- verbose=False)
- block = session.read_block(
- channels=range(1, 9), units='all', nsx_to_load='all',
- scaling='raw', load_waveforms=True, load_events=True)
- # Check if analog data on channels 1-8 are equal
- self.assertGreater(len(block.channel_indexes), 0)
- for chidx in block.channel_indexes:
- # Should only have one AnalogSignal per ChannelIndex
- self.assertEqual(len(chidx.analogsignals), 1)
- idx = chidx.analogsignals[0].annotations['channel_id']
- if idx in range(1, 9):
- # We ignore the last sample of the Analogsignal returned by the
- # Python implementation, since due to an error in the
- # corresponding matlab loader the last sample was ignored and
- # not saved to the test file
- assert_equal(np.squeeze(
- chidx.analogsignals[0].base[:-1]), lfp_ml[idx - 1, :])
- # Check if spikes in channels 1,3,5,7 are equal
- self.assertEqual(len(block.segments), 1)
- for st_i in block.segments[0].spiketrains:
- channelid = st_i.annotations['channel_id']
- if channelid in range(1, 7, 2):
- unitid = st_i.annotations['unit_id']
- matlab_spikes = ts_ml[np.nonzero(
- np.logical_and(elec_ml == channelid, unit_ml == unitid))]
- assert_equal(st_i.base, matlab_spikes)
- # Check waveforms of channel 1, unit 0
- if channelid == 1 and unitid == 0:
- assert_equal(np.squeeze(st_i.waveforms), wf_ml)
- # Check if digital input port events are equal
- self.assertGreater(len(block.segments[0].events), 0)
- for ea_i in block.segments[0].events:
- if ea_i.name == 'digital_input_port':
- # Get all digital event IDs in this recording
- marker_ids = set(ea_i.labels)
- for marker_id in marker_ids:
- python_digievents = ea_i.times.base[
- ea_i.labels == marker_id]
- matlab_digievents = mts_ml[
- np.nonzero(mid_ml == int(marker_id))]
- assert_equal(python_digievents, matlab_digievents)
- # Note: analog input events are not yet supported
- if __name__ == '__main__':
- unittest.main()
|