obci.analysis.obci_signal_processing.tests package¶
Submodules¶
obci.analysis.obci_signal_processing.tests.test_data_file_proxy module¶
>>> from obci.analysis.obci_signal_processing.signal.data_read_proxy import DataReadProxy
>>> from obci.analysis.obci_signal_processing.signal.data_raw_write_proxy import DataRawWriteProxy
>>> import os.path, os
>>> px = DataRawWriteProxy('./tescik.obci.dat')
>>> px.set_data_len(1, 1)
>>> px.data_received(1.2)
>>> px.data_received(0.0023)
>>> px.data_received(-123.456)
>>> px.data_received(3.3)
>>> px.data_received(5.0)
>>> nic = px.finish_saving()
>>> f = './tescik.obci.dat'
>>> py = DataReadProxy(f)
>>> abs(1.2 - py.get_next_value()) < 0.0001
True
>>> abs(0.0023-py.get_next_value()) < 0.000000001
True
>>> abs(-123.456-py.get_next_value()) < 0.00001
True
>>> py.goto_value(1)
>>> abs(0.0023-py.get_next_value()) < 0.000000001
True
>>> py.goto_value(4)
>>> abs(5.0-py.get_next_value()) < 0.0001
True
>>> py.goto_value(6)
>>> py.get_next_value()
Traceback (most recent call last):
...
obci.analysis.obci_signal_processing.signal.signal_exceptions.NoNextValue
>>> py.finish_reading()
>>> py.start_reading()
>>> vs = py.get_next_values(3)
>>> abs(vs[0]-1.20000000e+00)+abs(vs[1]-2.30000000e-03)+abs(vs[2]+1.23456000e+02)<3*0.0001
True
>>> py.get_next_values(3)
Traceback (most recent call last):
...
obci.analysis.obci_signal_processing.signal.signal_exceptions.NoNextValue
>>> #warning here
>>> py.finish_reading()
>>> os.remove(f)
obci.analysis.obci_signal_processing.tests.test_file_tags_writer_reader module¶
>>> from obci.analysis.obci_signal_processing.tags import tags_file_writer as p
>>> from obci.analysis.obci_signal_processing.tags import tags_file_reader as t
>>> px = p.TagsFileWriter('./tescik.obci.tags')
>>> px.tag_received({'start_timestamp':1001.0, 'end_timestamp':1002.0, 'name': 'nic', 'channels':'A B C', 'desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':1003.0, 'end_timestamp':1004.0, 'name': 'nic2', 'channels':'A B C', 'desc': {'x':1234, 'y':4567, 'z': 789}})
>>> px.tag_received({'start_timestamp':1005.0, 'end_timestamp':1006.0, 'name': 'nic3', 'channels':'A B C', 'desc': {'x':12345, 'y':45678, 'z': 789}})
>>> px.finish_saving(1000.0)
'./tescik.obci.tags'
>>> py = t.TagsFileReader('tescik.obci.tags')
>>> g = py.get_tags()[0]
>>> print(g['start_timestamp'])
1.0
>>> print(g['end_timestamp'])
2.0
>>> print(g['name'])
nic
>>> print(g['desc']['y'])
456
>>> print([int(t['start_timestamp']) for t in py.get_tags()])
[1, 3, 5]
>>> #REORDER AND TEST ORDERING AGAIN ....
>>> px = p.TagsFileWriter('./tescik.obci.tags')
>>> px.tag_received({'start_timestamp':1003.0, 'end_timestamp':1004.0, 'name': 'nic2', 'channels':'A B C', 'desc': {'x':1234, 'y':4567, 'z': 789}})
>>> px.tag_received({'start_timestamp':1005.0, 'end_timestamp':1006.0, 'name': 'nic3', 'channels':'A B C', 'desc': {'x':12345, 'y':45678, 'z': 789}})
>>> px.tag_received({'start_timestamp':1001.0, 'end_timestamp':1002.0, 'name': 'nic', 'channels':'A B C', 'desc': {'x':123, 'y':456, 'z': 789}})
>>> print([int(t['start_timestamp']) for t in py.get_tags()])
[1, 3, 5]
>>> import os
>>> import glob
>>> for fl in glob.glob('tescik*'): os.remove(fl)
obci.analysis.obci_signal_processing.tests.test_info_file_proxy module¶
>>> import os
>>> from obci.analysis.obci_signal_processing.signal import info_file_proxy as p
>>> px = p.InfoFileWriteProxy('./tescik.obci.svarog.info')
>>> px.set_attributes({'number_of_channels':2, 'sampling_frequency':128, 'channels_names': ['1','2'], 'file':'soufce.obci.dat', 'number_of_samples':3 } )
>>> px.finish_saving()
'./tescik.obci.svarog.info'
>>> py = p.InfoFileReadProxy('./tescik.obci.svarog.info')
>>> print(py.get_param('number_of_channels'))
2
>>> print(py.get_param('channels_names')[0])
1
>>> os.remove('./tescik.obci.svarog.info')
obci.analysis.obci_signal_processing.tests.test_read_data_source module¶
>>> from obci.analysis.obci_signal_processing.signal.data_raw_write_proxy import DataRawWriteProxy
>>> import os.path, os
>>> # PREPARE SOME SAMPLE FILE *************************************************
>>> px = DataRawWriteProxy('./tescik.obci.dat')
>>> px.data_received(1.2)
>>> px.data_received(0.0023)
>>> px.data_received(-123.456)
>>> px.data_received(3.3)
>>> px.data_received(5.0)
>>> px.data_received(0.0)
>>> nic = px.finish_saving()
>>> f = './tescik.obci.dat'
>>> from obci.analysis.obci_signal_processing.signal import read_data_source as s
>>> # TEST MEMORY DATA SOURCE **************************************************
>>> import numpy
>>> import numpy as np
>>> py = s.MemoryDataSource(numpy.zeros((2,3)))
>>> py.set_sample(0, [1.0, 2.0])
>>> py.set_sample(1, [3.0, 4.0])
>>> py.set_sample(2, [5.0, 6.0])
>>> [i for i in py.iter_samples()]
[array([ 1., 2.]), array([ 3., 4.]), array([ 5., 6.])]
>>> [i for i in py.iter_samples()]
[array([ 1., 2.]), array([ 3., 4.]), array([ 5., 6.])]
>>> py.get_samples(0, 1)
array([[ 1.],
[ 2.]])
>>> py.set_sample(3, [3.0, 4.0])
Traceback (most recent call last):
...
IndexError: index 3 is out of bounds for axis 1 with size 3
>>> py.set_sample(2, [1.0, 2.0, 3.0])
Traceback (most recent call last):
...
ValueError: cannot copy sequence with size 3 to array axis with dimension 2
>>> # TEST FILE DATA SOURCE ****************************************************
>>> py = s.FileDataSource(f, 2)
>>> py.get_samples(0, 0)
array([], shape=(2, 0), dtype=float64)
>>> np.abs(np.array([[ 1.20000000e+00, -1.23456000e+02], [ 2.30000000e-03, 3.30000000e+00]]) - py.get_samples(0, 2)) < 0.001
array([[ True, True],
[ True, True]], dtype=bool)
>>> py.get_samples(0, 10)
Traceback (most recent call last):
...
obci.analysis.obci_signal_processing.signal.signal_exceptions.NoNextValue
>>> np.abs(np.array([[ 1.20000000e+00, -1.23456000e+02, 5.00000000e+00], [ 2.30000000e-03, 3.30000000e+00, 0.00000000e+00]]) - py.get_samples(0, 3)) < 0.001
array([[ True, True, True],
[ True, True, True]], dtype=bool)
>>> py.get_samples(1, 3)
Traceback (most recent call last):
...
obci.analysis.obci_signal_processing.signal.signal_exceptions.NoNextValue
>>> np.abs(np.array([[ 1.20000000e+00, -1.23456000e+02, 5.00000000e+00], [ 2.30000000e-03, 3.30000000e+00, 0.00000000e+00]]) - py.get_samples()) < 0.001
array([[ True, True, True],
[ True, True, True]], dtype=bool)
>>> py = s.FileDataSource(f, 2)
>>> from numpy import array
>>> np.abs(array([[ 1.20000000e+00, -1.23456000e+02, 5.00000000e+00], [ 2.30000000e-03, 3.30000000e+00, 0.00000000e+00]]) - py.get_samples()) < 0.001
array([[ True, True, True],
[ True, True, True]], dtype=bool)
>>> py = s.FileDataSource(f, 2)
>>> np.abs(array([[ 1.20000000e+00, -1.23456000e+02, 5.00000000e+00], [ 2.30000000e-03, 3.30000000e+00, 0.00000000e+00]]) - py.get_samples()) < 0.001
array([[ True, True, True],
[ True, True, True]], dtype=bool)
>>> np.abs(array([[ 1.20000000e+00, -1.23456000e+02, 5.00000000e+00], [ 2.30000000e-03, 3.30000000e+00, 0.00000000e+00]]) - py.get_samples()) < 0.001
array([[ True, True, True],
[ True, True, True]], dtype=bool)
>>> [max(abs(i-y))<0.0001 for i,y in zip(py.iter_samples(), [array([ 1.2 , 0.0023]), array([-123.456, 3.3 ]), array([ 5., 0.])])]
[True, True, True]
>>> py = s.FileDataSource(f, 2)
>>> [max(abs(i-y))<0.0001 for i,y in zip(py.iter_samples(),[array([ 1.2 , 0.0023]), array([-123.456, 3.3 ]), array([ 5., 0.])])]
[True, True, True]
>>> [max(abs(i-y))<0.0001 for i,y in zip(py.iter_samples(), [array([ 1.2 , 0.0023]), array([-123.456, 3.3 ]), array([ 5., 0.])])]
[True, True, True]
>>> os.remove(f)
obci.analysis.obci_signal_processing.tests.test_read_manager module¶
>>> from obci.analysis.obci_signal_processing import read_manager
>>> import os, os.path, sys
>>> pth = os.path.abspath(read_manager.__file__)
>>> f = os.path.join(pth[:-len(os.path.basename(pth))], 'tests', 'data', 'data')
>>> mgr = read_manager.ReadManager(f+'.obci.info', f+'.obci.dat', f+'.obci.tags')
>>> mgr.get_samples(0, 1)[0,0]
-27075.0
>>> mgr.get_samples(0, 1)[1,0]
39641.0
>>> ch = mgr.get_samples()
>>> ch[0][0]
-27075.0
>>> ch[1][0]
39641.0
>>> [len(x) for x in ch]
[112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407, 112407]
>>> mgr.get_param(u'number_of_samples')
'2810175'
>>> mgr.get_param(u'sampling_frequency')
'128'
>>> mgr.get_param(u'number_of_channels')
'25'
>>> mgr.get_param('channels_names')
['Fp1', 'Fpz', 'Fp2', 'F7', 'F3', 'Fz', 'F4', 'F8', 'M1', 'C7', 'C3', 'Cz', 'C4', 'T8', 'M2', 'P7', 'P3', 'Pz', 'P4', 'P8', 'O1', 'Oz', 'O2', 'NIC', 'OKO_GORA_DOL']
>>> mgr.get_param('im_not_there')
Traceback (most recent call last):
...
obci.analysis.obci_signal_processing.signal.signal_exceptions.NoParameter: No parameter 'im_not_there' was found in info source!
>>> import numpy as np
>>> np.sum([np.abs(t['start_timestamp']-et)<0.0001 for t,et in zip(mgr.iter_tags(), [0.36085605621337891, 1.3654811382293701, 3.2938590049743652, 4.8368752002716064, 6.6318840980529785,8.2858760356903076, 9.8918850421905518, 11.599879026412964, 13.15486216545105, 14.747888088226318, 15.760866165161133,17.531879186630249, 19.168869018554688, 21.020870208740234, 22.716873168945312, 24.6128830909729, 26.412859201431274,28.320885181427002, 29.348877191543579, 30.735865116119385, 31.838881015777588, 33.505880117416382, 34.594882011413574,36.5418860912323, 37.700882196426392, 38.795868158340454, 40.185863971710205, 41.975889205932617, 43.608882188796997,44.695883989334106, 45.91289210319519, 47.635885000228882, 49.343885183334351, 50.655886173248291, 52.486284017562866,53.563876152038574, 55.53387713432312, 57.343875169754028, 59.317936182022095, 61.018882989883423, 62.311890125274658,63.758885145187378, 65.596870183944702, 67.496883153915405, 68.828874111175537, 70.646870136260986, 71.883864164352417,73.352570056915283, 75.352887153625488, 77.352883100509644, 78.996880054473877])])
51
>>> i = mgr.iter_tags()
>>> next(i)=={'channels': '', 'start_timestamp': 0.36085605621337891, 'desc': {u'value': u'1'}, 'name': u'trigger','end_timestamp': 1.3654811382293701}
True
>>> next(i)=={'channels': '', 'start_timestamp': 1.3654811382293701, 'desc': {u'value': u'0'}, 'name': u'trigger','end_timestamp': 3.2938590049743652}
True
obci.analysis.obci_signal_processing.tests.test_read_tags_source module¶
>>> tags = []
>>> tags.append({'start_timestamp':1001.0, 'end_timestamp':1002.0, 'name': 'nic', 'channels':'A B C','desc': {'x':123, 'y':456, 'z': 789}})
>>> tags.append({'start_timestamp':1003.0, 'end_timestamp':1004.0, 'name': 'nic2', 'channels':'A B C','desc': {'x':1234, 'y':4567, 'z': 789}})
>>> tags.append({'start_timestamp':1005.0, 'end_timestamp':1006.0, 'name': 'nic3', 'channels':'A B C','desc': {'x':12345, 'y':45678, 'z': 789}})
>>> tags.append({'start_timestamp':1005.5, 'end_timestamp':1007.0, 'name': 'nic2', 'channels':'A B C','desc': {'x':12345, 'y':45678, 'z': 789}})
>>> tags.append({'start_timestamp':1008.0, 'end_timestamp':1009.0, 'name': 'nic3', 'channels':'A B C','desc': {'x':12345, 'y':45678, 'z': 789}})
>>> from obci.analysis.obci_signal_processing.tags import read_tags_source
>>> s = read_tags_source.MemoryTagsSource(tags)
>>> s.get_tags()==[{'channels': 'A B C', 'start_timestamp': 1001.0, 'desc': {'y': 456, 'x': 123, 'z': 789}, 'name': 'nic','end_timestamp': 1002.0}, {'channels': 'A B C', 'start_timestamp': 1003.0, 'desc': {'y': 4567, 'x': 1234, 'z': 789},'name': 'nic2', 'end_timestamp': 1004.0}, {'channels': 'A B C', 'start_timestamp': 1005.0,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic3', 'end_timestamp': 1006.0},{'channels': 'A B C', 'start_timestamp': 1005.5, 'desc': {'y': 45678, 'x': 12345, 'z': 789},'name': 'nic2', 'end_timestamp': 1007.0}, {'channels': 'A B C', 'start_timestamp': 1008.0,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic3', 'end_timestamp': 1009.0}]
True
>>> s.get_tags('nic2')==[{'channels': 'A B C', 'start_timestamp': 1003.0, 'desc': {'y': 4567, 'x': 1234, 'z': 789}, 'name': 'nic2','end_timestamp': 1004.0}, {'channels': 'A B C', 'start_timestamp': 1005.5,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic2', 'end_timestamp': 1007.0}]
True
>>> s.get_tags(None, 1005.0, 1.0)==[{'channels': 'A B C', 'start_timestamp': 1005.0, 'desc': {'y': 45678, 'x': 12345, 'z': 789},'name': 'nic3', 'end_timestamp': 1006.0}, {'channels': 'A B C', 'start_timestamp': 1005.5,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic2', 'end_timestamp': 1007.0}]
True
>>> s.get_tags('nic3', 1005.0, 1.0)==[{'channels': 'A B C', 'start_timestamp': 1005.0, 'desc': {'y': 45678, 'x': 12345, 'z': 789},'name': 'nic3', 'end_timestamp': 1006.0}]
True
>>> s.get_tags(None, None, None, lambda tag: tag['desc']['x'] == 12345)==[{'channels': 'A B C', 'start_timestamp': 1005.0, 'desc': {'y': 45678, 'x': 12345, 'z': 789},'name': 'nic3', 'end_timestamp': 1006.0}, {'channels': 'A B C', 'start_timestamp': 1005.5,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic2', 'end_timestamp': 1007.0},{'channels': 'A B C', 'start_timestamp': 1008.0,'desc': {'y': 45678, 'x': 12345, 'z': 789}, 'name': 'nic3', 'end_timestamp': 1009.0}]
True
obci.analysis.obci_signal_processing.tests.test_smart_tag_definition module¶
>>> from obci.analysis.obci_signal_processing.tags import smart_tag_definition as d
>>> x = d.SmartTagDurationDefinition(start_tag_name='x', start_offset=1, end_offset=0, duration=21)
>>> print(x.__dict__['start_tag_name'])
x
>>> print(x.__dict__['start_offset'])
1
>>> print(x.__dict__['end_offset'])
0
>>> print(x.__dict__['duration'])
21
obci.analysis.obci_signal_processing.tests.test_smart_tags_manager module¶
>>> import os, os.path
>>> from obci.analysis.obci_signal_processing import smart_tags_manager as mgr
>>> from obci.analysis.obci_signal_processing.tags import smart_tag_definition as df
>>> d = df.SmartTagDurationDefinition(start_tag_name='trigger', start_offset=0, end_offset=0, duration=1.0)
>>> pth = os.path.abspath(mgr.__file__)
>>> f = {'info': os.path.join(pth[:-len(os.path.basename(pth))], 'tests', 'data', 'nic.obci.svarog.info'),'data':'nic.obci.dat', 'tags': os.path.join(pth[:-len(os.path.basename(pth))], 'tests', 'data', 'nic.obci.svarog.tags')}
>>> fabricate_data_file(f['data'])
>>> m = mgr.SmartTagsManager(d, f['info'], f['data'], f['tags'])
>>> print(len(m._smart_tags))
51
>>> tags, num = iter_all_tags(m)
>>> print(num)
51
>>> abs(tags[0].get_start_timestamp()-0.36085605621337891)<0.0001
True
>>> abs(tags[50].get_end_timestamp()-79.996880054473877)<0.0001
True
>>> abs(tags[2].get_channel_samples('Fp1')[0]-19389.0)<0.0001
True
>>> print(tags_len_ok(tags))
True
>>> dd = df.SmartTagEndTagDefinition(start_tag_name='trigger', start_offset=0, end_offset=0, end_tags_names=['trigger'])
>>> mm = mgr.SmartTagsManager(dd, f['info'], f['data'], f['tags'])
>>> tags, num = iter_all_tags(mm)
>>> print(num)
50
>>> print(not_duration_tags_data_len(tags))
20131
>>> print(not_duration_tags_data_sub(tags))
True
>>> # Another test
>>> from ..tags import tags_file_writer as p
>>> px = p.TagsFileWriter('./tescik.obci.tags')
>>> px.tag_received({'start_timestamp':1.0, 'end_timestamp':5.0, 'name': 'A', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':6.0, 'end_timestamp':10.0, 'name': 'A', 'channels':'','desc': {'x':1234, 'y':4567, 'z': 789}})
>>> px.tag_received({'start_timestamp':6.1, 'end_timestamp':6.5, 'name': 'B', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':7.0, 'end_timestamp':7.0, 'name': 'B', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':7.5, 'end_timestamp':8.7, 'name': 'B', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':9.0, 'end_timestamp':15.0, 'name': 'A', 'channels':'','desc': {'x':12345, 'y':45678, 'z': 789}})
>>> px.tag_received({'start_timestamp':10.2, 'end_timestamp':11.0, 'name': 'B', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':12.0, 'end_timestamp':12.5, 'name': 'B', 'channels':'','desc': {'x':123, 'y':456, 'z': 789}})
>>> px.tag_received({'start_timestamp':20.0, 'end_timestamp':25.0, 'name': 'A', 'channels':'','desc': {'x':12345, 'y':45678, 'z': 789}})
>>> px.finish_saving(0.0)
'./tescik.obci.tags'
>>> fabricate_data_file('./tescik.obci.dat', 1)
>>> fabricate_info_file('./tescik.obci.info', 1)
>>> dd = df.SmartTagEndTagDefinition(start_tag_name='A', start_offset=0, end_offset=0, end_tags_names=['A'])
>>> mm = mgr.SmartTagsManager(dd, './tescik.obci.info', './tescik.obci.dat', './tescik.obci.tags')
>>> tags = [i for i in mm.iter_smart_tags()]
>>> len(tags)
3
>>> tags2 = [i for i in mm]
>>> len(tags2)
3
>>> tags == tags2
True
>>> print(tags[1].get_samples()[0][0])
768.0
>>> dd = df.SmartTagDurationDefinition(start_tag_name='B', start_offset=0, end_offset=0, duration=1.0)
>>> mm = mgr.SmartTagsManager(dd, None, None, None, tags[1])
>>> tss = [i for i in mm.iter_smart_tags()]
>>> len(tss)
3
>>> tss_tags = [t.get_start_tag() for t in tss]
>>> tss_tags ==[{'channels': '', 'start_timestamp': 6.0999999999999996, 'desc': {u'y': u'456', u'x': u'123', u'z': u'789'},'name': u'B', 'end_timestamp': 6.5}, {'channels': '', 'start_timestamp': 7.0,'desc': {u'y': u'456', u'x': u'123', u'z': u'789'}, 'name': u'B', 'end_timestamp': 7.0},{'channels': '', 'start_timestamp': 7.5, 'desc': {u'y': u'456', u'x': u'123', u'z': u'789'},'name': u'B', 'end_timestamp': 8.6999999999999993}]
True
>>> print(tss[0].get_samples()[0][0])
780.0
>>> dd = df.SmartTagDurationDefinition(start_tag_name='B', start_offset=0, end_offset=0, duration=1.0)
>>> mm = mgr.SmartTagsManager(dd, None, None, None, tags[2])
>>> tss = [i for i in mm.iter_smart_tags()]
>>> len(tss)
2
>>> tss_tags = [t.get_start_tag() for t in tss]
>>> tss_tags==[{'channels': '', 'start_timestamp': 10.199999999999999, 'desc': {u'y': u'456', u'x': u'123', u'z': u'789'},'name': u'B', 'end_timestamp': 11.0}, {'channels': '', 'start_timestamp': 12.0,'desc': {u'y': u'456', u'x': u'123', u'z': u'789'}, 'name': u'B', 'end_timestamp': 12.5}]
True
>>> import os
>>> import glob
>>> for fl in glob.glob('tescik*'): os.remove(fl)
>>> for fl in glob.glob('nic.obci*'): os.remove(fl)
True if data in tags are sequential - they should be, as we have trigger-to-trigger tags and data in file are sequential.