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
|
# -*- coding: utf-8 -*-
from __future__ import division, print_function, absolute_import
from itertools import chain
from tempfile import NamedTemporaryFile
import pytest
from petl.test.helpers import ieq
from petl.transform.sorts import sort
import petl as etl
from petl.io.pytables import fromhdf5, fromhdf5sorted, tohdf5, appendhdf5
try:
import tables # noqa: F401
except ImportError as e:
pytest.skip('SKIP pytables tests: %s' % e, allow_module_level=True)
else:
class FooBar(tables.IsDescription):
foo = tables.Int32Col(pos=0)
bar = tables.StringCol(6, pos=2)
def test_fromhdf5():
f = NamedTemporaryFile()
# set up a new hdf5 table to work with
h5file = tables.open_file(f.name, mode='w', title='Test file')
h5file.create_group('/', 'testgroup', 'Test Group')
h5table = h5file.create_table('/testgroup', 'testtable', FooBar, 'Test Table')
# load some data into the table
table1 = (('foo', 'bar'),
(1, b'asdfgh'),
(2, b'qwerty'),
(3, b'zxcvbn'))
for row in table1[1:]:
for i, fld in enumerate(table1[0]):
h5table.row[fld] = row[i]
h5table.row.append()
h5file.flush()
h5file.close()
# verify we can get the data back out
table2a = fromhdf5(f.name, '/testgroup', 'testtable')
ieq(table1, table2a)
ieq(table1, table2a)
# verify we can get the data back out
table2b = fromhdf5(f.name, '/testgroup/testtable')
ieq(table1, table2b)
ieq(table1, table2b)
# verify using an existing tables.File object
h5file = tables.open_file(f.name)
table3 = fromhdf5(h5file, '/testgroup/testtable')
ieq(table1, table3)
# verify using an existing tables.Table object
h5tbl = h5file.get_node('/testgroup/testtable')
table4 = fromhdf5(h5tbl)
ieq(table1, table4)
# verify using a condition to filter data
table5 = fromhdf5(h5tbl, condition="(foo < 3)")
ieq(table1[:3], table5)
# clean up
h5file.close()
def test_fromhdf5sorted():
f = NamedTemporaryFile()
# set up a new hdf5 table to work with
h5file = tables.open_file(f.name, mode='w', title='Test file')
h5file.create_group('/', 'testgroup', 'Test Group')
h5table = h5file.create_table('/testgroup', 'testtable', FooBar,
'Test Table')
# load some data into the table
table1 = (('foo', 'bar'),
(3, b'asdfgh'),
(2, b'qwerty'),
(1, b'zxcvbn'))
for row in table1[1:]:
for i, f in enumerate(table1[0]):
h5table.row[f] = row[i]
h5table.row.append()
h5table.cols.foo.create_csindex()
h5file.flush()
# verify we can get the data back out
table2 = fromhdf5sorted(h5table, sortby='foo')
ieq(sort(table1, 'foo'), table2)
ieq(sort(table1, 'foo'), table2)
# clean up
h5file.close()
def test_tohdf5():
f = NamedTemporaryFile()
# set up a new hdf5 table to work with
h5file = tables.open_file(f.name, mode="w", title="Test file")
h5file.create_group('/', 'testgroup', 'Test Group')
h5file.create_table('/testgroup', 'testtable', FooBar, 'Test Table')
h5file.flush()
h5file.close()
# load some data via tohdf5
table1 = (('foo', 'bar'),
(1, b'asdfgh'),
(2, b'qwerty'),
(3, b'zxcvbn'))
tohdf5(table1, f.name, '/testgroup', 'testtable')
ieq(table1, fromhdf5(f.name, '/testgroup', 'testtable'))
tohdf5(table1, f.name, '/testgroup/testtable')
ieq(table1, fromhdf5(f.name, '/testgroup/testtable'))
h5file = tables.open_file(f.name, mode="a")
tohdf5(table1, h5file, '/testgroup/testtable')
ieq(table1, fromhdf5(h5file, '/testgroup/testtable'))
h5table = h5file.get_node('/testgroup/testtable')
tohdf5(table1, h5table)
ieq(table1, fromhdf5(h5table))
# clean up
h5file.close()
def test_tohdf5_create():
table1 = (('foo', 'bar'),
(1, b'asdfgh'),
(2, b'qwerty'),
(3, b'zxcvbn'))
f = NamedTemporaryFile()
# test creation with defined datatype
tohdf5(table1, f.name, '/testgroup', 'testtable', create=True,
drop=True, description=FooBar, createparents=True)
ieq(table1, fromhdf5(f.name, '/testgroup', 'testtable'))
# test dynamically determined datatype
tohdf5(table1, f.name, '/testgroup', 'testtable2', create=True,
drop=True, createparents=True)
ieq(table1, fromhdf5(f.name, '/testgroup', 'testtable2'))
def test_appendhdf5():
f = NamedTemporaryFile()
# set up a new hdf5 table to work with
h5file = tables.open_file(f.name, mode="w", title="Test file")
h5file.create_group('/', 'testgroup', 'Test Group')
h5file.create_table('/testgroup', 'testtable', FooBar, 'Test Table')
h5file.flush()
h5file.close()
# load some initial data via tohdf5()
table1 = (('foo', 'bar'),
(1, b'asdfgh'),
(2, b'qwerty'),
(3, b'zxcvbn'))
tohdf5(table1, f.name, '/testgroup', 'testtable')
ieq(table1, fromhdf5(f.name, '/testgroup', 'testtable'))
# append some more data
appendhdf5(table1, f.name, '/testgroup', 'testtable')
ieq(chain(table1, table1[1:]), fromhdf5(f.name, '/testgroup',
'testtable'))
def test_integration():
f = NamedTemporaryFile()
# set up a new hdf5 table to work with
h5file = tables.open_file(f.name, mode="w", title="Test file")
h5file.create_group('/', 'testgroup', 'Test Group')
h5file.create_table('/testgroup', 'testtable', FooBar, 'Test Table')
h5file.flush()
h5file.close()
# load some initial data via tohdf5()
table1 = etl.wrap((('foo', 'bar'),
(1, b'asdfgh'),
(2, b'qwerty'),
(3, b'zxcvbn')))
table1.tohdf5(f.name, '/testgroup', 'testtable')
ieq(table1, etl.fromhdf5(f.name, '/testgroup', 'testtable'))
# append some more data
table1.appendhdf5(f.name, '/testgroup', 'testtable')
ieq(chain(table1, table1[1:]), etl.fromhdf5(f.name, '/testgroup',
'testtable'))
|
