Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def put(self, compressed):
bwritten = blosc.decompress_ptr(compressed, self.ptr)
self.ptr += bwritten
return bwritten
def put(self, compressed):
bwritten = blosc.decompress_ptr(compressed, self.ptr)
self.ptr += bwritten
dtic = time.time()
out = blosc.unpack_array(c)
dtoc = time.time()
assert((in_ == out).all())
print(" Time for pack_array/unpack_array: %.3f/%.3f s." % \
(ctoc-ctic, dtoc-dtic), end='')
print("\tCompr ratio: %.2f" % (in_.size*in_.dtype.itemsize*1. / len(c)))
ctic = time.time()
c = blosc.compress_ptr(in_.__array_interface__['data'][0],
in_.size, in_.dtype.itemsize,
clevel=clevel, shuffle=True, cname=cname)
ctoc = time.time()
out = np.empty(in_.size, dtype=in_.dtype)
dtic = time.time()
blosc.decompress_ptr(c, out.__array_interface__['data'][0])
dtoc = time.time()
assert((in_ == out).all())
print(" Time for compress_ptr/decompress_ptr: %.3f/%.3f s." % \
(ctoc-ctic, dtoc-dtic), end='')
print("\tCompr ratio: %.2f" % (in_.size*in_.dtype.itemsize*1. / len(c)))
def decompress(size, dtype, data):
out = np.empty(size, dtype)
blosc.decompress_ptr(data, out.__array_interface__['data'][0])
return out
print("\nTimes for compressing/decompressing with clevel=%d and %d threads" % (
clevel, blosc.ncores))
for (in_, label) in arrays:
print("\n*** %s ***" % label)
for cname in blosc.compressor_list():
for filter in [blosc.NOSHUFFLE, blosc.SHUFFLE, blosc.BITSHUFFLE]:
t0 = time.time()
c = blosc.compress_ptr(in_.__array_interface__['data'][0],
in_.size, in_.dtype.itemsize,
clevel=clevel, shuffle=filter, cname=cname)
tc = time.time() - t0
# cause page faults here
out = np.full(in_.size, fill_value=0, dtype=in_.dtype)
t0 = time.time()
blosc.decompress_ptr(c, out.__array_interface__['data'][0])
td = time.time() - t0
assert((in_ == out).all())
print(" *** %-8s, %-10s *** %6.3f s (%.2f GB/s) / %5.3f s (%.2f GB/s)" % (
cname, blosc.filters[filter], tc, ((N*8 / tc) / 2**30), td, ((N*8 / td) / 2**30)), end='')
print("\tCompr. ratio: %5.1fx" % (N*8. / len(c)))