File: C:/Users/fred/anaconda3/Lib/site-packages/tables/tests/test_earray.py
import sys
from pathlib import Path
import numpy as np
import tables as tb
from tables.tests import common
class BasicTestCase(common.TempFileMixin, common.PyTablesTestCase):
# Default values
obj = None
flavor = "numpy"
type = 'int32'
dtype = 'int32'
shape = (2, 0)
start = 0
stop = 10
step = 1
length = 1
chunksize = 5
nappends = 10
compress = 0
complib = "zlib" # Default compression library
shuffle = 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,
fletcher32=self.fletcher32)
earray = self.h5file.create_earray(group, 'earray1',
atom=atom, shape=self.shape,
title=title, filters=filters,
expectedrows=1, obj=obj)
earray.flavor = self.flavor
# Fill it with rows
self.rowshape = list(earray.shape)
if obj is not None:
self.rowshape[0] = 0
self.objsize = self.length
for i in self.rowshape:
if i != 0:
self.objsize *= i
self.extdim = earray.extdim
self.objsize *= self.chunksize
self.rowshape[earray.extdim] = self.chunksize
if self.type == "string":
object = np.ndarray(buffer=b"a"*self.objsize,
shape=self.rowshape,
dtype="S%s" % earray.atom.itemsize)
else:
object = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object.shape = self.rowshape
if common.verbose:
if self.flavor == "numpy":
print("Object to append -->", object)
else:
print("Object to append -->", repr(object))
for i in range(self.nappends):
if self.type == "string":
earray.append(object)
else:
earray.append(object * i)
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._reopen()
obj = self.h5file.get_node("/earray1")
shape = self._get_shape()
shape = list(shape)
shape[self.extdim] = self.chunksize * self.nappends
if self.obj is not None:
shape[self.extdim] += len(self.obj)
shape = tuple(shape)
self.assertEqual(obj.flavor, self.flavor)
self.assertEqual(obj.shape, shape)
self.assertEqual(obj.ndim, len(shape))
self.assertEqual(obj.nrows, shape[self.extdim])
self.assertEqual(obj.atom.type, self.type)
def test01_iterEArray(self):
"""Checking enlargeable array iterator."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_iterEArray..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen()
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
earray.nrowsinbuf = 3
if common.verbose:
print("EArray descr:", repr(earray))
print("shape of read array ==>", earray.shape)
print("reopening?:", self.reopen)
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a"*self.objsize,
shape=self.rowshape,
dtype="S%s" % earray.atom.itemsize)
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
shape = self._get_shape()
# Read all the array
for idx, row in enumerate(earray):
if idx < initialrows:
self.assertTrue(common.allequal(
row, np.asarray(self.obj[idx]), self.flavor))
continue
chunk = int((earray.nrow - initialrows) % self.chunksize)
if chunk == 0:
if self.type == "string":
object__ = object_
else:
i = int(earray.nrow - initialrows)
object__ = object_ * (i // self.chunksize)
object = object__[chunk]
# The next adds much more verbosity
if common.verbose and 0:
print("number of row ==>", earray.nrow)
if hasattr(object, "shape"):
print("shape should look as:", object.shape)
print("row in earray ==>", repr(row))
print("Should look like ==>", repr(object))
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
if hasattr(row, "shape"):
self.assertEqual(len(row.shape), len(shape) - 1)
else:
# Scalar case
self.assertEqual(len(shape), 1)
# Check filters:
if self.compress != earray.filters.complevel and common.verbose:
print("Error in compress. Class:", self.__class__.__name__)
print("self, earray:", self.compress, earray.filters.complevel)
self.assertEqual(earray.filters.complevel, self.compress)
if self.compress > 0 and tb.which_lib_version(self.complib):
# Some libraries like Blosc support different compressors,
# specified after ":".
self.assertEqual(earray.filters.complib.split(':')[0], self.complib)
if self.shuffle != earray.filters.shuffle and common.verbose:
print("Error in shuffle. Class:", self.__class__.__name__)
print("self, earray:", self.shuffle, earray.filters.shuffle)
self.assertEqual(self.shuffle, earray.filters.shuffle)
if self.fletcher32 != earray.filters.fletcher32 and common.verbose:
print("Error in fletcher32. Class:", self.__class__.__name__)
print("self, earray:", self.fletcher32,
earray.filters.fletcher32)
self.assertEqual(self.fletcher32, earray.filters.fletcher32)
def test02_sssEArray(self):
"""Checking enlargeable array iterator with (start, stop, step)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_sssEArray..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen()
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
earray.nrowsinbuf = 3
if common.verbose:
print("EArray descr:", repr(earray))
print("shape of read array ==>", earray.shape)
print("reopening?:", self.reopen)
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a"*self.objsize,
shape=self.rowshape,
dtype="S%s" % earray.atom.itemsize)
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
shape = self._get_shape()
# Read all the array
for idx, row in enumerate(earray.iterrows(start=self.start,
stop=self.stop,
step=self.step)):
if idx < initialrows:
self.assertTrue(common.allequal(
row, np.asarray(self.obj[idx]), self.flavor))
continue
if self.chunksize == 1:
index = 0
else:
index = int((earray.nrow - initialrows) % self.chunksize)
if self.type == "string":
object__ = object_
else:
i = int(earray.nrow - initialrows)
object__ = object_ * (i // self.chunksize)
object = object__[index]
# The next adds much more verbosity
if common.verbose and 0:
print("number of row ==>", earray.nrow)
if hasattr(object, "shape"):
print("shape should look as:", object.shape)
print("row in earray ==>", repr(row))
print("Should look like ==>", repr(object))
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
if hasattr(row, "shape"):
self.assertEqual(len(row.shape), len(shape) - 1)
else:
# Scalar case
self.assertEqual(len(shape), 1)
def test03_readEArray(self):
"""Checking read() of enlargeable arrays."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_readEArray..." % self.__class__.__name__)
# This conversion made just in case indices are numpy scalars
if self.start is not None:
self.start = int(self.start)
if self.stop is not None:
self.stop = int(self.stop)
if self.step is not None:
self.step = int(self.step)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen()
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
earray.nrowsinbuf = 3
if common.verbose:
print("EArray descr:", repr(earray))
print("shape of read array ==>", earray.shape)
print("reopening?:", self.reopen)
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a"*self.objsize,
shape=self.rowshape,
dtype="S%s" % earray.atom.itemsize)
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
rowshape = self.rowshape
rowshape[self.extdim] *= (self.nappends + initialrows)
if self.type == "string":
object__ = np.empty(shape=rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object__ = np.empty(shape=rowshape, dtype=self.dtype)
object__ = object__.swapaxes(0, self.extdim)
if initialrows:
object__[0:initialrows] = self.obj
for i in range(self.nappends):
j = initialrows + i * self.chunksize
if self.type == "string":
object__[j:j + self.chunksize] = object_
else:
object__[j:j + self.chunksize] = object_ * i
stop = self.stop
if self.nappends:
# 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 = earray.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 rowshape[self.extdim] < stop:
# self.stop == 0 means last row only in read()
# and not in [::] slicing notation
stop = rowshape[self.extdim]
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
# object = object__[self.start:stop:self.step].copy()
object = object__[self.start:self.stop:self.step].copy()
# Swap the axes again to have normal ordering
if self.flavor == "numpy":
object = object.swapaxes(0, self.extdim)
else:
object = np.empty(shape=self.shape, dtype=self.dtype)
# Read all the array
try:
row = earray.read(self.start, self.stop, self.step)
except IndexError:
row = np.empty(shape=self.shape, dtype=self.dtype)
if common.verbose:
if hasattr(object, "shape"):
print("shape should look as:", object.shape)
print("Object read ==>", repr(row))
print("Should look like ==>", repr(object))
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
shape = self._get_shape()
if hasattr(row, "shape"):
self.assertEqual(len(row.shape), len(shape))
if self.flavor == "numpy":
self.assertEqual(row.itemsize, earray.atom.itemsize)
else:
# Scalar case
self.assertEqual(len(shape), 1)
def test03_readEArray_out_argument(self):
"""Checking read() of enlargeable arrays."""
# This conversion made just in case indices are numpy scalars
if self.start is not None:
self.start = int(self.start)
if self.stop is not None:
self.stop = int(self.stop)
if self.step is not None:
self.step = int(self.step)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen()
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
earray.nrowsinbuf = 3
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a"*self.objsize,
shape=self.rowshape,
dtype="S%s" % earray.atom.itemsize)
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
rowshape = self.rowshape
rowshape[self.extdim] *= (self.nappends + initialrows)
if self.type == "string":
object__ = np.empty(shape=rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object__ = np.empty(shape=rowshape, dtype=self.dtype)
object__ = object__.swapaxes(0, self.extdim)
if initialrows:
object__[0:initialrows] = self.obj
for i in range(self.nappends):
j = initialrows + i * self.chunksize
if self.type == "string":
object__[j:j + self.chunksize] = object_
else:
object__[j:j + self.chunksize] = object_ * i
stop = self.stop
if self.nappends:
# 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 = earray.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 rowshape[self.extdim] < stop:
# self.stop == 0 means last row only in read()
# and not in [::] slicing notation
stop = rowshape[self.extdim]
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
# object = object__[self.start:stop:self.step].copy()
object = object__[self.start:self.stop:self.step].copy()
# Swap the axes again to have normal ordering
if self.flavor == "numpy":
object = object.swapaxes(0, self.extdim)
else:
object = np.empty(shape=self.shape, dtype=self.dtype)
# Read all the array
try:
row = np.empty(earray.shape, dtype=earray.atom.dtype)
slice_obj = [slice(None)] * len(earray.shape)
# slice_obj[earray.maindim] = slice(self.start, stop, self.step)
slice_obj[earray.maindim] = slice(self.start, self.stop, self.step)
row = row[tuple(slice_obj)].copy()
earray.read(self.start, self.stop, self.step, out=row)
except IndexError:
row = np.empty(shape=self.shape, dtype=self.dtype)
if common.verbose:
if hasattr(object, "shape"):
print("shape should look as:", object.shape)
print("Object read ==>", repr(row))
print("Should look like ==>", repr(object))
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
shape = self._get_shape()
if hasattr(row, "shape"):
self.assertEqual(len(row.shape), len(shape))
if self.flavor == "numpy":
self.assertEqual(row.itemsize, earray.atom.itemsize)
else:
# Scalar case
self.assertEqual(len(shape), 1)
def test04_getitemEArray(self):
"""Checking enlargeable array __getitem__ special method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_getitemEArray..." %
self.__class__.__name__)
if not hasattr(self, "slices"):
# If there is not a slices attribute, create it
# This conversion made just in case indices are numpy scalars
if self.start is not None:
self.start = int(self.start)
if self.stop is not None:
self.stop = int(self.stop)
if self.step is not None:
self.step = int(self.step)
self.slices = (slice(self.start, self.stop, self.step),)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen()
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
# earray.nrowsinbuf = 3 # this does not really change the chunksize
if common.verbose:
print("EArray descr:", repr(earray))
print("shape of read array ==>", earray.shape)
print("reopening?:", self.reopen)
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
rowshape = self.rowshape
rowshape[self.extdim] *= (self.nappends + initialrows)
if self.type == "string":
object__ = np.empty(shape=rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object__ = np.empty(shape=rowshape, dtype=self.dtype)
# Additional conversion for the numpy case
object__ = object__.swapaxes(0, earray.extdim)
if initialrows:
object__[0:initialrows] = self.obj
for i in range(self.nappends):
j = initialrows + i * self.chunksize
if self.type == "string":
object__[j:j + self.chunksize] = object_
else:
object__[j:j + self.chunksize] = object_ * i
if self.nappends:
# Swap the axes again to have normal ordering
if self.flavor == "numpy":
object__ = object__.swapaxes(0, self.extdim)
else:
object__.swapaxes(0, self.extdim)
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
object = object__.__getitem__(self.slices).copy()
else:
object = np.empty(shape=self.shape, dtype=self.dtype)
# Read all the array
try:
row = earray.__getitem__(self.slices)
except IndexError:
row = np.empty(shape=self.shape, dtype=self.dtype)
if common.verbose:
print("Object read:\n", repr(row))
print("Should look like:\n", repr(object))
if hasattr(object, "shape"):
print("Original object shape:", self.shape)
print("Shape read:", row.shape)
print("shape should look as:", object.shape)
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
if not hasattr(row, "shape"):
# Scalar case
self.assertEqual(len(self.shape), 1)
def test05_setitemEArray(self):
"""Checking enlargeable array __setitem__ special method."""
if self.__class__.__name__ == "Ellipsis6EArrayTestCase":
# We have a problem with test design here, but I think
# it is not worth the effort to solve it
# F.Alted 2004-10-27
return
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_setitemEArray..." %
self.__class__.__name__)
if not hasattr(self, "slices"):
# If there is not a slices attribute, create it
# This conversion made just in case indices are numpy scalars
if self.start is not None:
self.start = int(self.start)
if self.stop is not None:
self.stop = int(self.stop)
if self.step is not None:
self.step = int(self.step)
self.slices = (slice(self.start, self.stop, self.step),)
# Create an instance of an HDF5 Table
if self.reopen:
self._reopen(mode="a")
earray = self.h5file.get_node("/earray1")
# Choose a small value for buffer size
# earray.nrowsinbuf = 3 # this does not really change the chunksize
if common.verbose:
print("EArray descr:", repr(earray))
print("shape of read array ==>", earray.shape)
print("reopening?:", self.reopen)
# Build the array to do comparisons
if self.type == "string":
object_ = np.ndarray(buffer=b"a" * self.objsize,
shape=self.rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object_ = np.arange(self.objsize, dtype=earray.atom.dtype.base)
object_.shape = self.rowshape
object_ = object_.swapaxes(earray.extdim, 0)
if self.obj is not None:
initialrows = len(self.obj)
else:
initialrows = 0
rowshape = self.rowshape
rowshape[self.extdim] *= (self.nappends + initialrows)
if self.type == "string":
object__ = np.empty(shape=rowshape,
dtype=f"S{earray.atom.itemsize}")
else:
object__ = np.empty(shape=rowshape, dtype=self.dtype)
# Additional conversion for the numpy case
object__ = object__.swapaxes(0, earray.extdim)
for i in range(self.nappends):
j = initialrows + i * self.chunksize
if self.type == "string":
object__[j:j + self.chunksize] = object_
else:
object__[j:j + self.chunksize] = object_ * i
# Modify the earray
# earray[j:j + self.chunksize] = object_ * i
# earray[self.slices] = 1
if initialrows:
object__[0:initialrows] = self.obj
if self.nappends:
# Swap the axes again to have normal ordering
if self.flavor == "numpy":
object__ = object__.swapaxes(0, self.extdim)
else:
object__.swapaxes(0, self.extdim)
# do a copy() in order to ensure that len(object._data)
# actually do a measure of its length
object = object__.__getitem__(self.slices).copy()
else:
object = np.empty(shape=self.shape, dtype=self.dtype)
if self.flavor == "numpy":
object = np.asarray(object)
if self.type == "string":
if hasattr(self, "wslice"):
# These sentences should be equivalent
# object[self.wslize] = object[self.wslice].pad("xXx")
# earray[self.wslice] = earray[self.wslice].pad("xXx")
object[self.wslize] = "xXx"
earray[self.wslice] = "xXx"
elif sum(object[self.slices].shape) != 0:
# object[:] = object.pad("xXx")
object[:] = "xXx"
if object.size > 0:
earray[self.slices] = object
else:
if hasattr(self, "wslice"):
object[self.wslice] = object[self.wslice] * 2 + 3
earray[self.wslice] = earray[self.wslice] * 2 + 3
elif sum(object[self.slices].shape) != 0:
object = object * 2 + 3
if np.prod(object.shape) > 0:
earray[self.slices] = earray[self.slices] * 2 + 3
# Read all the array
row = earray.__getitem__(self.slices)
try:
row = earray.__getitem__(self.slices)
except IndexError:
print("IndexError!")
row = np.empty(shape=self.shape, dtype=self.dtype)
if common.verbose:
print("Object read:\n", repr(row))
print("Should look like:\n", repr(object))
if hasattr(object, "shape"):
print("Original object shape:", self.shape)
print("Shape read:", row.shape)
print("shape should look as:", object.shape)
self.assertEqual(initialrows + self.nappends * self.chunksize,
earray.nrows)
self.assertTrue(common.allequal(row, object, self.flavor))
if not hasattr(row, "shape"):
# Scalar case
self.assertEqual(len(self.shape), 1)
class BasicWriteTestCase(BasicTestCase):
type = 'int32'
shape = (0,)
chunksize = 5
nappends = 10
step = 1
# wslice = slice(1,nappends,2)
wslice = 1 # single element case
class Basic2WriteTestCase(BasicTestCase):
type = 'int32'
dtype = 'i4'
shape = (0,)
chunksize = 5
nappends = 10
step = 1
wslice = slice(chunksize-2, nappends, 2) # range of elements
reopen = 0 # This case does not reopen files
class Basic3WriteTestCase(BasicTestCase):
obj = [1, 2]
type = np.asarray(obj).dtype.name
dtype = np.asarray(obj).dtype.str
shape = (0,)
chunkshape = (5,)
step = 1
reopen = 0 # This case does not reopen files
class Basic4WriteTestCase(BasicTestCase):
obj = np.array([1, 2])
type = obj.dtype.name
dtype = obj.dtype.str
shape = None
chunkshape = (5,)
step = 1
reopen = 0 # This case does not reopen files
class Basic5WriteTestCase(BasicTestCase):
obj = [1, 2]
type = np.asarray(obj).dtype.name
dtype = np.asarray(obj).dtype.str
shape = (0,)
chunkshape = (5,)
step = 1
reopen = 1 # This case does reopen files
class Basic6WriteTestCase(BasicTestCase):
obj = np.array([1, 2])
type = obj.dtype.name
dtype = obj.dtype.str
shape = None
chunkshape = (5,)
step = 1
reopen = 1 # This case does reopen files
class Basic7WriteTestCase(BasicTestCase):
obj = [[1, 2], [3, 4]]
type = np.asarray(obj).dtype.name
dtype = np.asarray(obj).dtype.str
shape = (0, 2)
chunkshape = (5,)
step = 1
reopen = 0 # This case does not reopen files
class Basic8WriteTestCase(BasicTestCase):
obj = [[1, 2], [3, 4]]
type = np.asarray(obj).dtype.name
dtype = np.asarray(obj).dtype.str
shape = (0, 2)
chunkshape = (5,)
step = 1
reopen = 1 # This case does reopen files
class EmptyEArrayTestCase(BasicTestCase):
type = 'int32'
dtype = np.dtype('int32')
shape = (2, 0)
chunksize = 5
nappends = 0
start = 0
stop = 10
step = 1
class NP_EmptyEArrayTestCase(BasicTestCase):
type = 'int32'
dtype = np.dtype('()int32')
shape = (2, 0)
chunksize = 5
nappends = 0
class Empty2EArrayTestCase(BasicTestCase):
type = 'int32'
dtype = 'int32'
shape = (2, 0)
chunksize = 5
nappends = 0
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 SlicesEArrayTestCase(BasicTestCase):
compress = 1
complib = "lzo"
type = 'int32'
shape = (2, 0)
chunksize = 5
nappends = 2
slices = (slice(1, 2, 1), slice(1, 3, 1))
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class Slices2EArrayTestCase(BasicTestCase):
compress = 1
complib = "blosc"
type = 'int32'
shape = (2, 0, 4)
chunksize = 5
nappends = 20
slices = (slice(1, 2, 1), slice(None, None, None), slice(1, 4, 2))
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2SlicesOptEArrayTestCase(BasicTestCase):
compress = 1
complib = "blosc2"
type = 'int32'
shape = (0, 13, 13)
chunkshape = (4, 4, 4)
nappends = 20
slices = (slice(None, None), slice(2, 10), slice(0, 10))
@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
@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 = (0, 10)
compress = 1 # sss
complib = "blosc2"
chunkshape = (4, 4)
nappends = 10
start = 3
stop = 6
step = 3
@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 Blosc2InnerCrossChunkTestCase(Blosc2CrossChunkTestCase):
shape = (10, 0)
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2InnerCrossChunkOptTestCase(Blosc2InnerCrossChunkTestCase):
step = 1 # optimized
byteorder = sys.byteorder
@common.unittest.skipIf(not common.blosc2_avail,
'BLOSC2 compression library not available')
class Blosc2PastLastChunkTestCase(BasicTestCase):
shape = (0, 10)
compress = 1 # sss
complib = "blosc2"
chunkshape = (4, 4)
nappends = 10
start = 8
stop = 100
step = 3
class EllipsisEArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 0)
chunksize = 5
nappends = 2
# slices = (slice(1,2,1), Ellipsis)
slices = (Ellipsis, slice(1, 2, 1))
class Ellipsis2EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 0, 4)
chunksize = 5
nappends = 20
slices = (slice(1, 2, 1), Ellipsis, slice(1, 4, 2))
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class Slices3EArrayTestCase(BasicTestCase):
compress = 1 # To show the chunks id DEBUG is on
complib = "blosc"
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 20
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 Slices4EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0, 5, 6)
chunksize = 5
nappends = 20
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 Ellipsis3EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 20
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 Ellipsis4EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 20
slices = (Ellipsis, slice(0, 4, None), slice(1, 4, 2))
slices = (slice(1, 2, 1), Ellipsis, slice(1, 4, 2))
class Ellipsis5EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 20
slices = (slice(1, 2, 1), slice(0, 4, None), Ellipsis)
class Ellipsis6EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 2
# The next slices gives problems with setting values (test05)
# This is a problem on the test design, not the Array.__setitem__
# code, though.
slices = (slice(1, 2, 1), slice(0, 4, None), 2, Ellipsis)
class Ellipsis7EArrayTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 4, 0)
chunksize = 5
nappends = 2
slices = (slice(1, 2, 1), slice(0, 4, None), slice(2, 3), Ellipsis)
class MD3WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 0, 3)
chunksize = 4
step = 2
class MD5WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 0, 3, 4, 5) # ok
# shape = (1, 1, 0, 1) # Minimum shape that shows problems with HDF5 1.6.1
# shape = (2, 3, 0, 4, 5) # Floating point exception (HDF5 1.6.1)
# shape = (2, 3, 3, 0, 5, 6) # Segmentation fault (HDF5 1.6.1)
chunksize = 1
nappends = 1
start = 1
stop = 10
step = 10
class MD6WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 3, 0, 5, 6)
chunksize = 1
nappends = 10
start = 1
stop = 10
step = 3
class NP_MD6WriteTestCase(BasicTestCase):
"""Testing NumPy scalars as indexes"""
type = 'int32'
shape = (2, 3, 3, 0, 5, 6)
chunksize = 1
nappends = 10
class MD6WriteTestCase__(BasicTestCase):
type = 'int32'
shape = (2, 0)
chunksize = 1
nappends = 3
start = 1
stop = 3
step = 1
class MD7WriteTestCase(BasicTestCase):
type = 'int32'
shape = (2, 3, 3, 4, 5, 0, 3)
chunksize = 10
nappends = 1
start = 1
stop = 10
step = 2
class MD10WriteTestCase(BasicTestCase):
type = 'int32'
shape = (1, 2, 3, 4, 5, 5, 4, 3, 2, 0)
chunksize = 5
nappends = 10
start = -1
stop = -1
step = 10
class NP_MD10WriteTestCase(BasicTestCase):
type = 'int32'
shape = (1, 2, 3, 4, 5, 5, 4, 3, 2, 0)
chunksize = 5
nappends = 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"
chunksize = 10
nappends = 100
start = 3
stop = 10
step = 3
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscShuffleTestCase(BasicTestCase):
compress = 1
shuffle = 1
complib = "blosc"
chunksize = 100
nappends = 10
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class LZOComprTestCase(BasicTestCase):
compress = 1 # sss
complib = "lzo"
chunksize = 10
nappends = 100
start = 3
stop = 10
step = 3
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class LZOShuffleTestCase(BasicTestCase):
compress = 1
shuffle = 1
complib = "lzo"
chunksize = 100
nappends = 10
start = 3
stop = 10
step = 7
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class Bzip2ComprTestCase(BasicTestCase):
compress = 1
complib = "bzip2"
chunksize = 100
nappends = 10
start = 3
stop = 10
step = 8
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class Bzip2ShuffleTestCase(BasicTestCase):
compress = 1
shuffle = 1
complib = "bzip2"
chunksize = 100
nappends = 10
start = 3
stop = 10
step = 6
class Fletcher32TestCase(BasicTestCase):
compress = 0
fletcher32 = 1
chunksize = 50
nappends = 20
start = 4
stop = 20
step = 7
class AllFiltersTestCase(BasicTestCase):
compress = 1
shuffle = 1
fletcher32 = 1
complib = "zlib"
chunksize = 20 # sss
nappends = 50
start = 2
stop = 99
step = 6
# chunksize = 3
# nappends = 2
# start = 1
# stop = 10
# step = 2
class FloatTypeTestCase(BasicTestCase):
type = 'float64'
dtype = 'float64'
shape = (2, 0)
chunksize = 5
nappends = 10
start = 3
stop = 10
step = 20
class ComplexTypeTestCase(BasicTestCase):
type = 'complex128'
dtype = 'complex128'
shape = (2, 0)
chunksize = 5
nappends = 10
start = 3
stop = 10
step = 20
class StringTestCase(BasicTestCase):
type = "string"
length = 20
shape = (2, 0)
# shape = (2,0,20)
chunksize = 5
nappends = 10
start = 3
stop = 10
step = 20
slices = (slice(0, 1), slice(1, 2))
class String2TestCase(BasicTestCase):
type = "string"
length = 20
shape = (0,)
# shape = (0, 20)
chunksize = 5
nappends = 10
start = 1
stop = 10
step = 2
class StringComprTestCase(BasicTestCase):
type = "string"
length = 20
shape = (20, 0, 10)
# shape = (20,0,10,20)
compr = 1
# shuffle = 1 # this shouldn't do nothing on chars
chunksize = 50
nappends = 10
start = -1
stop = 100
step = 20
class SizeOnDiskInMemoryPropertyTestCase(common.TempFileMixin,
common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.array_size = (0, 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_earray('/', 'earray',
atom=tb.Int32Atom(),
shape=self.array_size,
filters=filters,
chunkshape=self.chunkshape)
def test_zero_length(self):
complevel = 0
self.create_array(complevel)
self.assertEqual(self.array.size_on_disk, 0)
self.assertEqual(self.array.size_in_memory, 0)
# add 10 chunks of data in one append
def test_no_compression_one_append(self):
complevel = 0
self.create_array(complevel)
self.array.append([tuple(range(10))] * self.chunkshape[0] * 10)
self.assertEqual(self.array.size_on_disk, 10 * 1000 * 10 * 4)
self.assertEqual(self.array.size_in_memory, 10 * 1000 * 10 * 4)
# add 10 chunks of data in two appends
def test_no_compression_multiple_appends(self):
complevel = 0
self.create_array(complevel)
self.array.append([tuple(range(10))] * self.chunkshape[0] * 5)
self.array.append([tuple(range(10))] * self.chunkshape[0] * 5)
self.assertEqual(self.array.size_on_disk, 10 * 1000 * 10 * 4)
self.assertEqual(self.array.size_in_memory, 10 * 1000 * 10 * 4)
def test_with_compression(self):
complevel = 1
self.create_array(complevel)
self.array.append([tuple(range(10))] * self.chunkshape[0] * 10)
file_size = Path(self.h5fname).stat().st_size
self.assertTrue(
abs(self.array.size_on_disk - file_size) <= self.hdf_overhead)
self.assertEqual(self.array.size_in_memory, 10 * 1000 * 10 * 4)
self.assertLess(self.array.size_on_disk, self.array.size_in_memory)
class OffsetStrideTestCase(common.TempFileMixin, common.PyTablesTestCase):
mode = "w"
compress = 0
complib = "zlib" # Default compression library
def setUp(self):
super().setUp()
self.rootgroup = self.h5file.root
def test01a_String(self):
"""Checking earray with offset numpy strings appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01a_StringAtom..." % self.__class__.__name__)
earray = self.h5file.create_earray(root, 'strings',
atom=tb.StringAtom(itemsize=3),
shape=(0, 2, 2),
title="Array of strings")
a = np.array([[["a", "b"], ["123", "45"], ["45", "123"]]], dtype="S3")
earray.append(a[:, 1:])
a = np.array([[["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"]]])
earray.append(a[:, 2:])
# Read all the rows:
row = earray.read()
if common.verbose:
print("Object read:", row)
print("Nrows in", earray._v_pathname, ":", earray.nrows)
print("Second row in earray ==>", row[1].tolist())
self.assertEqual(earray.nrows, 2)
self.assertEqual(row[0].tolist(), [[b"123", b"45"], [b"45", b"123"]])
self.assertEqual(row[1].tolist(), [[b"s", b"abc"], [b"abc", b"f"]])
self.assertEqual(len(row[0]), 2)
self.assertEqual(len(row[1]), 2)
def test01b_String(self):
"""Checking earray with strided numpy strings appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01b_StringAtom..." % self.__class__.__name__)
earray = self.h5file.create_earray(root, 'strings',
atom=tb.StringAtom(itemsize=3),
shape=(0, 2, 2),
title="Array of strings")
a = np.array([[["a", "b"], ["123", "45"], ["45", "123"]]], dtype="S3")
earray.append(a[:, ::2])
a = np.array([[["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"]]])
earray.append(a[:, ::2])
# Read all the rows:
row = earray.read()
if common.verbose:
print("Object read:", row)
print("Nrows in", earray._v_pathname, ":", earray.nrows)
print("Second row in earray ==>", row[1].tolist())
self.assertEqual(earray.nrows, 2)
self.assertEqual(row[0].tolist(), [[b"a", b"b"], [b"45", b"123"]])
self.assertEqual(row[1].tolist(), [[b"s", b"a"], [b"s", b"abc"]])
self.assertEqual(len(row[0]), 2)
self.assertEqual(len(row[1]), 2)
def test02a_int(self):
"""Checking earray with offset NumPy ints appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02a_int..." % self.__class__.__name__)
# Create a string atom
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Int32Atom(), shape=(0, 3),
title="array of ints")
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (0, 0, 0)],
dtype='int32')
earray.append(a[2:]) # Create an offset
a = np.array([(1, 1, 1), (-1, 0, 0)], dtype='int32')
earray.append(a[1:]) # Create an offset
# Read all the rows:
row = earray.read()
if common.verbose:
print("Object read:", row)
print("Nrows in", earray._v_pathname, ":", earray.nrows)
print("Third row in vlarray ==>", row[2])
self.assertEqual(earray.nrows, 3)
self.assertTrue(common.allequal(
row[0], np.array([1, 1, 1], dtype='int32')))
self.assertTrue(common.allequal(
row[1], np.array([0, 0, 0], dtype='int32')))
self.assertTrue(common.allequal(
row[2], np.array([-1, 0, 0], dtype='int32')))
def test02b_int(self):
"""Checking earray with strided NumPy ints appends."""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02b_int..." % self.__class__.__name__)
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Int32Atom(), shape=(0, 3),
title="array of ints")
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='int32')
earray.append(a[::3]) # Create an offset
a = np.array([(1, 1, 1), (-1, 0, 0)], dtype='int32')
earray.append(a[::2]) # Create an offset
# Read all the rows:
row = earray.read()
if common.verbose:
print("Object read:", row)
print("Nrows in", earray._v_pathname, ":", earray.nrows)
print("Third row in vlarray ==>", row[2])
self.assertEqual(earray.nrows, 3)
self.assertTrue(common.allequal(
row[0], np.array([0, 0, 0], dtype='int32')))
self.assertTrue(common.allequal(
row[1], np.array([3, 3, 3], dtype='int32')))
self.assertTrue(common.allequal(
row[2], np.array([1, 1, 1], dtype='int32')))
def test03a_int(self):
"""Checking earray with byteswapped appends (ints)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03a_int..." % self.__class__.__name__)
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Int32Atom(), shape=(0, 3),
title="array of ints")
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='int32')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Native rows:", native)
print("Byteorder native rows:", native.dtype.byteorder)
print("Swapped rows:", swapped)
print("Byteorder swapped rows:", swapped.dtype.byteorder)
self.assertTrue(common.allequal(native, swapped))
def test03b_float(self):
"""Checking earray with byteswapped appends (floats)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03b_float..." % self.__class__.__name__)
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Float64Atom(), shape=(0, 3),
title="array of floats")
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='float64')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Native rows:", native)
print("Byteorder native rows:", native.dtype.byteorder)
print("Swapped rows:", swapped)
print("Byteorder swapped rows:", swapped.dtype.byteorder)
self.assertTrue(common.allequal(native, swapped))
def test04a_int(self):
"""Checking earray with byteswapped appends (2, ints)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04a_int..." % self.__class__.__name__)
byteorder = {'little': 'big', 'big': 'little'}[sys.byteorder]
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Int32Atom(), shape=(0, 3),
title="array of ints",
byteorder=byteorder)
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='int32')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Byteorder native rows:",
tb.utils.byteorders[native.dtype.byteorder])
print("Byteorder earray on-disk:", earray.byteorder)
self.assertEqual(tb.utils.byteorders[native.dtype.byteorder],
sys.byteorder)
self.assertEqual(earray.byteorder, byteorder)
self.assertTrue(common.allequal(native, swapped))
def test04b_int(self):
"""Checking earray with byteswapped appends (2, ints, reopen)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04b_int..." % self.__class__.__name__)
byteorder = {'little': 'big', 'big': 'little'}[sys.byteorder]
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Int32Atom(), shape=(0, 3),
title="array of ints",
byteorder=byteorder)
self._reopen(mode="a")
earray = self.h5file.get_node("/EAtom")
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='int32')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Byteorder native rows:",
tb.utils.byteorders[native.dtype.byteorder])
print("Byteorder earray on-disk:", earray.byteorder)
self.assertEqual(tb.utils.byteorders[native.dtype.byteorder],
sys.byteorder)
self.assertEqual(earray.byteorder, byteorder)
self.assertTrue(common.allequal(native, swapped))
def test04c_float(self):
"""Checking earray with byteswapped appends (2, floats)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04c_float..." % self.__class__.__name__)
byteorder = {'little': 'big', 'big': 'little'}[sys.byteorder]
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Float64Atom(), shape=(0, 3),
title="array of floats",
byteorder=byteorder)
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='float64')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Byteorder native rows:",
tb.utils.byteorders[native.dtype.byteorder])
print("Byteorder earray on-disk:", earray.byteorder)
self.assertEqual(tb.utils.byteorders[native.dtype.byteorder],
sys.byteorder)
self.assertEqual(earray.byteorder, byteorder)
self.assertTrue(common.allequal(native, swapped))
def test04d_float(self):
"""Checking earray with byteswapped appends (2, floats, reopen)"""
root = self.rootgroup
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04d_float..." % self.__class__.__name__)
byteorder = {'little': 'big', 'big': 'little'}[sys.byteorder]
earray = self.h5file.create_earray(root, 'EAtom',
atom=tb.Float64Atom(), shape=(0, 3),
title="array of floats",
byteorder=byteorder)
self._reopen(mode='a')
earray = self.h5file.get_node("/EAtom")
# Add a native ordered array
a = np.array([(0, 0, 0), (1, 0, 3), (1, 1, 1), (3, 3, 3)],
dtype='float64')
earray.append(a)
# Change the byteorder of the array
a = a.byteswap()
a = a.view(a.dtype.newbyteorder())
# Add a byteswapped array
earray.append(a)
# Read all the rows:
native = earray[:4, :]
swapped = earray[4:, :]
if common.verbose:
print("Byteorder native rows:",
tb.utils.byteorders[native.dtype.byteorder])
print("Byteorder earray on-disk:", earray.byteorder)
self.assertEqual(tb.utils.byteorders[native.dtype.byteorder],
sys.byteorder)
self.assertEqual(earray.byteorder, byteorder)
self.assertTrue(common.allequal(native, swapped))
class CopyTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test01_copy(self):
"""Checking EArray.copy() method."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_copy..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.append(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.atom.itemsize, array2.atom.itemsize)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
def test02_copy(self):
"""Checking EArray.copy() method (where specified)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_copy..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.append(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.atom.itemsize, array2.atom.itemsize)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
def test03a_copy(self):
"""Checking EArray.copy() method (python flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03b_copy..." % self.__class__.__name__)
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.flavor = "python"
array1.append(((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.atom.itemsize, array2.atom.itemsize)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
def test03b_copy(self):
"""Checking EArray.copy() method (python string flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03d_copy..." % self.__class__.__name__)
atom = tb.StringAtom(itemsize=3)
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.flavor = "python"
array1.append([["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.atom.itemsize, array2.atom.itemsize)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
def test03c_copy(self):
"""Checking EArray.copy() method (String flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03e_copy..." % self.__class__.__name__)
atom = tb.StringAtom(itemsize=4)
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.flavor = "numpy"
array1.append(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.atom.itemsize, array2.atom.itemsize)
self.assertEqual(array1.title, array2.title)
self.assertEqual(str(array1.atom), str(array2.atom))
def test04_copy(self):
"""Checking EArray.copy() method (checking title copying)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_copy..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.append(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 EArray.copy() method (user attributes copied)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_copy..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.append(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 EArray.copy() method (user attributes not copied)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05b_copy..." % self.__class__.__name__)
# Create an Array
atom = tb.Int16Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
array1.append(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 EArray.copy() method with indexes."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_index..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int32Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
r = np.arange(200, dtype='int32')
r.shape = (100, 2)
array1.append(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])
self.assertEqual(r2.shape[0], array2.nrows)
def test02_indexclosef(self):
"""Checking EArray.copy() method with indexes (close file version)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_indexclosef..." % self.__class__.__name__)
# Create an EArray
atom = tb.Int32Atom()
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
r = np.arange(200, dtype='int32')
r.shape = (100, 2)
array1.append(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.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
class TruncateTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
# Create an EArray
atom = tb.Int16Atom(dflt=3)
array1 = self.h5file.create_earray(self.h5file.root, 'array1',
atom=atom, shape=(0, 2),
title="title array1")
# Add a couple of rows
array1.append(np.array([[456, 2], [3, 457]], dtype='int16'))
def test00_truncate(self):
"""Checking EArray.truncate() method (truncating to 0 rows)"""
array1 = self.h5file.root.array1
# Truncate to 0 elements
array1.truncate(0)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
if common.verbose:
print("array1-->", array1.read())
self.assertTrue(common.allequal(
array1[:], np.array([], dtype='int16').reshape(0, 2)))
def test01_truncate(self):
"""Checking EArray.truncate() method (truncating to 1 rows)"""
array1 = self.h5file.root.array1
# Truncate to 1 element
array1.truncate(1)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
if common.verbose:
print("array1-->", array1.read())
self.assertTrue(common.allequal(
array1.read(), np.array([[456, 2]], dtype='int16')))
def test02_truncate(self):
"""Checking EArray.truncate() method (truncating to == self.nrows)"""
array1 = self.h5file.root.array1
# Truncate to 2 elements
array1.truncate(2)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
if common.verbose:
print("array1-->", array1.read())
self.assertTrue(common.allequal(
array1.read(), np.array([[456, 2], [3, 457]], dtype='int16')))
def test03_truncate(self):
"""Checking EArray.truncate() method (truncating to > self.nrows)"""
array1 = self.h5file.root.array1
# Truncate to 4 elements
array1.truncate(4)
if self.close:
if common.verbose:
print("(closing file version)")
self._reopen()
array1 = self.h5file.root.array1
if common.verbose:
print("array1-->", array1.read())
self.assertEqual(array1.nrows, 4)
# Check the original values
self.assertTrue(common.allequal(
array1[:2], np.array([[456, 2], [3, 457]], dtype='int16')))
# Check that the added rows have the default values
self.assertTrue(common.allequal(
array1[2:], np.array([[3, 3], [3, 3]], dtype='int16')))
class TruncateOpenTestCase(TruncateTestCase):
close = 0
class TruncateCloseTestCase(TruncateTestCase):
close = 1
# The next test should be run only in **common.heavy** mode
class Rows64bitsTestCase(common.TempFileMixin, common.PyTablesTestCase):
open_mode = 'a'
narows = 1000 * 1000 # each numpy object 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 EArray
array = self.h5file.create_earray(
self.h5file.root, 'array',
atom=tb.Int8Atom(), shape=(0,),
filters=tb.Filters(complib='lzo', complevel=1),
# Specifying expectedrows takes more
# CPU, but less disk
expectedrows=self.narows * self.nanumber)
# Fill the array
na = np.arange(self.narows, dtype='int8')
for i in range(self.nanumber):
array.append(na)
def test01_basiccheck(self):
"""Some basic checks for earrays 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)
# Close 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("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
# Test for appending zero-sized arrays
class ZeroSizedTestCase(common.TempFileMixin, common.PyTablesTestCase):
open_mode = 'a'
def setUp(self):
super().setUp()
# Create an EArray
ea = self.h5file.create_earray('/', 'test',
atom=tb.Int32Atom(), shape=(3, 0))
# Append a single row
ea.append([[1], [2], [3]])
def test01_canAppend(self):
"""Appending zero length array."""
fileh = self.h5file
ea = fileh.root.test
arr = np.empty(shape=(3, 0), dtype='int32')
ea.append(arr)
self.assertEqual(ea.nrows, 1, "The number of rows should be 1.")
def test02_appendWithWrongShape(self):
"""Appending zero length array with wrong dimension."""
fileh = self.h5file
ea = fileh.root.test
arr = np.empty(shape=(3, 0, 3), dtype='int32')
self.assertRaises(ValueError, ea.append, arr)
# Test for dealing with multidimensional atoms
class MDAtomTestCase(common.TempFileMixin, common.PyTablesTestCase):
def test01a_append(self):
"""Append a row to a (unidimensional) EArray with a MD tables.Atom."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test',
atom=tb.Int32Atom((2, 2)), shape=(0,))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append one row
ea.append([[[1, 3], [4, 5]]])
self.assertEqual(ea.nrows, 1)
if common.verbose:
print("First row-->", ea[0])
self.assertTrue(common.allequal(
ea[0], np.array([[1, 3], [4, 5]], 'i4')))
def test01b_append(self):
"""Append several rows to a (unidimensional) EArray with a MD
tables.Atom."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test',
atom=tb.Int32Atom((2, 2)), shape=(0,))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append three rows
ea.append([[[1]], [[2]], [[3]]]) # Simple broadcast
self.assertEqual(ea.nrows, 3)
if common.verbose:
print("Third row-->", ea[2])
self.assertTrue(common.allequal(
ea[2], np.array([[3, 3], [3, 3]], 'i4')))
def test02a_append(self):
"""Append a row to a (multidimensional) EArray with a
MD tables.Atom."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test',
atom=tb.Int32Atom((2,)), shape=(0, 3))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append one row
ea.append([[[1, 3], [4, 5], [7, 9]]])
self.assertEqual(ea.nrows, 1)
if common.verbose:
print("First row-->", ea[0])
self.assertTrue(common.allequal(
ea[0], np.array([[1, 3], [4, 5], [7, 9]], 'i4')))
def test02b_append(self):
"""Append several rows to a (multidimensional) EArray with a MD
tables.Atom."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test',
atom=tb.Int32Atom((2,)), shape=(0, 3))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append three rows
ea.append([[[1, -3], [4, -5], [-7, 9]],
[[-1, 3], [-4, 5], [7, -8]],
[[-2, 3], [-5, 5], [7, -9]]])
self.assertEqual(ea.nrows, 3)
if common.verbose:
print("Third row-->", ea[2])
self.assertTrue(common.allequal(
ea[2], np.array([[-2, 3], [-5, 5], [7, -9]], 'i4')))
def test03a_MDMDMD(self):
"""Complex append of a MD array in a MD EArray with a
MD tables.Atom."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test', atom=tb.Int32Atom((2, 4)),
shape=(0, 2, 3))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append three rows
# 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))
ea.append([a * 1, a*2, a*3])
self.assertEqual(ea.nrows, 3)
if common.verbose:
print("Third row-->", ea[2])
self.assertTrue(common.allequal(ea[2], a * 3))
def test03b_MDMDMD(self):
"""Complex append of a MD array in a MD EArray with a MD atom (II)."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test', atom=tb.Int32Atom((2, 4)),
shape=(2, 0, 3))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append three rows
# 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, 1, 3, 2, 4))
ea.append(a * 1)
ea.append(a * 2)
ea.append(a * 3)
self.assertEqual(ea.nrows, 3)
if common.verbose:
print("Third row-->", ea[:, 2, ...])
self.assertTrue(common.allequal(ea[:, 2, ...],
a.reshape((2, 3, 2, 4))*3))
def test03c_MDMDMD(self):
"""Complex append of a MD array in a MD EArray with a MD atom (III)."""
# Create an EArray
ea = self.h5file.create_earray('/', 'test', atom=tb.Int32Atom((2, 4)),
shape=(2, 3, 0))
if self.reopen:
self._reopen('a')
ea = self.h5file.root.test
# Append three rows
# 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, 1, 2, 4))
ea.append(a * 1)
ea.append(a * 2)
ea.append(a * 3)
self.assertEqual(ea.nrows, 3)
if common.verbose:
print("Third row-->", ea[:, :, 2, ...])
self.assertTrue(common.allequal(ea[:, :, 2, ...],
a.reshape((2, 3, 2, 4))*3))
class MDAtomNoReopen(MDAtomTestCase):
reopen = False
class MDAtomReopen(MDAtomTestCase):
reopen = True
class AccessClosedTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.array = self.h5file.create_earray(self.h5file.root, 'array',
atom=tb.Int32Atom(),
shape=(0, 10))
self.array.append(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)
def test_append(self):
self.h5file.close()
self.assertRaises(tb.ClosedNodeError,
self.array.append, np.zeros((10, 10)))
class TestCreateEArrayArgs(common.TempFileMixin, common.PyTablesTestCase):
obj = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])
where = '/'
name = 'earray'
atom = tb.Atom.from_dtype(obj.dtype)
shape = (0,) + obj.shape[1:]
title = 'title'
filters = None
expectedrows = 1000
chunkshape = (1, 2)
byteorder = None
createparents = False
def test_positional_args_01(self):
self.h5file.create_earray(self.where, self.name,
self.atom, self.shape,
self.title, self.filters,
self.expectedrows, self.chunkshape)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.nrows, 0)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
def test_positional_args_02(self):
ptarr = self.h5file.create_earray(self.where, self.name,
self.atom, self.shape,
self.title,
self.filters,
self.expectedrows,
self.chunkshape)
ptarr.append(self.obj)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name,
None, None,
self.title,
self.filters,
self.expectedrows,
self.chunkshape,
self.byteorder,
self.createparents,
self.obj)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name, title=self.title,
chunkshape=self.chunkshape,
obj=self.obj)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
atom=self.atom, shape=self.shape)
ptarr.append(self.obj)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
atom=self.atom, shape=self.shape)
# ptarr.append(self.obj)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.shape)
self.assertEqual(ptarr.nrows, 0)
self.assertEqual(ptarr.atom, self.atom)
self.assertEqual(ptarr.atom.dtype, self.atom.dtype)
self.assertEqual(ptarr.chunkshape, self.chunkshape)
def test_kwargs_obj_atom(self):
ptarr = self.h5file.create_earray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
atom=self.atom)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
shape=self.shape)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray(self.where, self.name,
title=self.title,
chunkshape=self.chunkshape,
obj=self.obj,
atom=self.atom,
shape=self.shape)
self._reopen()
ptarr = self.h5file.get_node(self.where, self.name)
nparr = ptarr.read()
self.assertEqual(ptarr.title, self.title)
self.assertEqual(ptarr.shape, self.obj.shape)
self.assertEqual(ptarr.nrows, self.obj.shape[0])
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_earray,
self.where,
self.name,
title=self.title,
obj=self.obj,
atom=atom)
def test_kwargs_obj_shape_error(self):
# atom = tables.Atom.from_dtype(np.dtype('complex'))
shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_earray,
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_earray,
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 = tables.Atom.from_dtype(np.dtype('complex'))
shape = self.shape + self.shape
self.assertRaises(TypeError,
self.h5file.create_earray,
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_earray,
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(BasicWriteTestCase))
# theSuite.addTest(make_suite(Rows64bitsTestCase1))
# theSuite.addTest(make_suite(Rows64bitsTestCase2))
for n in range(niter):
theSuite.addTest(common.make_suite(BasicWriteTestCase))
theSuite.addTest(common.make_suite(Basic2WriteTestCase))
theSuite.addTest(common.make_suite(Basic3WriteTestCase))
theSuite.addTest(common.make_suite(Basic4WriteTestCase))
theSuite.addTest(common.make_suite(Basic5WriteTestCase))
theSuite.addTest(common.make_suite(Basic6WriteTestCase))
theSuite.addTest(common.make_suite(Basic7WriteTestCase))
theSuite.addTest(common.make_suite(Basic8WriteTestCase))
theSuite.addTest(common.make_suite(EmptyEArrayTestCase))
theSuite.addTest(common.make_suite(Empty2EArrayTestCase))
theSuite.addTest(common.make_suite(SlicesEArrayTestCase))
theSuite.addTest(common.make_suite(Slices2EArrayTestCase))
theSuite.addTest(common.make_suite(EllipsisEArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis2EArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis3EArrayTestCase))
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(Blosc2SlicesOptEArrayTestCase))
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(Blosc2InnerCrossChunkTestCase))
theSuite.addTest(common.make_suite(Blosc2InnerCrossChunkOptTestCase))
theSuite.addTest(common.make_suite(Blosc2PastLastChunkTestCase))
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(
SizeOnDiskInMemoryPropertyTestCase))
theSuite.addTest(common.make_suite(OffsetStrideTestCase))
theSuite.addTest(common.make_suite(Fletcher32TestCase))
theSuite.addTest(common.make_suite(AllFiltersTestCase))
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(TruncateOpenTestCase))
theSuite.addTest(common.make_suite(TruncateCloseTestCase))
theSuite.addTest(common.make_suite(ZeroSizedTestCase))
theSuite.addTest(common.make_suite(MDAtomNoReopen))
theSuite.addTest(common.make_suite(MDAtomReopen))
theSuite.addTest(common.make_suite(AccessClosedTestCase))
theSuite.addTest(common.make_suite(TestCreateEArrayArgs))
if common.heavy:
theSuite.addTest(common.make_suite(Slices3EArrayTestCase))
theSuite.addTest(common.make_suite(Slices4EArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis4EArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis5EArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis6EArrayTestCase))
theSuite.addTest(common.make_suite(Ellipsis7EArrayTestCase))
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')