test_brainwaref32io.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148
  1. """
  2. Tests of neo.io.brainwaref32io
  3. """
  4. import os.path
  5. import unittest
  6. import numpy as np
  7. import quantities as pq
  8. from neo.core import Block, Segment, SpikeTrain, Group
  9. from neo.io import BrainwareF32IO
  10. from neo.test.iotest.common_io_test import BaseTestIO
  11. from neo.test.tools import (assert_same_sub_schema,
  12. assert_neo_object_is_compliant)
  13. from neo.test.iotest.tools import create_generic_reader
  14. def proc_f32(filename):
  15. '''Load an f32 file that has already been processed by the official matlab
  16. file converter. That matlab data is saved to an m-file, which is then
  17. converted to a numpy '.npz' file. This numpy file is the file actually
  18. loaded. This function converts it to a neo block and returns the block.
  19. This block can be compared to the block produced by BrainwareF32IO to
  20. make sure BrainwareF32IO is working properly
  21. block = proc_f32(filename)
  22. filename: The file name of the numpy file to load. It should end with
  23. '*_f32_py?.npz'. This will be converted to a neo 'file_origin' property
  24. with the value '*.f32', so the filename to compare should fit that pattern.
  25. 'py?' should be 'py2' for the python 2 version of the numpy file or 'py3'
  26. for the python 3 version of the numpy file.
  27. example: filename = 'file1_f32_py2.npz'
  28. f32 file name = 'file1.f32'
  29. '''
  30. filenameorig = os.path.basename(filename[:-12] + '.f32')
  31. # create the objects to store other objects
  32. block = Block(file_origin=filenameorig)
  33. gr = Group(file_origin=filenameorig)
  34. block.groups.append(gr)
  35. try:
  36. with np.load(filename, allow_pickle=True) as f32obj:
  37. f32file = list(f32obj.items())[0][1].flatten()
  38. except OSError as exc:
  39. if 'as a pickle' in exc.message:
  40. block.create_many_to_one_relationship()
  41. return block
  42. else:
  43. raise
  44. sweeplengths = [res[0, 0].tolist() for res in f32file['sweeplength']]
  45. stims = [res.flatten().tolist() for res in f32file['stim']]
  46. sweeps = [res['spikes'].flatten() for res in f32file['sweep'] if res.size]
  47. fullf32 = zip(sweeplengths, stims, sweeps)
  48. for sweeplength, stim, sweep in fullf32:
  49. for trainpts in sweep:
  50. if trainpts.size:
  51. trainpts = trainpts.flatten().astype('float32')
  52. else:
  53. trainpts = []
  54. paramnames = ['Param%s' % i for i in range(len(stim))]
  55. params = dict(zip(paramnames, stim))
  56. train = SpikeTrain(trainpts, units=pq.ms,
  57. t_start=0, t_stop=sweeplength,
  58. file_origin=filenameorig)
  59. segment = Segment(file_origin=filenameorig, **params)
  60. segment.spiketrains = [train]
  61. gr.spiketrains.append(train)
  62. block.segments.append(segment)
  63. block.create_many_to_one_relationship()
  64. return block
  65. class BrainwareF32IOTestCase(BaseTestIO, unittest.TestCase):
  66. '''
  67. Unit test testcase for neo.io.BrainwareF32IO
  68. '''
  69. ioclass = BrainwareF32IO
  70. read_and_write_is_bijective = False
  71. # These are the files it tries to read and test for compliance
  72. files_to_test = ['block_300ms_4rep_1clust_part_ch1.f32',
  73. 'block_500ms_5rep_empty_fullclust_ch1.f32',
  74. 'block_500ms_5rep_empty_partclust_ch1.f32',
  75. 'interleaved_500ms_5rep_ch2.f32',
  76. 'interleaved_500ms_5rep_nospikes_ch1.f32',
  77. 'multi_500ms_mulitrep_ch1.f32',
  78. 'random_500ms_12rep_noclust_part_ch2.f32',
  79. 'sequence_500ms_5rep_ch2.f32']
  80. # add the appropriate suffix
  81. suffix = '_f32_py3.npz'
  82. files_to_download = files_to_test[:]
  83. # add the reference files to the list of files to download
  84. files_to_compare = []
  85. for fname in files_to_test:
  86. if fname:
  87. files_to_compare.append(os.path.splitext(fname)[0] + suffix)
  88. # Will fetch from g-node if they don't already exist locally
  89. # How does it know to do this before any of the other tests?
  90. files_to_download = files_to_test + files_to_compare
  91. def test_reading_same(self):
  92. for ioobj, path in self.iter_io_objects(return_path=True):
  93. obj_reader_base = create_generic_reader(ioobj, target=False)
  94. obj_reader_single = create_generic_reader(ioobj)
  95. obj_base = obj_reader_base()
  96. obj_single = obj_reader_single()
  97. try:
  98. assert_same_sub_schema(obj_base, [obj_single])
  99. except BaseException as exc:
  100. exc.args += ('from ' + os.path.basename(path),)
  101. raise
  102. def test_against_reference(self):
  103. for obj, path in self.iter_objects(return_path=True):
  104. filename = os.path.basename(path)
  105. refpath = os.path.splitext(path)[0] + self.suffix
  106. refobj = proc_f32(refpath)
  107. try:
  108. assert_neo_object_is_compliant(obj)
  109. assert_neo_object_is_compliant(refobj)
  110. assert_same_sub_schema(obj, refobj)
  111. except BaseException as exc:
  112. exc.args += ('from ' + filename,)
  113. raise
  114. if __name__ == '__main__':
  115. unittest.main()