multielectrode_grasp/code/reachgraspio/test_reachgraspio.py

import sys
from os.path import join, realpath, dirname, split

sys.path.insert(0, realpath(join(dirname(__file__), '..')))
import numpy as np

import unittest
import reachgraspio.reachgraspio as rg

rio = rg.ReachGraspIO("../../datasets_blackrock/i140703-001",
                      odml_directory="../../datasets_blackrock",
                      nsx_to_load='all')
block1 = rio.read_block(lazy=True)

rio = rg.ReachGraspIO("../../datasets_blackrock/l101210-001",
                      odml_directory="../../datasets_blackrock",
                      nsx_to_load='all')
block2 = rio.read_block(lazy=True)

class RGIOTestCase(unittest.TestCase):
    def setUp(self):
        self.blocks = [block1, block2]

    def test_group_units_present(self):
        for block in self.blocks:
            unit_groups = [g for g in block.groups if 'Unit' in g.name]
            self.assertGreater(len(unit_groups), 0)
            self.assertGreaterEqual(len(unit_groups), len(block.segments[0].spiketrains))

    def test_channel_infos_present(self):
        for block in self.blocks:
            for seg in block.segments:
                for anasig in seg.analogsignals:
                    if anasig.annotations['neural_signal']:
                        self.assertIn('connector_aligned_ids', anasig.array_annotations)
                        self.assertIn('coordinates_x', anasig.array_annotations)
                        self.assertIn('coordinates_y', anasig.array_annotations)

    def test_group_unit_annotations(self):
        for block in self.blocks:
            for group in block.groups:
                if 'Unit' in group.name:
                    self.assertIn('unit_id', group.annotations)
                    self.assertIn('connector_aligned_id', group.annotations)
                    self.assertIn('sua', group.annotations)
                    self.assertIn('mua', group.annotations)
                    self.assertIn('noise', group.annotations)

                    # To be investigated
                    # if group.annotations['unit_id'] > 0:
                    #     print(group.annotations)
                    #     self.assertIn('spike_duration', group.annotations)
                    #     self.assertIn('spike_amplitude', group.annotations)
                    #     self.assertIn('spike_count', group.annotations)

    def test_group_unit_linking(self):
        for block in self.blocks:
            # run this test only if neuronal signals are present
            if not [a for seg in block.segments for a in seg.analogsignals if
                    a.annotations['neural_signal']]:
                return
            for group in block.groups:
                if 'Unit' in group.name:
                    asig_annotations = group.channelviews[0].obj.array_annotations
                    idx = group.channelviews[0].index
                    self.assertEqual(group.annotations['channel_id'],
                                     asig_annotations['channel_ids'][idx])

    def test_consecutive_trial_ids(self):
        for block in self.blocks:
            for seg in block.segments:
                for ev in seg.events:
                    if 'trial_id' in ev.array_annotations:
                        # exclude invalid trial_ids:
                        trial_ids = ev.array_annotations['trial_id']
                        valid_ids = trial_ids[trial_ids != -1]

                        # Lilou session has one trial (197), where the object
                        # release happens late and during the next trial, thus
                        # disrupting incremental trial IDs
                        if split(
                                block.file_origin)[1]=='l101210-001':
                            valid_ids = valid_ids[valid_ids != 197]

                        self.assertTrue(all(np.diff(valid_ids) >= 0))

    def test_data_annotations_present(self):
        for block in self.blocks:
            for seg in block.segments:
                for anasig in seg.analogsignals:
                    if anasig.annotations['neural_signal']:
                        for key in ['channel_ids', 'channel_names',
                                    'hi_pass_freq', 'lo_pass_freq',
                                    'hi_pass_order', 'lo_pass_order', 'filter_type',
                                    'electrode_reject_IFC', 'electrode_reject_LFC',
                                    'electrode_reject_HFC', 'coordinates_x', 'coordinates_y']:
                            self.assertIn(key, anasig.array_annotations)

                for st in seg.spiketrains:
                    for key in ['id', 'channel_id', 'unit_id', 'unit_tag',
                                'electrode_reject_IFC', 'electrode_reject_LFC', 'electrode_reject_HFC',
                                'sua', 'mua', 'noise', 'coordinate_x', 'coordinate_y', 'connector_aligned_id']:
                        self.assertIn(key, st.annotations)

    def test_event_annotations(self):
        for block in self.blocks:
            for seg in block.segments:
                for ev in seg.events:
                    if ev.name in ['DigitalTrialEvents', 'AnalogTrialEvents', 'TrialEvents']:
                        for key in ['trial_event_labels', 'trial_timestamp_id', 'trial_id',
                                    'belongs_to_trialtype', 'performance_in_trial',
                                    'performance_in_trial_str', 'trial_reject_HFC', 'trial_reject_LFC',
                                    'trial_reject_IFC']:
                            self.assertIn(key, ev.array_annotations)

                ev_names = [ev.name for ev in seg.events]
                for key in ['AnalogTrialEvents', 'DigitalTrialEvents', 'TrialEvents']:
                    self.assertIn(key, ev_names)

    def test_block_annotation(self):
        for block in self.blocks:
            self.assertIn('conditions', block.annotations)
            self.assertGreater(len(block.annotations['conditions']), 0)

    def test_connector_aligned_ids_coordinates(self):
        for block in self.blocks:
            for seg in block.segments:
                for anasig in seg.analogsignals:
                    if not anasig.annotations['neural_signal']:
                        continue  # only check annotations of neural signals
                    for coords in ['coordinates_x', 'coordinates_y']:
                        if coords in anasig.array_annotations:
                            self.assertEqual(len(np.unique(anasig.array_annotations[coords])), 10)
                        self.assertIn('connector_aligned_ids', anasig.array_annotations)


def suite():
    suite = unittest.makeSuite(RGIOTestCase, 'test')
    return suite

if __name__ == "__main__":
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite())