1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
# -*- coding: utf-8 -*-
# Copyright © 2014, German Neuroinformatics Node (G-Node)
#
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted under the terms of the BSD License. See
# LICENSE file in the root of the Project.
import os
import unittest
import nixio as nix
from .tmp import TempDir
class TestBlock(unittest.TestCase):
def setUp(self):
self.tmpdir = TempDir("blocktest")
self.testfilename = os.path.join(self.tmpdir.path, "blocktest.nix")
self.file = nix.File.open(self.testfilename, nix.FileMode.Overwrite)
self.block = self.file.create_block("test block", "recordingsession")
self.other = self.file.create_block("other block", "recordingsession")
def tearDown(self):
del self.file.blocks[self.block.id]
del self.file.blocks[self.other.id]
self.file.close()
self.tmpdir.cleanup()
def test_block_eq(self):
assert self.block == self.block
assert not self.block == self.other
assert self.block is not None
def test_block_id(self):
assert self.block.id is not None
def test_block_name(self):
assert self.block.name is not None
def test_block_type(self):
def set_none():
self.block.type = None
assert self.block.type is not None
self.assertRaises(Exception, set_none)
self.block.type = "foo type"
assert self.block.type == "foo type"
def test_block_definition(self):
assert self.block.definition is None
self.block.definition = "definition"
assert self.block.definition == "definition"
self.block.definition = None
assert self.block.definition is None
def test_block_timestamps(self):
created_at = self.block.created_at
assert created_at > 0
updated_at = self.block.updated_at
assert updated_at > 0
self.block.force_created_at(1403530068)
assert self.block.created_at == 1403530068
def test_block_data_arrays(self):
assert len(self.block.data_arrays) == 0
data_array = self.block.create_data_array("test data_array",
"recordingsession",
nix.DataType.Int32, (0, ))
assert len(self.block.data_arrays) == 1
# TODO implement __eq__ for DataArray
assert data_array.id in self.block.data_arrays
assert "notexist" not in self.block.data_arrays
assert data_array.id == self.block.data_arrays[0].id
assert data_array.id == self.block.data_arrays[-1].id
del self.block.data_arrays[0]
assert len(self.block.data_arrays) == 0
data_array = self.block.create_data_array("test data_array",
"recordingsession",
nix.DataType.Int32, (0, ), label="voltage", unit="V")
assert len(self.block.data_arrays) == 1
assert data_array.unit == "V"
assert data_array.label == "voltage"
del self.block.data_arrays[0]
assert len(self.block.data_arrays) == 0
def test_block_multi_tags(self):
assert len(self.block.multi_tags) == 0
data_array = self.block.create_data_array("test array",
"recording",
nix.DataType.Int32, (0, ))
multi_tag = self.block.create_multi_tag("test multi_tag",
"recordingsession", data_array)
assert len(self.block.multi_tags) == 1
# TODO implement __eq__ for MultiTag
# assert multi_tag in self.block.multi_tags
assert multi_tag.id in self.block.multi_tags
assert "notexist" not in self.block.multi_tags
assert multi_tag.id == self.block.multi_tags[0].id
assert multi_tag.id == self.block.multi_tags[-1].id
del self.block.multi_tags[0]
assert len(self.block.multi_tags) == 0
def test_block_tags(self):
assert len(self.block.tags) == 0
tag = self.block.create_tag("test tag", "recordingsession", [0])
assert len(self.block.tags) == 1
assert tag in self.block.tags
assert tag.id in self.block.tags
assert "notexist" not in self.block.tags
assert tag.id == self.block.tags[0].id
assert tag.id == self.block.tags[-1].id
del self.block.tags[0]
assert len(self.block.tags) == 0
def test_block_sources(self):
assert len(self.block.sources) == 0
source = self.block.create_source("test source", "electrode")
assert len(self.block.sources) == 1
assert source in self.block.sources
assert source.id in self.block.sources
assert "notexist" not in self.block.sources
assert source.id == self.block.sources[0].id
assert source.id == self.block.sources[-1].id
del self.block.sources[0]
assert len(self.block.sources) == 0
def test_block_find_sources(self):
for i in range(2):
self.block.create_source("level1-p0-s" + str(i), "dummy")
for i in range(2):
self.block.sources[0].create_source("level2-p1-s" + str(i),
"dummy")
for i in range(2):
self.block.sources[1].create_source("level2-p2-s" + str(i),
"dummy")
for i in range(2):
self.block.sources[0].sources[0].create_source(
"level3-p1-s" + str(i), "dummy"
)
assert len(self.block.find_sources()) == 8
assert len(self.block.find_sources(limit=1)) == 2
assert len(self.block.find_sources(filtr=lambda x: "level2-p1-s" in x.name)) == 2
assert len(self.block.find_sources(filtr=lambda x: "level2-p1-s" in x.name, limit=1)) == 0
def test_block_groups(self):
assert len(self.block.groups) == 0
group = self.block.create_group("test group", "RecordingChannelGroup")
assert len(self.block.groups) == 1
assert group in self.block.groups
assert group.id in self.block.groups
assert "notexist" not in self.block.groups
assert group.id == self.block.groups[0].id
assert group.id == self.block.groups[-1].id
del self.block.groups[0]
assert len(self.block.groups) == 0
def test_copy_on_block(self):
tar_filename = os.path.join(self.tmpdir.path, "copytarget.nix")
tar_file = nix.File.open(tar_filename, nix.FileMode.Overwrite)
blk2 = tar_file.create_block("blk2", "blk")
data = [(1, 2, 3), (4, 5, 6)]
da = self.block.create_data_array("da1", 'grp da1', data=data)
mt = self.block.create_multi_tag("mt1", "some mt", da)
namelist = ['name', 'id', 'time']
dtlist = [nix.DataType.Int64, str,
nix.DataType.Float]
arr = [(1, "cat", 20.18), (2, 'dog', 20.15), (2, 'dog', 20.15)]
df = self.block.create_data_frame('test1', 'for_test',
col_names=namelist,
col_dtypes=dtlist, data=arr)
tag = self.block.create_tag("a tag", "some tag", position=(4, 5, 6))
blk2.create_multi_tag(copy_from=mt)
blk2.create_data_array(copy_from=da)
blk2.create_data_frame(copy_from=df)
blk2.create_tag(copy_from=tag)
assert mt == blk2.multi_tags[0]
assert da == blk2.data_arrays[0]
assert df == blk2.data_frames[0]
assert tag == blk2.tags[0]
tar_file.close()
self.block.create_data_array("da2", 'grp da2', data=[(1, 2, 3)])
da2 = self.block.data_arrays[1]
mt2 = self.block.multi_tags[0]
self.assertRaises(NameError, self.block.create_data_array,
copy_from=da2)
self.block.create_data_array(name="new da name", copy_from=da2)
self.block.create_multi_tag(name="new mt name", copy_from=mt2)
assert self.block.multi_tags[0] == mt2
assert self.block.data_arrays[1] == da2
|
