File: C:/Users/fred/anaconda3/Lib/site-packages/tables/tests/test_carray.py
import sys
from pathlib import Path
import numpy as np
import tables as tb
from tables.tests import common
def foreign_byteorder():
return {'little': 'big', 'big': 'little'}[sys.byteorder]
class BasicTestCase(common.TempFileMixin, common.PyTablesTestCase):
# Default values
obj = None
flavor = "numpy"
type = 'int32'
shape = (2, 2)
start = 0
stop = 10
step = 1
length = 1
chunkshape = (5, 5)
byteorder = None
compress = 0
complib = "zlib" # Default compression library
shuffle = 0
bitshuffle = 0
fletcher32 = 0
reopen = 1 # Tells whether the file has to be reopened on each test or not
def setUp(self):
super().setUp()
# Create an instance of an HDF5 Table
self.rootgroup = self.h5file.root
self.populateFile()
if self.reopen:
# Close the file
self.h5file.close()
def populateFile(self):
group = self.rootgroup
obj = self.obj
if obj is None:
if self.type == "string":
atom = tb.StringAtom(itemsize=self.length)
else:
atom = tb.Atom.from_type(self.type)
else:
atom = None
title = self.__class__.__name__
filters = tb.Filters(complevel=self.compress,
complib=self.complib,
shuffle=self.shuffle,
bitshuffle=self.bitshuffle,
fletcher32=self.fletcher32)
carray = self.h5file.create_carray(group, 'carray1',
atom=atom, shape=self.shape,
title=title, filters=filters,
chunkshape=self.chunkshape,
byteorder=self.byteorder,
obj=obj)
carray.flavor = self.flavor
# Fill it with data
self.rowshape = list(carray.shape)
self.objsize = self.length * np.prod(carray.shape)
if self.flavor == "numpy":
if self.type == "string":
object = np.ndarray(buffer=b"a"*self.objsize,
shape=self.shape,
dtype="S%s" % carray.atom.itemsize)
else:
object = np.arange(self.objsize, dtype=carray.atom.dtype)
object.shape = carray.shape
if common.verbose:
print("Object to append -->", repr(object))
carray[...] = object
def _get_shape(self):
if self.shape is not None:
shape = self.shape
else:
shape = np.asarray(self.obj).shape
return shape
def test00_attributes(self):
if self.reopen:
self.h5file = tb.open_file(self.h5fname, "r")
obj = self.h5file.get_node("/carray1")
shape = self._get_shape()
self.assertEqual(obj.flavor, self.flavor)
self.assertEqual(obj.shape, shape)
self.assertEqual(obj.ndim, len(shape))
self.assertEqual(obj.chunkshape, self.chunkshape)
self.assertEqual(obj.nrows, shape[0])
self.assertEqual(obj.atom.type, self.type)
def test01_readCArray(self):
"""Checking read() of chunked layout arrays."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_readCArray..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
if self.reopen:
self.h5file = tb.open_file(self.h5fname, "r")
carray = self.h5file.get_node("/carray1")
# Choose a small value for buffer size
carray.nrowsinbuf = 3
if common.verbose:
print("CArray descr:", repr(carray))
print("shape of read array ==>", carray.shape)
print("reopening?:", self.reopen)
shape = self._get_shape()
# Build the array to do comparisons
if self.flavor == "numpy":
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.shape,
dtype=f"S{carray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=carray.atom.dtype)
object_.shape = shape
stop = self.stop
# stop == None means read only the element designed by start
# (in read() contexts)
if self.stop is None:
if self.start == -1: # corner case
stop = carray.nrows
else:
stop = self.start + 1
# Protection against number of elements less than existing
# if rowshape[self.extdim] < self.stop or self.stop == 0:
if carray.nrows < stop:
# self.stop == 0 means last row only in read()
# and not in [::] slicing notation
stop = int(carray.nrows)
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
obj = object_[self.start:stop:self.step].copy()
# Read all the array
try:
data = carray.read(self.start, stop, self.step)
except IndexError:
if self.flavor == "numpy":
data = np.empty(shape=self.shape, dtype=self.type)
else:
data = np.empty(shape=self.shape, dtype=self.type)
if common.verbose:
if hasattr(obj, "shape"):
print("shape should look as:", obj.shape)
print("Object read ==>", repr(data))
print("Should look like ==>", repr(obj))
if hasattr(data, "shape"):
self.assertEqual(len(data.shape), len(shape))
else:
# Scalar case
self.assertEqual(len(self.shape), 1)
self.assertEqual(carray.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(data, obj, self.flavor))
def test01_readCArray_out_argument(self):
"""Checking read() of chunked layout arrays."""
# Create an instance of an HDF5 Table
if self.reopen:
self.h5file = tb.open_file(self.h5fname, "r")
carray = self.h5file.get_node("/carray1")
shape = self._get_shape()
# Choose a small value for buffer size
carray.nrowsinbuf = 3
# Build the array to do comparisons
if self.flavor == "numpy":
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.shape,
dtype=f"S{carray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=carray.atom.dtype)
object_.shape = shape
stop = self.stop
# stop == None means read only the element designed by start
# (in read() contexts)
if self.stop is None:
if self.start == -1: # corner case
stop = carray.nrows
else:
stop = self.start + 1
# Protection against number of elements less than existing
# if rowshape[self.extdim] < self.stop or self.stop == 0:
if carray.nrows < stop:
# self.stop == 0 means last row only in read()
# and not in [::] slicing notation
stop = int(carray.nrows)
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
obj = object_[self.start:stop:self.step].copy()
# Read all the array
try:
data = np.empty(shape, dtype=carray.atom.dtype)
data = data[self.start:stop:self.step].copy()
carray.read(self.start, stop, self.step, out=data)
except IndexError:
if self.flavor == "numpy":
data = np.empty(shape=shape, dtype=self.type)
else:
data = np.empty(shape=shape, dtype=self.type)
if hasattr(data, "shape"):
self.assertEqual(len(data.shape), len(shape))
else:
# Scalar case
self.assertEqual(len(shape), 1)
self.assertEqual(carray.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(data, obj, self.flavor))
def test02_getitemCArray(self):
"""Checking chunked layout array __getitem__ special method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_getitemCArray..." %
self.__class__.__name__)
if not hasattr(self, "slices"):
# If there is not a slices attribute, create it
self.slices = (slice(self.start, self.stop, self.step),)
# Create an instance of an HDF5 Table
if self.reopen:
self.h5file = tb.open_file(self.h5fname, "r")
carray = self.h5file.get_node("/carray1")
if common.verbose:
print("CArray descr:", repr(carray))
print("shape of read array ==>", carray.shape)
print("reopening?:", self.reopen)
shape = self._get_shape()
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.shape,
dtype=f"S{carray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=carray.atom.dtype)
object_.shape = shape
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
obj = object_.__getitem__(self.slices).copy()
# Read data from the array
try:
data = carray.__getitem__(self.slices)
except IndexError:
print("IndexError!")
if self.flavor == "numpy":
data = np.empty(shape=self.shape, dtype=self.type)
else:
data = np.empty(shape=self.shape, dtype=self.type)
if common.verbose:
print("Object read:\n", repr(data)) # , data.info()
print("Should look like:\n", repr(obj)) # , object.info()
if hasattr(obj, "shape"):
print("Original object shape:", self.shape)
print("Shape read:", data.shape)
print("shape should look as:", obj.shape)
if not hasattr(data, "shape"):
# Scalar case
self.assertEqual(len(self.shape), 1)
self.assertEqual(carray.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(data, obj, self.flavor))
def test03_setitemCArray(self):
"""Checking chunked layout array __setitem__ special method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_setitemCArray..." %
self.__class__.__name__)
if not hasattr(self, "slices"):
# If there is not a slices attribute, create it
self.slices = (slice(self.start, self.stop, self.step),)
# Create an instance of an HDF5 Table
if self.reopen:
self.h5file = tb.open_file(self.h5fname, "a")
carray = self.h5file.get_node("/carray1")
if common.verbose:
print("CArray descr:", repr(carray))
print("shape of read array ==>", carray.shape)
print("reopening?:", self.reopen)
shape = self._get_shape()
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.shape,
dtype=f"S{carray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=carray.atom.dtype)
object_.shape = shape
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
obj = object_.__getitem__(self.slices).copy()
if self.type == "string":
if hasattr(self, "wslice"):
obj[self.wslize] = "xXx"
carray[self.wslice] = "xXx"
elif sum(obj[self.slices].shape) != 0:
obj[:] = "xXx"
if obj.size > 0:
carray[self.slices] = obj
else:
if hasattr(self, "wslice"):
obj[self.wslice] = obj[self.wslice] * 2 + 3
carray[self.wslice] = carray[self.wslice] * 2 + 3
elif sum(obj[self.slices].shape) != 0:
obj = obj * 2 + 3
if np.prod(obj.shape) > 0:
carray[self.slices] = carray[self.slices] * 2 + 3
# Cast again object to its original type
obj = np.array(obj, dtype=carray.atom.dtype)
# Read datafrom the array
try:
data = carray.__getitem__(self.slices)
except IndexError:
print("IndexError!")
if self.flavor == "numpy":
data = np.empty(shape=self.shape, dtype=self.type)
else:
data = np.empty(shape=self.shape, dtype=self.type)
if common.verbose:
print("Object read:\n", repr(data)) # , data.info()
print("Should look like:\n", repr(obj)) # , object.info()
if hasattr(obj, "shape"):
print("Original object shape:", self.shape)
print("Shape read:", data.shape)
print("shape should look as:", obj.shape)
if not hasattr(data, "shape"):
# Scalar case
self.assertEqual(len(self.shape), 1)
self.assertEqual(carray.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(data, obj, self.flavor))
class BasicWriteTestCase(BasicTestCase):
type = 'int32'
shape = (2,)
chunkshape = (5,)
step = 1
wslice = 1 # single element case
class BasicWrite2TestCase(BasicTestCase):
type = 'int32'
shape = (2,)
chunkshape = (5,)
step = 1
wslice = slice(shape[0]-2, shape[0], 2) # range of elements
reopen = 0 # This case does not reopen files
class BasicWrite3TestCase(BasicTestCase):
obj = [1, 2]
type = np.asarray(obj).dtype.name
shape = None
chunkshape = (5,)
step = 1
reopen = 0 # This case does not reopen files
class BasicWrite4TestCase(BasicTestCase):
obj = np.array([1, 2])
type = obj.dtype.name
shape = None
chunkshape = (5,)
step = 1
reopen = 0 # This case does not reopen files
class BasicWrite5TestCase(BasicTestCase):
obj = [[1, 2], [3, 4]]
type = np.asarray(obj).dtype.name
shape = None
chunkshape = (5, 1)
step = 1
reopen = 0 # This case does not reopen files
class BasicWrite6TestCase(BasicTestCase):
obj = [1, 2]
type = np.asarray(obj).dtype.name
shape = None
chunkshape = (5,)
step = 1
reopen = 1 # This case does reopen files
class BasicWrite7TestCase(BasicTestCase):
obj = np.array([1, 2])
type = obj.dtype.name
shape = None
chunkshape = (5,)
step = 1
reopen = 1 # This case does reopen files
class BasicWrite8TestCase(BasicTestCase):
obj = [[1, 2], [3, 4]]
type = np.asarray(obj).dtype.name
shape = None
chunkshape = (5, 1)
step = 1
reopen = 1 # This case does reopen files
class EmptyCArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 2)
chunkshape = (5, 5)
start = 0
stop = 10
step = 1
class EmptyCArray2TestCase(BasicTestCase):
type = 'int32'
shape = (2, 2)
chunkshape = (5, 5)
start = 0
stop = 10
step = 1
reopen = 0 # This case does not reopen files
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class SlicesCArrayTestCase(BasicTestCase):
compress = 1
complib = "lzo"
type = 'int32'
shape = (2, 2)
chunkshape = (5, 5)
slices = (slice(1, 2, 1), slice(1, 3, 1))
class EllipsisCArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 2)
chunkshape = (5, 5)
# slices = (slice(1,2,1), Ellipsis)
slices = (Ellipsis, slice(1, 2, 1))
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class Slices2CArrayTestCase(BasicTestCase):
compress = 1
complib = "lzo"
type = 'int32'
shape = (2, 2, 4)
chunkshape = (5, 5, 5)
slices = (slice(1, 2, 1), slice(None, None, None), slice(1, 4, 2))
class Ellipsis2CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 2, 4)
chunkshape = (5, 5, 5)
slices = (slice(1, 2, 1), Ellipsis, slice(1, 4, 2))
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class Slices3CArrayTestCase(BasicTestCase):
compress = 1 # To show the chunks id DEBUG is on
complib = "lzo"
type = 'int32'
shape = (2, 3, 4, 2)
chunkshape = (5, 5, 5, 5)
slices = (slice(1, 2, 1), slice(
0, None, None), slice(1, 4, 2)) # Don't work
# slices = (slice(None, None, None), slice(0, None, None),
# slice(1,4,1)) # W
# slices = (slice(None, None, None), slice(None, None, None),
# slice(1,4,2)) # N
# slices = (slice(1,2,1), slice(None, None, None), slice(1,4,2)) # N
# Disable the failing test temporarily with a working test case
slices = (slice(1, 2, 1), slice(1, 4, None), slice(1, 4, 2)) # Y
# slices = (slice(1,2,1), slice(0, 4, None), slice(1,4,1)) # Y
slices = (slice(1, 2, 1), slice(0, 4, None), slice(1, 4, 2)) # N
# slices = (slice(1,2,1), slice(0, 4, None), slice(1,4,2),
# slice(0,100,1)) # N
class Slices4CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 2, 5, 6)
chunkshape = (5, 5, 5, 5, 5, 5)
slices = (slice(1, 2, 1), slice(0, None, None), slice(1, 4, 2),
slice(0, 4, 2), slice(3, 5, 2), slice(2, 7, 1))
class Ellipsis3CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 2)
chunkshape = (5, 5, 5, 5)
slices = (Ellipsis, slice(0, 4, None), slice(1, 4, 2))
slices = (slice(1, 2, 1), slice(0, 4, None), slice(1, 4, 2), Ellipsis)
class Ellipsis4CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 5)
chunkshape = (5, 5, 5, 5)
slices = (Ellipsis, slice(0, 4, None), slice(1, 4, 2))
slices = (slice(1, 2, 1), Ellipsis, slice(1, 4, 2))
class Ellipsis5CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 5)
chunkshape = (5, 5, 5, 5)
slices = (slice(1, 2, 1), slice(0, 4, None), Ellipsis)
class Ellipsis6CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 5)
chunkshape = (5, 5, 5, 5)
# The next slices gives problems with setting values (test03)
# This is a problem on the test design, not the Array.__setitem__
# code, though. See # see test_earray.py Ellipsis6EArrayTestCase
slices = (slice(1, 2, 1), slice(0, 4, None), 2, Ellipsis)
class Ellipsis7CArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 5)
chunkshape = (5, 5, 5, 5)
slices = (slice(1, 2, 1), slice(0, 4, None), slice(2, 3), Ellipsis)
class MD3WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 2, 3)
chunkshape = (4, 4, 4)
step = 2
class MD5WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 2, 3, 4, 5) # ok
# shape = (1, 1, 2, 1) # Minimum shape that shows problems with HDF5 1.6.1
# shape = (2, 3, 2, 4, 5) # Floating point exception (HDF5 1.6.1)
# shape = (2, 3, 3, 2, 5, 6) # Segmentation fault (HDF5 1.6.1)
chunkshape = (1, 1, 1, 1, 1)
start = 1
stop = 10
step = 10
class MD6WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 3, 2, 5, 6)
chunkshape = (1, 1, 1, 1, 5, 6)
start = 1
stop = 10
step = 3
class MD6WriteTestCase__(BasicTestCase):
type = 'int32'
shape = (2, 2)
chunkshape = (1, 1)
start = 1
stop = 3
step = 1
class MD7WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 3, 4, 5, 2, 3)
chunkshape = (10, 10, 10, 10, 10, 10, 10)
start = 1
stop = 10
step = 2
class MD10WriteTestCase(BasicTestCase):
type = 'int32'
shape = (1, 2, 3, 4, 5, 5, 4, 3, 2, 2)
chunkshape = (5, 5, 5, 5, 5, 5, 5, 5, 5, 5)
start = -1
stop = -1
step = 10
class ZlibComprTestCase(BasicTestCase):
compress = 1
complib = "zlib"
start = 3
# stop = 0 # means last row
stop = None # means last row from 0.8 on
step = 10
class ZlibShuffleTestCase(BasicTestCase):
shuffle = 1
compress = 1
complib = "zlib"
# case start < stop , i.e. no rows read
start = 3
stop = 1
step = 10
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscComprTestCase(BasicTestCase):
compress = 1 # sss
complib = "blosc"
chunkshape = (10, 10)
start = 3
stop = 10
step = 3
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscShuffleTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscBitShuffleTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
bitshuffle = 1
complib = "blosc"
chunkshape = (200, 100)
start = 2
stop = 11
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscFletcherTestCase(BasicTestCase):
# see gh-21
shape = (200, 300)
compress = 1
shuffle = 1
fletcher32 = 1
complib = "blosc"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscBloscLZTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:blosclz"
chunkshape = (200, 100)
start = 2
stop = 11
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
@common.unittest.skipIf(
'lz4' not in tb.blosc_compressor_list(), 'lz4 required')
class BloscLZ4TestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:lz4"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
@common.unittest.skipIf(
'lz4' not in tb.blosc_compressor_list(), 'lz4 required')
class BloscLZ4HCTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:lz4hc"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
@common.unittest.skipIf('snappy' not in tb.blosc_compressor_list(),
'snappy required')
class BloscSnappyTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:snappy"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
@common.unittest.skipIf(
'zlib' not in tb.blosc_compressor_list(), 'zlib required')
class BloscZlibTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:zlib"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
@common.unittest.skipIf(
'zstd' not in tb.blosc_compressor_list(), 'zstd required')
class BloscZstdTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "blosc:zstd"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2ComprTestCase(BasicTestCase):
compress = 1 # sss
complib = "blosc2"
chunkshape = (10, 10)
start = 3
stop = 10
step = 3
byteorder = foreign_byteorder()
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2FletcherTestCase(Blosc2ComprTestCase):
fletcher32 = 1
start = 0
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2CrossChunkTestCase(BasicTestCase):
shape = (10, 10)
compress = 1 # sss
complib = "blosc2"
chunkshape = (4, 4)
start = 3
stop = 6
step = 3
byteorder = foreign_byteorder()
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2CrossChunkOptTestCase(Blosc2CrossChunkTestCase):
step = 1 # optimized
byteorder = sys.byteorder
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2PastLastChunkTestCase(BasicTestCase):
shape = (10, 10)
compress = 1 # sss
complib = "blosc2"
chunkshape = (4, 4)
start = 8
stop = 100
step = 3
byteorder = foreign_byteorder()
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2PastLastChunkOptTestCase(Blosc2PastLastChunkTestCase):
step = 1 # optimized
byteorder = sys.byteorder
# Minimal test which can be figured out manually::
#
# z Data: 1 Chunk0: Chunk1: 1 Slice:
# / /|\ |\
# |\ 0 5 3 0 5 3 5
# x y |X X| |\ \| / \
# 4 2 7 4 2 7 4 7
# \|/ \| \ /
# 6 6 6
#
# Chunk0 & Slice: 4 Chunk1 & Slice: 5
# \ \
# 6 7
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2Ndim3MinChunkOptTestCase(BasicTestCase):
shape = (2, 2, 2)
compress = 1
complib = "blosc2"
chunkshape = (2, 2, 1)
byteorder = sys.byteorder
type = 'int8'
slices = (slice(1, 2), slice(0, 2), slice(0, 2))
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2Ndim3ChunkOptTestCase(BasicTestCase):
shape = (10, 10, 10)
compress = 1
complib = "blosc2"
chunkshape = (7, 7, 7)
byteorder = sys.byteorder
type = 'int32'
slices = (slice(1, 2), Ellipsis, slice(1, 4))
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2Ndim4ChunkOptTestCase(BasicTestCase):
shape = (13, 13, 13, 3)
compress = 1
complib = "blosc2"
chunkshape = (5, 5, 5, 3)
byteorder = sys.byteorder
type = 'int32'
slices = (slice(0, 8), slice(7, 13), slice(3, 12), slice(1, 3))
# The file used in the test below is created with this script,
# producing a chunked array that lacks chunk rank/shape in filter args.
# It is a reduced version of ``examples/direct-chunk-shape.py``,
# check there for more info and the assemblage of the data array.
# An h5py release is used which contains a version of hdf5-blosc2
# that does not include chunk rank/shape in filter arguments.
#
# ::
#
# import blosc2
# import h5py
# import hdf5plugin
# import numpy
#
# assert(hdf5plugin.version_info < (4, 2, 1))
#
# fparams = hdf5plugin.Blosc2(cname='zstd', clevel=1,
# filters=hdf5plugin.Blosc2.SHUFFLE)
# cparams = {
# "codec": blosc2.Codec.ZSTD,
# "clevel": 1,
# "filters": [blosc2.Filter.SHUFFLE],
# }
#
# achunk = numpy.arange(4 * 4, dtype='int8').reshape((4, 4))
# adata = numpy.zeros((6, 6), dtype=achunk.dtype)
# adata[0:4, 0:4] = achunk[:, :]
# adata[0:4, 4:6] = achunk[:, 0:2]
# adata[4:6, 0:4] = achunk[0:2, :]
# adata[4:6, 4:6] = achunk[0:2, 0:2]
#
# h5f = h5py.File("b2nd-no-chunkshape.h5", "w")
# dataset = h5f.create_dataset(
# "data", adata.shape, dtype=adata.dtype, chunks=achunk.shape,
# **fparams)
# b2chunk = blosc2.asarray(achunk,
# chunks=achunk.shape, blocks=achunk.shape,
# cparams=cparams)
# b2frame = b2chunk._schunk.to_cframe()
# dataset.id.write_direct_chunk((0, 0), b2frame)
# dataset.id.write_direct_chunk((0, 4), b2frame)
# dataset.id.write_direct_chunk((4, 0), b2frame)
# dataset.id.write_direct_chunk((4, 4), b2frame)
# h5f.close()
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2NDNoChunkshape(common.TestFileMixin,
common.PyTablesTestCase):
h5fname = common.test_filename('b2nd-no-chunkshape.h5')
adata = np.array(
[[ 0, 1, 2, 3, 0, 1],
[ 4, 5, 6, 7, 4, 5],
[ 8, 9, 10, 11, 8, 9],
[12, 13, 14, 15, 12, 13],
[ 0, 1, 2, 3, 0, 1],
[ 4, 5, 6, 7, 4, 5]],
dtype='int8')
def test_data_opt(self):
array = self.h5file.get_node('/data')
self.assertTrue(common.areArraysEqual(array[:], self.adata[:]))
def test_data_filter(self):
array = self.h5file.get_node('/data')
self.assertTrue(common.areArraysEqual(array[::2], self.adata[::2]))
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class LZOComprTestCase(BasicTestCase):
compress = 1 # sss
complib = "lzo"
chunkshape = (10, 10)
start = 3
stop = 10
step = 3
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class LZOShuffleTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "lzo"
chunkshape = (100, 100)
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class Bzip2ComprTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
complib = "bzip2"
chunkshape = (100, 100)
start = 3
stop = 10
step = 8
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class Bzip2ShuffleTestCase(BasicTestCase):
shape = (20, 30)
compress = 1
shuffle = 1
complib = "bzip2"
chunkshape = (100, 100)
start = 3
stop = 10
step = 6
class Fletcher32TestCase(BasicTestCase):
shape = (60, 50)
compress = 0
fletcher32 = 1
chunkshape = (50, 50)
start = 4
stop = 20
step = 7
class AllFiltersTestCase(BasicTestCase):
compress = 1
shuffle = 1
fletcher32 = 1
complib = "zlib"
chunkshape = (20, 20) # sss
start = 2
stop = 99
step = 6
class FloatTypeTestCase(BasicTestCase):
type = 'float64'
shape = (2, 2)
chunkshape = (5, 5)
start = 3
stop = 10
step = 20
class ComplexTypeTestCase(BasicTestCase):
type = 'complex128'
shape = (2, 2)
chunkshape = (5, 5)
start = 3
stop = 10
step = 20
class StringTestCase(BasicTestCase):
type = "string"
length = 20
shape = (2, 2)
# shape = (2,2,20)
chunkshape = (5, 5)
start = 3
stop = 10
step = 20
slices = (slice(0, 1), slice(1, 2))
class String2TestCase(BasicTestCase):
type = "string"
length = 20
shape = (2, 20)
chunkshape = (5, 5)
start = 1
stop = 10
step = 2
class StringComprTestCase(BasicTestCase):
type = "string"
length = 20
shape = (20, 2, 10)
# shape = (20,0,10,20)
compr = 1
# shuffle = 1 # this shouldn't do nothing on chars
chunkshape = (50, 50, 2)
start = -1
stop = 100
step = 20
class Int8TestCase(BasicTestCase):
type = "int8"
shape = (2, 2)
compress = 1
shuffle = 1
chunkshape = (50, 50)
start = -1
stop = 100
step = 20
class Int16TestCase(BasicTestCase):
type = "int16"
shape = (2, 2)
compress = 1
shuffle = 1
chunkshape = (50, 50)
start = 1
stop = 100
step = 1
class Int32TestCase(BasicTestCase):
type = "int32"
shape = (2, 2)
compress = 1
shuffle = 1
chunkshape = (50, 50)
start = -1
stop = 100
step = 20
@common.unittest.skipUnless(hasattr(tb, 'Float16Atom'),
'Float16Atom not available')
class Float16TestCase(BasicTestCase):
type = "float16"
shape = (200,)
compress = 1
shuffle = 1
chunkshape = (20,)
start = -1
stop = 100
step = 20
class Float32TestCase(BasicTestCase):
type = "float32"
shape = (200,)
compress = 1
shuffle = 1
chunkshape = (20,)
start = -1
stop = 100
step = 20
class Float64TestCase(BasicTestCase):
type = "float64"
shape = (200,)
compress = 1
shuffle = 1
chunkshape = (20,)
start = -1
stop = 100
step = 20
@common.unittest.skipUnless(hasattr(tb, 'Float96Atom'),
'Float96Atom not available')
class Float96TestCase(BasicTestCase):
type = "float96"
shape = (200,)
compress = 1
shuffle = 1
chunkshape = (20,)
start = -1
stop = 100
step = 20
@common.unittest.skipUnless(hasattr(tb, 'Float128Atom'),
'Float128Atom not available')
class Float128TestCase(BasicTestCase):
type = "float128"
shape = (200,)
compress = 1
shuffle = 1
chunkshape = (20,)
start = -1
stop = 100
step = 20
class Complex64TestCase(BasicTestCase):
type = "complex64"
shape = (4,)
compress = 1
shuffle = 1
chunkshape = (2,)
start = -1
stop = 100
step = 20
class Complex128TestCase(BasicTestCase):
type = "complex128"
shape = (20,)
compress = 1
shuffle = 1
chunkshape = (2,)
start = -1
stop = 100
step = 20
@common.unittest.skipUnless(hasattr(tb, 'Complex192Atom'),
'Complex192Atom not available')
class Complex192TestCase(BasicTestCase):
type = "complex192"
shape = (20,)
compress = 1
shuffle = 1
chunkshape = (2,)
start = -1
stop = 100
step = 20
@common.unittest.skipUnless(hasattr(tb, 'Complex256Atom'),
'Complex256Atom not available')
class Complex256TestCase(BasicTestCase):
type = "complex256"
shape = (20,)
compress = 1
shuffle = 1
chunkshape = (2,)
start = -1
stop = 100
step = 20
class ComprTestCase(BasicTestCase):
type = "float64"
compress = 1
shuffle = 1
shape = (200,)
compr = 1
chunkshape = (21,)
start = 51
stop = 100
step = 7
# this is a subset of the tests in test_array.py, mostly to verify that errors
# are handled in the same way
class ReadOutArgumentTests(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.size = 1000
self.filters = tb.Filters(complevel=1, complib='blosc')
def create_array(self):
array = np.arange(self.size, dtype='i8')
disk_array = self.h5file.create_carray('/', 'array',
atom=tb.Int64Atom(),
shape=(self.size,),
filters=self.filters)
disk_array[:] = array
return array, disk_array
def test_read_entire_array(self):
array, disk_array = self.create_array()
out_buffer = np.empty((self.size, ), 'i8')
disk_array.read(out=out_buffer)
np.testing.assert_equal(out_buffer, array)
def test_read_non_contiguous_buffer(self):
array, disk_array = self.create_array()
out_buffer = np.empty((self.size, ), 'i8')
out_buffer_slice = out_buffer[0:self.size:2]
with self.assertRaisesRegex(ValueError,
'output array not C contiguous'):
disk_array.read(0, self.size, 2, out_buffer_slice)
def test_buffer_too_small(self):
array, disk_array = self.create_array()
out_buffer = np.empty((self.size // 2, ), 'i8')
self.assertRaises(ValueError, disk_array.read, 0, self.size, 1,
out_buffer)
try:
disk_array.read(0, self.size, 1, out_buffer)
except ValueError as exc:
self.assertIn('output array size invalid, got', str(exc))
class SizeOnDiskInMemoryPropertyTestCase(common.TempFileMixin,
common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.array_size = (10_000, 10)
# set chunkshape so it divides evenly into array_size, to avoid
# partially filled chunks
self.chunkshape = (1000, 10)
# approximate size (in bytes) of non-data portion of hdf5 file
self.hdf_overhead = 6000
def create_array(self, complevel):
filters = tb.Filters(complevel=complevel, complib='blosc')
self.array = self.h5file.create_carray('/', 'somearray',
atom=tb.Int16Atom(),
shape=self.array_size,
filters=filters,
chunkshape=self.chunkshape)
def test_no_data(self):
complevel = 0
self.create_array(complevel)
self.assertEqual(self.array.size_on_disk, 0)
self.assertEqual(self.array.size_in_memory, 10_000 * 10 * 2)
def test_data_no_compression(self):
complevel = 0
self.create_array(complevel)
self.array[:] = 1
self.assertEqual(self.array.size_on_disk, 10_000 * 10 * 2)
self.assertEqual(self.array.size_in_memory, 10_000 * 10 * 2)
def test_highly_compressible_data(self):
complevel = 1
self.create_array(complevel)
self.array[:] = 1
self.h5file.flush()
file_size = Path(self.h5fname).stat().st_size
self.assertTrue(
abs(self.array.size_on_disk - file_size) <= self.hdf_overhead)
self.assertTrue(self.array.size_on_disk < self.array.size_in_memory)
self.assertEqual(self.array.size_in_memory, 10_000 * 10 * 2)
# XXX
def test_random_data(self):
complevel = 1
self.create_array(complevel)
self.array[:] = np.random.randint(0, 1e6, self.array_size)
self.h5file.flush()
file_size = Path(self.h5fname).stat().st_size
self.assertTrue(
abs(self.array.size_on_disk - file_size) <= self.hdf_overhead)
# XXX: check. The test fails if blosc is not available
if tb.which_lib_version('blosc') is not None:
self.assertAlmostEqual(self.array.size_on_disk, 10_000 * 10 * 2)
else:
self.assertTrue(
abs(self.array.size_on_disk - 10_000 * 10 * 2) < 200)
class OffsetStrideTestCase(common.TempFileMixin, common.PyTablesTestCase):
compress = 0
complib = "zlib" # Default compression library
def setUp(self):
super().setUp()
# Create an instance of an HDF5 Table
self.rootgroup = self.h5file.root
def test01a_String(self):
"""Checking carray with offset NumPy strings appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01a_String..." % self.__class__.__name__)
shape = (3, 2, 2)
# Create a string atom
carray = self.h5file.create_carray(root, 'strings',
atom=tb.StringAtom(itemsize=3),
shape=shape,
title="Array of strings",
chunkshape=(1, 2, 2))
a = np.array([[["a", "b"], ["123", "45"], ["45", "123"]]], dtype="S3")
carray[0] = a[0, 1:]
a = np.array([[["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"]]])
carray[1] = a[0, 2:]
# Read all the data:
data = carray.read()
if common.verbose:
print("Object read:", data)
print("Nrows in", carray._v_pathname, ":", carray.nrows)
print("Second row in carray ==>", data[1].tolist())
self.assertEqual(carray.nrows, 3)
self.assertEqual(data[0].tolist(), [[b"123", b"45"], [b"45", b"123"]])
self.assertEqual(data[1].tolist(), [[b"s", b"abc"], [b"abc", b"f"]])
self.assertEqual(len(data[0]), 2)
self.assertEqual(len(data[1]), 2)
def test01b_String(self):
"""Checking carray with strided NumPy strings appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01b_String..." % self.__class__.__name__)
shape = (3, 2, 2)
# Create a string atom
carray = self.h5file.create_carray(root, 'strings',
atom=tb.StringAtom(itemsize=3),
shape=shape,
title="Array of strings",
chunkshape=(1, 2, 2))
a = np.array([[["a", "b"], ["123", "45"], ["45", "123"]]], dtype="S3")
carray[0] = a[0, ::2]
a = np.array([[["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"]]])
carray[1] = a[0, ::2]
# Read all the rows:
data = carray.read()
if common.verbose:
print("Object read:", data)
print("Nrows in", carray._v_pathname, ":", carray.nrows)
print("Second row in carray ==>", data[1].tolist())
self.assertEqual(carray.nrows, 3)
self.assertEqual(data[0].tolist(), [[b"a", b"b"], [b"45", b"123"]])
self.assertEqual(data[1].tolist(), [[b"s", b"a"], [b"s", b"abc"]])
self.assertEqual(len(data[0]), 2)
self.assertEqual(len(data[1]), 2)
def test02a_int(self):
"""Checking carray with offset NumPy ints appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02a_int..." % self.__class__.__name__)
shape = (3, 3)
# Create a string atom
carray = self.h5file.create_carray(root, 'CAtom',
atom=tb.Int32Atom(), shape=shape,
title="array of ints",
chunkshape=(1, 3))
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (0, 0, 0)],
dtype='int32')
carray[0:2] = a[2:] # Introduce an offset
a = np.array([(1, 1, 1), (-1, 0, 0)], dtype='int32')
carray[2:3] = a[1:] # Introduce an offset
# Read all the rows:
data = carray.read()
if common.verbose:
print("Object read:", data)
print("Nrows in", carray._v_pathname, ":", carray.nrows)
print("Third row in carray ==>", data[2])
self.assertEqual(carray.nrows, 3)
self.assertTrue(common.allequal(
data[0], np.array([1, 1, 1], dtype='int32')))
self.assertTrue(common.allequal(
data[1], np.array([0, 0, 0], dtype='int32')))
self.assertTrue(common.allequal(
data[2], np.array([-1, 0, 0], dtype='int32')))
def test02b_int(self):
"""Checking carray with strided NumPy ints appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02b_int..." % self.__class__.__name__)
shape = (3, 3)
# Create a string atom
carray = self.h5file.create_carray(root, 'CAtom',
atom=tb.Int32Atom(), shape=shape,
title="array of ints",
chunkshape=(1, 3))
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='int32')
carray[0:2] = a[::3] # Create an offset
a = np.array([(1, 1, 1), (-1, 0, 0)], dtype='int32')
carray[2:3] = a[::2] # Create an offset
# Read all the rows:
data = carray.read()
if common.verbose:
print("Object read:", data)
print("Nrows in", carray._v_pathname, ":", carray.nrows)
print("Third row in carray ==>", data[2])
self.assertEqual(carray.nrows, 3)
self.assertTrue(common.allequal(
data[0], np.array([0, 0, 0], dtype='int32')))
self.assertTrue(common.allequal(
data[1], np.array([3, 3, 3], dtype='int32')))
self.assertTrue(common.allequal(
data[2], np.array([1, 1, 1], dtype='int32')))
class CopyTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test01a_copy(self):
"""Checking CArray.copy() method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01a_copy..." % self.__class__.__name__)
# Create an CArray
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[...] = np.array([[456, 2], [3, 457]], dtype='int16')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
# print("dirs-->", dir(array1), dir(array2))
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
self.assertTrue(common.allequal(array1.read(), array2.read()))
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor)
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test01b_copy(self):
"""Checking CArray.copy() method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01b_copy..." % self.__class__.__name__)
# Create an CArray
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(5, 5))
array1[...] = np.array([[456, 2], [3, 457]], dtype='int16')
if self.close:
if common.verbose:
print("(closing h5fname version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing h5fname version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
# print("dirs-->", dir(array1), dir(array2))
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
self.assertTrue(common.allequal(array1.read(), array2.read()))
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor)
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# By default, the chunkshape should be the same
self.assertEqual(array1.chunkshape, array2.chunkshape)
def test01c_copy(self):
"""Checking CArray.copy() method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01c_copy..." % self.__class__.__name__)
# Create an CArray
shape = (5, 5)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[:2, :2] = np.array([[456, 2], [3, 457]], dtype='int16')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
# print("dirs-->", dir(array1), dir(array2))
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
self.assertTrue(common.allequal(array1.read(), array2.read()))
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor)
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test02_copy(self):
"""Checking CArray.copy() method (where specified)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_copy..." % self.__class__.__name__)
# Create an CArray
shape = (5, 5)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[:2, :2] = np.array([[456, 2], [3, 457]], dtype='int16')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy to another location
group1 = self.h5file.create_group("/", "group1")
array2 = array1.copy(group1, 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.group1.array2
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
# print("dirs-->", dir(array1), dir(array2))
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
self.assertTrue(common.allequal(array1.read(), array2.read()))
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor)
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test03a_copy(self):
"""Checking CArray.copy() method (python flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03c_copy..." % self.__class__.__name__)
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1.flavor = "python"
array1[...] = [[456, 2], [3, 457]]
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all elements are equal
self.assertEqual(array1.read(), array2.read())
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor) # Very important here!
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test03b_copy(self):
"""Checking CArray.copy() method (string python flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03d_copy..." % self.__class__.__name__)
shape = (2, 2)
atom = tb.StringAtom(itemsize=4)
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1.flavor = "python"
array1[...] = [["456", "2"], ["3", "457"]]
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("type value-->", type(array2[:][0][0]))
print("value-->", array2[:])
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all elements are equal
self.assertEqual(array1.read(), array2.read())
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor) # Very important here!
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test03c_copy(self):
"""Checking CArray.copy() method (chararray flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03e_copy..." % self.__class__.__name__)
shape = (2, 2)
atom = tb.StringAtom(itemsize=4)
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[...] = np.array([["456", "2"], ["3", "457"]], dtype="S4")
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy to another location
array2 = array1.copy('/', 'array2')
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all elements are equal
self.assertTrue(common.allequal(array1.read(), array2.read()))
# Assert other properties in array
self.assertEqual(array1.nrows, array2.nrows)
self.assertEqual(array1.shape, array2.shape)
self.assertEqual(array1.extdim, array2.extdim)
self.assertEqual(array1.flavor, array2.flavor) # Very important here!
self.assertEqual(array1.atom.dtype, array2.atom.dtype)
self.assertEqual(array1.atom.type, array2.atom.type)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# self.assertEqual(array1.chunkshape, array2.chunkshape)
def test04_copy(self):
"""Checking CArray.copy() method (checking title copying)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_copy..." % self.__class__.__name__)
# Create an CArray
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[...] = np.array([[456, 2], [3, 457]], dtype='int16')
# Append some user attrs
array1.attrs.attr1 = "attr1"
array1.attrs.attr2 = 2
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another Array
array2 = array1.copy('/', 'array2', title="title array2")
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
# Assert user attributes
if common.verbose:
print("title of destination array-->", array2.title)
self.assertEqual(array2.title, "title array2")
def test05_copy(self):
"""Checking CArray.copy() method (user attributes copied)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_copy..." % self.__class__.__name__)
# Create an CArray
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[...] = np.array([[456, 2], [3, 457]], dtype='int16')
# Append some user attrs
array1.attrs.attr1 = "attr1"
array1.attrs.attr2 = 2
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another Array
array2 = array1.copy('/', 'array2', copyuserattrs=1)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Assert user attributes
self.assertEqual(array2.attrs.attr1, "attr1")
self.assertEqual(array2.attrs.attr2, 2)
def test05b_copy(self):
"""Checking CArray.copy() method (user attributes not copied)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05b_copy..." % self.__class__.__name__)
# Create an Array
shape = (2, 2)
atom = tb.Int16Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
array1[...] = np.array([[456, 2], [3, 457]], dtype='int16')
# Append some user attrs
array1.attrs.attr1 = "attr1"
array1.attrs.attr2 = 2
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen(mode="a")
array1 = self.h5file.root.array1
# Copy it to another Array
array2 = array1.copy('/', 'array2', copyuserattrs=0)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Assert user attributes
self.assertEqual(hasattr(array2.attrs, "attr1"), 0)
self.assertEqual(hasattr(array2.attrs, "attr2"), 0)
class CloseCopyTestCase(CopyTestCase):
close = 1
class OpenCopyTestCase(CopyTestCase):
close = 0
class CopyIndexTestCase(common.TempFileMixin, common.PyTablesTestCase):
nrowsinbuf = 2
def test01_index(self):
"""Checking CArray.copy() method with indexes."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_index..." % self.__class__.__name__)
# Create an CArray
shape = (100, 2)
atom = tb.Int32Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
r = np.arange(200, dtype='int32')
r.shape = shape
array1[...] = r
# Select a different buffer size:
array1.nrowsinbuf = self.nrowsinbuf
# Copy to another array
array2 = array1.copy("/", 'array2',
start=self.start,
stop=self.stop,
step=self.step)
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
r2 = r[self.start:self.stop:self.step]
self.assertTrue(common.allequal(r2, array2.read()))
# Assert the number of rows in array
if common.verbose:
print("nrows in array2-->", array2.nrows)
print("and it should be-->", r2.shape[0])
# The next line is commented out because a copy should not
# keep the same chunkshape anymore.
# F. Alted 2006-11-27
# assert array1.chunkshape == array2.chunkshape
self.assertEqual(r2.shape[0], array2.nrows)
def _test02_indexclosef(self):
"""Checking CArray.copy() method with indexes (close file version)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_indexclosef..." % self.__class__.__name__)
# Create an CArray
shape = (100, 2)
atom = tb.Int32Atom()
array1 = self.h5file.create_carray(
self.h5file.root, 'array1', atom=atom, shape=shape,
title="title array1", chunkshape=(2, 2))
r = np.arange(200, dtype='int32')
r.shape = shape
array1[...] = r
# Select a different buffer size:
array1.nrowsinbuf = self.nrowsinbuf
# Copy to another array
array2 = array1.copy("/", 'array2',
start=self.start,
stop=self.stop,
step=self.step)
# Close and reopen the file
self._reopen()
array1 = self.h5file.root.array1
array2 = self.h5file.root.array2
if common.verbose:
print("array1-->", array1.read())
print("array2-->", array2.read())
print("attrs array1-->", repr(array1.attrs))
print("attrs array2-->", repr(array2.attrs))
# Check that all the elements are equal
r2 = r[self.start:self.stop:self.step]
self.assertEqual(array1.chunkshape, array2.chunkshape)
self.assertTrue(common.allequal(r2, array2.read()))
# Assert the number of rows in array
if common.verbose:
print("nrows in array2-->", array2.nrows)
print("and it should be-->", r2.shape[0])
self.assertEqual(r2.shape[0], array2.nrows)
class CopyIndex1TestCase(CopyIndexTestCase):
nrowsinbuf = 1
start = 0
stop = 7
step = 1
class CopyIndex2TestCase(CopyIndexTestCase):
nrowsinbuf = 2
start = 0
stop = -1
step = 1
class CopyIndex3TestCase(CopyIndexTestCase):
nrowsinbuf = 3
start = 1
stop = 7
step = 1
class CopyIndex4TestCase(CopyIndexTestCase):
nrowsinbuf = 4
start = 0
stop = 6
step = 1
class CopyIndex5TestCase(CopyIndexTestCase):
nrowsinbuf = 2
start = 3
stop = 7
step = 1
class CopyIndex6TestCase(CopyIndexTestCase):
nrowsinbuf = 2
start = 3
stop = 6
step = 2
class CopyIndex7TestCase(CopyIndexTestCase):
start = 0
stop = 7
step = 10
class CopyIndex8TestCase(CopyIndexTestCase):
start = 6
stop = -1 # Negative values means starting from the end
step = 1
class CopyIndex9TestCase(CopyIndexTestCase):
start = 3
stop = 4
step = 1
class CopyIndex10TestCase(CopyIndexTestCase):
nrowsinbuf = 1
start = 3
stop = 4
step = 2
class CopyIndex11TestCase(CopyIndexTestCase):
start = -3
stop = -1
step = 2
class CopyIndex12TestCase(CopyIndexTestCase):
start = -1 # Should point to the last element
stop = None # None should mean the last element (including it)
step = 1
# The next test should be run only in **heavy** mode
class Rows64bitsTestCase(common.TempFileMixin, common.PyTablesTestCase):
narows = 1000 * 1000 # each array will have 1 million entries
# narows = 1000 # for testing only
nanumber = 1000 * 3 # That should account for more than 2**31-1
def setUp(self):
super().setUp()
# Create an CArray
shape = (self.narows * self.nanumber,)
array = self.h5file.create_carray(
self.h5file.root, 'array',
atom=tb.Int8Atom(), shape=shape,
filters=tb.Filters(complib='lzo', complevel=1))
# Fill the array
na = np.arange(self.narows, dtype='int8')
# for i in xrange(self.nanumber):
# s = slice(i * self.narows, (i + 1)*self.narows)
# array[s] = na
s = slice(0, self.narows)
array[s] = na
s = slice((self.nanumber-1)*self.narows, self.nanumber * self.narows)
array[s] = na
def test01_basiccheck(self):
"""Some basic checks for carrays exceeding 2**31 rows"""
array = self.h5file.root.array
if self.close:
if common.verbose:
# Check how many entries there are in the array
print("Before closing")
print("Entries:", array.nrows, type(array.nrows))
print("Entries:", array.nrows / (1000 * 1000), "Millions")
print("Shape:", array.shape)
# Re-open the file
self._reopen()
array = self.h5file.root.array
if common.verbose:
print("After re-open")
# Check how many entries there are in the array
if common.verbose:
print("Entries:", array.nrows, type(array.nrows))
print("Entries:", array.nrows / (1000 * 1000), "Millions")
print("Shape:", array.shape)
print("Last 10 elements-->", array[-10:])
stop = self.narows % 256
if stop > 127:
stop -= 256
start = stop - 10
# print("start, stop-->", start, stop)
print("Should look like:", np.arange(start, stop, dtype='int8'))
nrows = self.narows * self.nanumber
# check nrows
self.assertEqual(array.nrows, nrows)
# Check shape
self.assertEqual(array.shape, (nrows,))
# check the 10 first elements
self.assertTrue(common.allequal(
array[:10], np.arange(10, dtype='int8')))
# check the 10 last elements
stop = self.narows % 256
if stop > 127:
stop -= 256
start = stop - 10
self.assertTrue(common.allequal(
array[-10:], np.arange(start, stop, dtype='int8')))
class Rows64bitsTestCase1(Rows64bitsTestCase):
close = 0
class Rows64bitsTestCase2(Rows64bitsTestCase):
close = 1
class BigArrayTestCase(common.TempFileMixin, common.PyTablesTestCase):
shape = (3_000_000_000,) # more than 2**31-1
def setUp(self):
super().setUp()
# This should be fast since disk space isn't actually allocated,
# so this case is OK for non-heavy test runs.
self.h5file.create_carray('/', 'array',
atom=tb.Int8Atom(), shape=self.shape)
def test00_shape(self):
"""Check that the shape doesn't overflow."""
# See ticket #147.
self.assertEqual(self.h5file.root.array.shape, self.shape)
try:
self.assertEqual(len(self.h5file.root.array), self.shape[0])
except OverflowError:
# This can't be avoided in 32-bit platforms.
self.assertTrue(self.shape[0] > np.iinfo(int).max,
"Array length overflowed but ``int`` "
"is wide enough.")
def test01_shape_reopen(self):
"""Check that the shape doesn't overflow after reopening."""
self._reopen('r')
self.test00_shape()
# Test for default values when creating arrays.
class DfltAtomTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test00_dflt(self):
"""Check that Atom.dflt is honored (string version)."""
# Create a CArray with default values
self.h5file.create_carray(
'/', 'bar', atom=tb.StringAtom(itemsize=5, dflt=b"abdef"),
shape=(10, 10))
if self.reopen:
self._reopen()
# Check the values
values = self.h5file.root.bar[:]
if common.verbose:
print("Read values:", values)
self.assertTrue(common.allequal(
values, np.array(["abdef"] * 100, "S5").reshape(10, 10)))
def test01_dflt(self):
"""Check that Atom.dflt is honored (int version)."""
# Create a CArray with default values
self.h5file.create_carray('/', 'bar',
atom=tb.IntAtom(dflt=1), shape=(10, 10))
if self.reopen:
self._reopen()
# Check the values
values = self.h5file.root.bar[:]
if common.verbose:
print("Read values:", values)
self.assertTrue(common.allequal(values, np.ones((10, 10), "i4")))
def test02_dflt(self):
"""Check that Atom.dflt is honored (float version)."""
# Create a CArray with default values
self.h5file.create_carray(
'/', 'bar', atom=tb.FloatAtom(dflt=1.134), shape=(10, 10))
if self.reopen:
self._reopen()
# Check the values
values = self.h5file.root.bar[:]
if common.verbose:
print("Read values:", values)
self.assertTrue(common.allequal(values, np.ones((10, 10), "f8")*1.134))
class DfltAtomNoReopen(DfltAtomTestCase):
reopen = False
class DfltAtomReopen(DfltAtomTestCase):
reopen = True
# Test for representation of defaults in atoms. Ticket #212.
class AtomDefaultReprTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test00a_zeros(self):
"""Testing default values. Zeros (scalar)."""
N = ()
atom = tb.StringAtom(itemsize=3, shape=N, dflt=b"")
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", repr(ca[0]))
print("Defaults-->", repr(ca.atom.dflt))
self.assertTrue(common.allequal(ca[0], np.zeros(N, 'S3')))
self.assertTrue(common.allequal(ca.atom.dflt, np.zeros(N, 'S3')))
def test00b_zeros(self):
"""Testing default values. Zeros (array)."""
N = 2
atom = tb.StringAtom(itemsize=3, shape=N, dflt=b"")
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", ca[0])
print("Defaults-->", ca.atom.dflt)
self.assertTrue(common.allequal(ca[0], np.zeros(N, 'S3')))
self.assertTrue(common.allequal(ca.atom.dflt, np.zeros(N, 'S3')))
def test01a_values(self):
"""Testing default values. Ones."""
N = 2
atom = tb.Int32Atom(shape=N, dflt=1)
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", ca[0])
print("Defaults-->", ca.atom.dflt)
self.assertTrue(common.allequal(ca[0], np.ones(N, 'i4')))
self.assertTrue(common.allequal(ca.atom.dflt, np.ones(N, 'i4')))
def test01b_values(self):
"""Testing default values. Generic value."""
N = 2
generic = 112.32
atom = tb.Float32Atom(shape=N, dflt=generic)
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", ca[0])
print("Defaults-->", ca.atom.dflt)
self.assertTrue(common.allequal(ca[0], np.ones(N, 'f4')*generic))
self.assertTrue(common.allequal(
ca.atom.dflt, np.ones(N, 'f4')*generic))
def test02a_None(self):
"""Testing default values. None (scalar)."""
N = ()
atom = tb.Int32Atom(shape=N, dflt=None)
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", repr(ca[0]))
print("Defaults-->", repr(ca.atom.dflt))
self.assertTrue(common.allequal(ca.atom.dflt, np.zeros(N, 'i4')))
def test02b_None(self):
"""Testing default values. None (array)."""
N = 2
atom = tb.Int32Atom(shape=N, dflt=None)
ca = self.h5file.create_carray('/', 'test', atom=atom, shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", ca[0])
print("Defaults-->", ca.atom.dflt)
self.assertTrue(common.allequal(ca.atom.dflt, np.zeros(N, 'i4')))
class AtomDefaultReprNoReopen(AtomDefaultReprTestCase):
reopen = False
class AtomDefaultReprReopen(AtomDefaultReprTestCase):
reopen = True
class TruncateTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test(self):
"""Test for inability to truncate Array objects."""
array1 = self.h5file.create_carray('/', 'array1', tb.IntAtom(), [2, 2])
self.assertRaises(TypeError, array1.truncate, 0)
# Test for dealing with multidimensional atoms
class MDAtomTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test01a_assign(self):
"""Assign a row to a (unidimensional) CArray with a MD atom."""
# Create an CArray
ca = self.h5file.create_carray('/', 'test',
atom=tb.Int32Atom((2, 2)), shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign one row
ca[0] = [[1, 3], [4, 5]]
self.assertEqual(ca.nrows, 1)
if common.verbose:
print("First row-->", ca[0])
self.assertTrue(common.allequal(
ca[0], np.array([[1, 3], [4, 5]], 'i4')))
def test01b_assign(self):
"""Assign several rows to a (unidimensional) CArray with a MD atom."""
# Create an CArray
ca = self.h5file.create_carray('/', 'test',
atom=tb.Int32Atom((2, 2)), shape=(3,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign three rows
ca[:] = [[[1]], [[2]], [[3]]] # Simple broadcast
self.assertEqual(ca.nrows, 3)
if common.verbose:
print("Third row-->", ca[2])
self.assertTrue(common.allequal(
ca[2], np.array([[3, 3], [3, 3]], 'i4')))
def test02a_assign(self):
"""Assign a row to a (multidimensional) CArray with a MD atom."""
# Create an CArray
ca = self.h5file.create_carray('/', 'test',
atom=tb.Int32Atom((2,)), shape=(1, 3))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign one row
ca[:] = [[[1, 3], [4, 5], [7, 9]]]
self.assertEqual(ca.nrows, 1)
if common.verbose:
print("First row-->", ca[0])
self.assertTrue(common.allequal(ca[0], np.array(
[[1, 3], [4, 5], [7, 9]], 'i4')))
def test02b_assign(self):
"""Assign several rows to a (multidimensional) CArray with
a MD atom."""
# Create an CArray
ca = self.h5file.create_carray('/', 'test',
atom=tb.Int32Atom((2,)), shape=(3, 3))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign three rows
ca[:] = [[[1, -3], [4, -5], [-7, 9]],
[[-1, 3], [-4, 5], [7, -8]],
[[-2, 3], [-5, 5], [7, -9]]]
self.assertEqual(ca.nrows, 3)
if common.verbose:
print("Third row-->", ca[2])
self.assertTrue(common.allequal(
ca[2], np.array([[-2, 3], [-5, 5], [7, -9]], 'i4')))
def test03a_MDMDMD(self):
"""Complex assign of a MD array in a MD CArray with a MD atom."""
# Create an CArray
ca = self.h5file.create_carray(
'/', 'test', atom=tb.Int32Atom((2, 4)), shape=(3, 2, 3))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign values
# The shape of the atom should be added at the end of the arrays
a = np.arange(2 * 3*2*4, dtype='i4').reshape((2, 3, 2, 4))
ca[:] = [a * 1, a*2, a*3]
self.assertEqual(ca.nrows, 3)
if common.verbose:
print("Third row-->", ca[2])
self.assertTrue(common.allequal(ca[2], a * 3))
def test03b_MDMDMD(self):
"""Complex assign of a MD array in a MD CArray with a MD atom (II)."""
# Create an CArray
ca = self.h5file.create_carray(
'/', 'test', atom=tb.Int32Atom((2, 4)), shape=(2, 3, 3))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign values
# The shape of the atom should be added at the end of the arrays
a = np.arange(2 * 3*3*2*4, dtype='i4').reshape((2, 3, 3, 2, 4))
ca[:] = a
self.assertEqual(ca.nrows, 2)
if common.verbose:
print("Third row-->", ca[:, 2, ...])
self.assertTrue(common.allequal(ca[:, 2, ...], a[:, 2, ...]))
def test03c_MDMDMD(self):
"""Complex assign of a MD array in a MD CArray with a MD atom (III)."""
# Create an CArray
ca = self.h5file.create_carray(
'/', 'test', atom=tb.Int32Atom((2, 4)), shape=(3, 1, 2))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Assign values
# The shape of the atom should be added at the end of the arrays
a = np.arange(3 * 1*2*2*4, dtype='i4').reshape((3, 1, 2, 2, 4))
ca[:] = a
self.assertEqual(ca.nrows, 3)
if common.verbose:
print("Second row-->", ca[:, :, 1, ...])
self.assertTrue(common.allequal(ca[:, :, 1, ...], a[:, :, 1, ...]))
class MDAtomNoReopen(MDAtomTestCase):
reopen = False
class MDAtomReopen(MDAtomTestCase):
reopen = True
# Test for building very large MD atoms without defaults. Ticket #211.
class MDLargeAtomTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test01_create(self):
"""Create a CArray with a very large MD atom."""
N = 2**16 # 4x larger than maximum object header size (64 KB)
ca = self.h5file.create_carray('/', 'test',
atom=tb.Int32Atom(shape=N), shape=(1,))
if self.reopen:
self._reopen('a')
ca = self.h5file.root.test
# Check the value
if common.verbose:
print("First row-->", ca[0])
self.assertTrue(common.allequal(ca[0], np.zeros(N, 'i4')))
class MDLargeAtomNoReopen(MDLargeAtomTestCase):
reopen = False
class MDLargeAtomReopen(MDLargeAtomTestCase):
reopen = True
class AccessClosedTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.array = self.h5file.create_carray(self.h5file.root, 'array',
atom=tb.Int32Atom(),
shape=(10, 10))
self.array[...] = np.zeros((10, 10))
def test_read(self):
self.h5file.close()
self.assertRaises(tb.ClosedNodeError, self.array.read)
def test_getitem(self):
self.h5file.close()
self.assertRaises(tb.ClosedNodeError, self.array.__getitem__, 0)
def test_setitem(self):
self.h5file.close()
self.assertRaises(tb.ClosedNodeError, self.array.__setitem__, 0, 0)
class TestCreateCArrayArgs(common.TempFileMixin, common.PyTablesTestCase):
obj = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
where = '/'
name = 'carray'
atom = tb.Atom.from_dtype(obj.dtype)
shape = obj.shape
title = 'title'
filters = None
chunkshape = (1, 2)
byteorder = None
createparents = False
def test_positional_args_01(self):
self.h5file.create_carray(self.where, self.name,
self.atom, self.shape,
self.title, self.filters, self.chunkshape)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(np.zeros_like(self.obj), nparr))
def test_positional_args_02(self):
ptarr = self.h5file.create_carray(self.where, self.name,
self.atom, self.shape,
self.title,
self.filters, self.chunkshape)
ptarr[...] = self.obj
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_positional_args_obj(self):
self.h5file.create_carray(self.where, self.name,
None, None,
self.title,
self.filters, self.chunkshape,
self.byteorder, self.createparents,
self.obj)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_obj(self):
self.h5file.create_carray(self.where, self.name, title=self.title,
chunkshape=self.chunkshape,
obj=self.obj)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_atom_shape_01(self):
ptarr = self.h5file.create_carray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
atom=self.atom, shape=self.shape)
ptarr[...] = self.obj
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_atom_shape_02(self):
ptarr = self.h5file.create_carray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
atom=self.atom, shape=self.shape)
# ptarr[...] = self.obj
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(np.zeros_like(self.obj), nparr))
def test_kwargs_obj_atom(self):
ptarr = self.h5file.create_carray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
atom=self.atom)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_obj_shape(self):
ptarr = self.h5file.create_carray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
shape=self.shape)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_obj_atom_shape(self):
ptarr = self.h5file.create_carray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
atom=self.atom,
shape=self.shape)
self.h5file.close()
self.h5file = tb.open_file(self.h5fname)
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
self.assertTrue(common.allequal(self.obj, nparr))
def test_kwargs_obj_atom_error(self):
atom = tb.Atom.from_dtype(np.dtype('complex'))
# shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_carray,
self.where,
self.name,
title=self.title,
obj=self.obj,
atom=atom)
def test_kwargs_obj_shape_error(self):
# atom = Atom.from_dtype(np.dtype('complex'))
shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_carray,
self.where,
self.name,
title=self.title,
obj=self.obj,
shape=shape)
def test_kwargs_obj_atom_shape_error_01(self):
atom = tb.Atom.from_dtype(np.dtype('complex'))
# shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_carray,
self.where,
self.name,
title=self.title,
obj=self.obj,
atom=atom,
shape=self.shape)
def test_kwargs_obj_atom_shape_error_02(self):
# atom = Atom.from_dtype(np.dtype('complex'))
shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_carray,
self.where,
self.name,
title=self.title,
obj=self.obj,
atom=self.atom,
shape=shape)
def test_kwargs_obj_atom_shape_error_03(self):
atom = tb.Atom.from_dtype(np.dtype('complex'))
shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_carray,
self.where,
self.name,
title=self.title,
obj=self.obj,
atom=atom,
shape=shape)
def suite():
theSuite = common.unittest.TestSuite()
niter = 1
# common.heavy = 1 # uncomment this only for testing purposes
# theSuite.addTest(make_suite(BasicTestCase))
for n in range(niter):
theSuite.addTest(common.make_suite(BasicWriteTestCase))
theSuite.addTest(common.make_suite(BasicWrite2TestCase))
theSuite.addTest(common.make_suite(BasicWrite3TestCase))
theSuite.addTest(common.make_suite(BasicWrite4TestCase))
theSuite.addTest(common.make_suite(BasicWrite5TestCase))
theSuite.addTest(common.make_suite(BasicWrite6TestCase))
theSuite.addTest(common.make_suite(BasicWrite7TestCase))
theSuite.addTest(common.make_suite(BasicWrite8TestCase))
theSuite.addTest(common.make_suite(EmptyCArrayTestCase))
theSuite.addTest(common.make_suite(EmptyCArray2TestCase))
theSuite.addTest(common.make_suite(SlicesCArrayTestCase))
theSuite.addTest(common.make_suite(Slices2CArrayTestCase))
theSuite.addTest(common.make_suite(EllipsisCArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis2CArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis3CArrayTestCase))
theSuite.addTest(common.make_suite(ZlibComprTestCase))
theSuite.addTest(common.make_suite(ZlibShuffleTestCase))
theSuite.addTest(common.make_suite(BloscComprTestCase))
theSuite.addTest(common.make_suite(BloscShuffleTestCase))
theSuite.addTest(common.make_suite(BloscBitShuffleTestCase))
theSuite.addTest(common.make_suite(BloscFletcherTestCase))
theSuite.addTest(common.make_suite(BloscBloscLZTestCase))
theSuite.addTest(common.make_suite(BloscLZ4TestCase))
theSuite.addTest(common.make_suite(BloscLZ4HCTestCase))
theSuite.addTest(common.make_suite(BloscSnappyTestCase))
theSuite.addTest(common.make_suite(BloscZlibTestCase))
theSuite.addTest(common.make_suite(BloscZstdTestCase))
theSuite.addTest(common.make_suite(Blosc2ComprTestCase))
theSuite.addTest(common.make_suite(Blosc2FletcherTestCase))
theSuite.addTest(common.make_suite(Blosc2CrossChunkTestCase))
theSuite.addTest(common.make_suite(Blosc2CrossChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2PastLastChunkTestCase))
theSuite.addTest(common.make_suite(Blosc2PastLastChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2Ndim3MinChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2Ndim3ChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2Ndim4ChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2NDNoChunkshape))
theSuite.addTest(common.make_suite(LZOComprTestCase))
theSuite.addTest(common.make_suite(LZOShuffleTestCase))
theSuite.addTest(common.make_suite(Bzip2ComprTestCase))
theSuite.addTest(common.make_suite(Bzip2ShuffleTestCase))
theSuite.addTest(common.make_suite(FloatTypeTestCase))
theSuite.addTest(common.make_suite(ComplexTypeTestCase))
theSuite.addTest(common.make_suite(StringTestCase))
theSuite.addTest(common.make_suite(String2TestCase))
theSuite.addTest(common.make_suite(StringComprTestCase))
theSuite.addTest(common.make_suite(Int8TestCase))
theSuite.addTest(common.make_suite(Int16TestCase))
theSuite.addTest(common.make_suite(Int32TestCase))
theSuite.addTest(common.make_suite(Float16TestCase))
theSuite.addTest(common.make_suite(Float32TestCase))
theSuite.addTest(common.make_suite(Float64TestCase))
theSuite.addTest(common.make_suite(Float96TestCase))
theSuite.addTest(common.make_suite(Float128TestCase))
theSuite.addTest(common.make_suite(Complex64TestCase))
theSuite.addTest(common.make_suite(Complex128TestCase))
theSuite.addTest(common.make_suite(Complex192TestCase))
theSuite.addTest(common.make_suite(Complex256TestCase))
theSuite.addTest(common.make_suite(ComprTestCase))
theSuite.addTest(common.make_suite(OffsetStrideTestCase))
theSuite.addTest(common.make_suite(Fletcher32TestCase))
theSuite.addTest(common.make_suite(AllFiltersTestCase))
theSuite.addTest(common.make_suite(ReadOutArgumentTests))
theSuite.addTest(common.make_suite(
SizeOnDiskInMemoryPropertyTestCase))
theSuite.addTest(common.make_suite(CloseCopyTestCase))
theSuite.addTest(common.make_suite(OpenCopyTestCase))
theSuite.addTest(common.make_suite(CopyIndex1TestCase))
theSuite.addTest(common.make_suite(CopyIndex2TestCase))
theSuite.addTest(common.make_suite(CopyIndex3TestCase))
theSuite.addTest(common.make_suite(CopyIndex4TestCase))
theSuite.addTest(common.make_suite(CopyIndex5TestCase))
theSuite.addTest(common.make_suite(BigArrayTestCase))
theSuite.addTest(common.make_suite(DfltAtomNoReopen))
theSuite.addTest(common.make_suite(DfltAtomReopen))
theSuite.addTest(common.make_suite(AtomDefaultReprNoReopen))
theSuite.addTest(common.make_suite(AtomDefaultReprReopen))
theSuite.addTest(common.make_suite(TruncateTestCase))
theSuite.addTest(common.make_suite(MDAtomNoReopen))
theSuite.addTest(common.make_suite(MDAtomReopen))
theSuite.addTest(common.make_suite(MDLargeAtomNoReopen))
theSuite.addTest(common.make_suite(MDLargeAtomReopen))
theSuite.addTest(common.make_suite(AccessClosedTestCase))
theSuite.addTest(common.make_suite(TestCreateCArrayArgs))
if common.heavy:
theSuite.addTest(common.make_suite(Slices3CArrayTestCase))
theSuite.addTest(common.make_suite(Slices4CArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis4CArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis5CArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis6CArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis7CArrayTestCase))
theSuite.addTest(common.make_suite(MD3WriteTestCase))
theSuite.addTest(common.make_suite(MD5WriteTestCase))
theSuite.addTest(common.make_suite(MD6WriteTestCase))
theSuite.addTest(common.make_suite(MD7WriteTestCase))
theSuite.addTest(common.make_suite(MD10WriteTestCase))
theSuite.addTest(common.make_suite(CopyIndex6TestCase))
theSuite.addTest(common.make_suite(CopyIndex7TestCase))
theSuite.addTest(common.make_suite(CopyIndex8TestCase))
theSuite.addTest(common.make_suite(CopyIndex9TestCase))
theSuite.addTest(common.make_suite(CopyIndex10TestCase))
theSuite.addTest(common.make_suite(CopyIndex11TestCase))
theSuite.addTest(common.make_suite(CopyIndex12TestCase))
theSuite.addTest(common.make_suite(Rows64bitsTestCase1))
theSuite.addTest(common.make_suite(Rows64bitsTestCase2))
return theSuite
if __name__ == '__main__':
common.parse_argv(sys.argv)
common.print_versions()
common.unittest.main(defaultTest='suite')