multielectrode_grasp_pd/code/python-neo/neo/rawio/tests/test_blackrockrawio.py

# -*- coding: utf-8 -*-
"""
Tests of neo.rawio.examplerawio
"""

# needed for python 3 compatibility
from __future__ import unicode_literals, print_function, division, absolute_import

import unittest

from neo.rawio.blackrockrawio import BlackrockRawIO
from neo.rawio.tests.common_rawio_test import BaseTestRawIO

import numpy as np
from numpy.testing import assert_equal

try:
    import scipy.io

    HAVE_SCIPY = True
except ImportError:
    HAVE_SCIPY = False


class TestBlackrockRawIO(BaseTestRawIO, unittest.TestCase, ):
    rawioclass = BlackrockRawIO
    entities_to_test = ['FileSpec2.3001']

    files_to_download = [
        'FileSpec2.3001.nev',
        'FileSpec2.3001.ns5',
        'FileSpec2.3001.ccf',
        'FileSpec2.3001.mat',
        'blackrock_2_1/l101210-001.mat',
        'blackrock_2_1/l101210-001_nev-02_ns5.mat',
        'blackrock_2_1/l101210-001.ns2',
        'blackrock_2_1/l101210-001.ns5',
        'blackrock_2_1/l101210-001.nev',
        'blackrock_2_1/l101210-001-02.nev']

    @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

        Ported to the rawio API by Samuel Garcia.
        """

        # Load data from Matlab generated files
        ml = scipy.io.loadmat(self.get_filename_path('FileSpec2.3001.mat'))

        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
        reader = BlackrockRawIO(filename=self.get_filename_path('FileSpec2.3001'))
        reader.parse_header()

        # Check if analog data on channels 1-8 are equal
        self.assertGreater(reader.signal_channels_count(), 0)
        for c in range(0, 8):
            raw_sigs = reader.get_analogsignal_chunk(channel_indexes=[c])
            raw_sigs = raw_sigs.flatten()
            assert_equal(raw_sigs[:-1], lfp_ml[c, :])

        # Check if spikes in channels are equal
        nb_unit = reader.unit_channels_count()
        for unit_index in range(nb_unit):
            unit_name = reader.header['unit_channels'][unit_index]['name']
            # name is chXX#YY where XX is channel_id and YY is unit_id
            channel_id, unit_id = unit_name.split('#')
            channel_id = int(channel_id.replace('ch', ''))
            unit_id = int(unit_id)

            matlab_spikes = ts_ml[(elec_ml == channel_id) & (unit_ml == unit_id)]

            io_spikes = reader.get_spike_timestamps(unit_index=unit_index)
            assert_equal(io_spikes, matlab_spikes)

            # Check waveforms of channel 1, unit 0
            if channel_id == 1 and unit_id == 0:
                io_waveforms = reader.get_spike_raw_waveforms(unit_index=unit_index)
                io_waveforms = io_waveforms[:, 0, :]  # remove dim 1
                assert_equal(io_waveforms, wf_ml)

        # Check if digital input port events are equal
        nb_ev_chan = reader.event_channels_count()
        # ~ print(reader.header['event_channels'])
        for ev_chan in range(nb_ev_chan):
            name = reader.header['event_channels']['name'][ev_chan]
            # ~ print(name)
            all_timestamps, _, labels = reader.get_event_timestamps(
                event_channel_index=ev_chan)
            if name == 'digital_input_port':
                for label in np.unique(labels):
                    python_digievents = all_timestamps[labels == label]
                    matlab_digievents = mts_ml[mid_ml == int(label)]
                    assert_equal(python_digievents, matlab_digievents)
            elif name == 'comments':
                pass
                # TODO: Save comments to Matlab file.

    @unittest.skipUnless(HAVE_SCIPY, "requires scipy")
    def test_compare_blackrockio_with_matlabloader_v21(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.

        Ported to the rawio API by Samuel Garcia.
        """

        dirname = self.get_filename_path('blackrock_2_1/l101210-001')
        # First run with parameters for ns5, then run with correct parameters for ns2
        parameters = [('blackrock_2_1/l101210-001_nev-02_ns5.mat',
                       {'nsx_to_load': 5, 'nev_override': '-'.join([dirname, '02'])}, 96),
                      ('blackrock_2_1/l101210-001.mat', {'nsx_to_load': 2}, 6)]
        for param in parameters:
            # Load data from Matlab generated files
            ml = scipy.io.loadmat(self.get_filename_path(filename=param[0]))
            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']  # waveforms
            mts_ml = ml['mts']  # marker time stamps
            mid_ml = ml['mid']  # marker IDs

            # Load data from original data files using the Neo BlackrockIO
            reader = BlackrockRawIO(dirname, **param[1])
            reader.parse_header()

            # Check if analog data are equal
            self.assertGreater(reader.signal_channels_count(), 0)

            for c in range(0, param[2]):
                raw_sigs = reader.get_analogsignal_chunk(channel_indexes=[c])
                raw_sigs = raw_sigs.flatten()
                assert_equal(raw_sigs[:], lfp_ml[c, :])

            # Check if spikes in channels are equal
            nb_unit = reader.unit_channels_count()
            for unit_index in range(nb_unit):
                unit_name = reader.header['unit_channels'][unit_index]['name']
                # name is chXX#YY where XX is channel_id and YY is unit_id
                channel_id, unit_id = unit_name.split('#')
                channel_id = int(channel_id.replace('ch', ''))
                unit_id = int(unit_id)

                matlab_spikes = ts_ml[(elec_ml == channel_id) & (unit_ml == unit_id)]

                io_spikes = reader.get_spike_timestamps(unit_index=unit_index)
                assert_equal(io_spikes, matlab_spikes)

                # Check all waveforms
                io_waveforms = reader.get_spike_raw_waveforms(unit_index=unit_index)
                io_waveforms = io_waveforms[:, 0, :]  # remove dim 1
                matlab_wf = wf_ml[np.nonzero(
                    np.logical_and(elec_ml == channel_id, unit_ml == unit_id)), :][0]
                assert_equal(io_waveforms, matlab_wf)

            # Check if digital input port events are equal
            nb_ev_chan = reader.event_channels_count()
            # ~ print(reader.header['event_channels'])
            for ev_chan in range(nb_ev_chan):
                name = reader.header['event_channels']['name'][ev_chan]
                # ~ print(name)
                if name == 'digital_input_port':
                    all_timestamps, _, labels = reader.get_event_timestamps(
                        event_channel_index=ev_chan)

                    for label in np.unique(labels):
                        python_digievents = all_timestamps[labels == label]
                        matlab_digievents = mts_ml[mid_ml == int(label)]
                        assert_equal(python_digievents, matlab_digievents)


if __name__ == '__main__':
    unittest.main()