-
Notifications
You must be signed in to change notification settings - Fork 0
/
h5.py
162 lines (125 loc) · 5.59 KB
/
h5.py
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
#Python 3.5+
#Depends on addict https://github.com/mewwts/addict
#Depends on h5py, numpy
from addict import Dict
import warnings
with warnings.catch_warnings():
warnings.filterwarnings("ignore",category=FutureWarning)
import h5py
import numpy
import subprocess
import pprint
import copy
class H5():
pp = pprint.PrettyPrinter(indent=4)
def transform(self, x):
return x
def inverse_transform(self, x):
return x
def __init__(self, *data):
self.root = Dict()
self.filename = None
for i in data:
self.root.update(copy.deepcopy(i))
def load(self):
if self.filename is not None:
with h5py.File(self.filename, 'r') as h5file:
self.root = Dict(recursively_load(h5file, '/', self.inverse_transform))
else:
print('Filename must be set before load can be used')
def save(self):
if self.filename is not None:
with h5py.File(self.filename, 'w') as h5file:
recursively_save(h5file, '/', self.root, self.transform)
else:
print("Filename must be set before save can be used")
def append(self):
"This can only be used to write new keys to the system, this is faster than having to read the data before writing it"
if self.filename is not None:
with h5py.File(self.filename, 'a') as h5file:
recursively_save(h5file, '/', self.root, self.transform)
else:
print("Filename must be set before save can be used")
def __str__(self):
temp = []
temp.append('Filename = %s' % self.filename)
temp.append(self.pp.pformat(self.root))
return '\n'.join(temp)
def update(self, merge):
self.root.update(merge.root)
def __getitem__(self, key):
key = key.lower()
obj = self.root
for i in key.split('/'):
if i:
obj = obj[i]
return obj
def __setitem__(self, key, value):
key = key.lower()
obj = self.root
parts = key.split('/')
for i in parts[:-1]:
if i:
obj = obj[i]
obj[parts[-1]] = value
def recursively_load( h5file, path, func):
ans = {}
for key, item in h5file[path].items():
key = func(key)
if isinstance(item, h5py._hl.dataset.Dataset):
ans[key] = item[()]
elif isinstance(item, h5py._hl.group.Group):
ans[key] = recursively_load(h5file, path + key + '/', func)
return ans
def recursively_save( h5file, path, dic, func):
# argument type checking
if not isinstance(dic, dict):
raise ValueError("must provide a dictionary")
if not isinstance(path, str):
raise ValueError("path must be a string")
if not isinstance(h5file, h5py._hl.files.File):
raise ValueError("must be an open h5py file")
# save items to the hdf5 file
for key, item in dic.items():
key = str(key)
if not isinstance(key, str):
raise ValueError("dict keys must be strings to save to hdf5")
#handle int, float, string and ndarray of int32, int64, float64
if isinstance(item, str):
h5file[path + func(key)] = numpy.array(item, dtype='S')
elif isinstance(item, int):
h5file[path + func(key)] = numpy.array(item, dtype=numpy.int32)
elif isinstance(item, float):
h5file[path + func(key)] = numpy.array(item, dtype=numpy.float64)
elif isinstance(item, numpy.ndarray) and item.dtype == numpy.float64:
h5file[path + func(key)] = item
elif isinstance(item, numpy.ndarray) and item.dtype == numpy.float32:
h5file[path + func(key)] = numpy.array(item, dtype=numpy.float64)
elif isinstance(item, numpy.ndarray) and item.dtype == numpy.int32:
h5file[path + func(key)] = item
elif isinstance(item, numpy.ndarray) and item.dtype == numpy.int64:
h5file[path + func(key)] = item.astype(numpy.int32)
elif isinstance(item, numpy.ndarray) and item.dtype.kind == 'S':
h5file[path + func(key)] = item
elif isinstance(item, list) and all(isinstance(i, int) for i in item):
h5file[path + func(key)] = numpy.array(item, dtype=numpy.int32)
elif isinstance(item, list) and any(isinstance(i, float) for i in item):
h5file[path + func(key)] = numpy.array(item, dtype=numpy.float64)
elif isinstance(item, numpy.int32):
h5file[path + func(key)] = item
elif isinstance(item, numpy.float64):
h5file[path + func(key)] = item
elif isinstance(item, numpy.float32):
h5file[path + func(key)] = numpy.array(item, dtype=numpy.float64)
elif isinstance(item, numpy.bytes_):
h5file[path + func(key)] = item
elif isinstance(item, bytes):
h5file[path + func(key)] = item
elif isinstance(item, list) and all(isinstance(i, str) for i in item):
h5file[path + func(key)] = numpy.array(item, dtype="S")
# save dictionaries
elif isinstance(item, dict):
recursively_save(h5file, path + key + '/', item, func)
# other types cannot be saved and will result in an error
else:
raise ValueError('Cannot save %s/%s key with %s type.' % (path, func(key), type(item)))