import sys import numpy as np import tables as tb from tables.tests import common class C: c = (3, 4.5) class BasicTestCase(common.TempFileMixin, common.PyTablesTestCase): compress = 0 complib = "zlib" shuffle = 0 bitshuffle = 0 fletcher32 = 0 flavor = "numpy" def setUp(self): super().setUp() # Create an instance of an HDF5 Table self.rootgroup = self.h5file.root self.populateFile() self.h5file.close() def populateFile(self): group = self.rootgroup filters = tb.Filters( complevel=self.compress, complib=self.complib, shuffle=self.shuffle, bitshuffle=self.bitshuffle, fletcher32=self.fletcher32, ) vlarray = self.h5file.create_vlarray( group, "vlarray1", atom=tb.Int32Atom(), title="ragged array of ints", filters=filters, expectedrows=1000, ) vlarray.flavor = self.flavor # Fill it with 5 rows vlarray.append([1, 2]) if self.flavor == "numpy": vlarray.append(np.array([3, 4, 5], dtype="int32")) vlarray.append(np.array([], dtype="int32")) # Empty entry elif self.flavor == "python": vlarray.append((3, 4, 5)) vlarray.append(()) # Empty entry vlarray.append([6, 7, 8, 9]) vlarray.append([10, 11, 12, 13, 14]) def test00_attributes(self): self.h5file = tb.open_file(self.h5fname, "r") obj = self.h5file.get_node("/vlarray1") self.assertEqual(obj.flavor, self.flavor) self.assertEqual(obj.shape, (5,)) self.assertEqual(obj.ndim, 1) self.assertEqual(obj.nrows, 5) self.assertEqual(obj.atom.type, "int32") def test01_read(self): """Checking vlarray read.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_read..." % self.__class__.__name__) # Create an instance of an HDF5 Table self.h5file = tb.open_file(self.h5fname, "r") vlarray = self.h5file.get_node("/vlarray1") # Choose a small value for buffer size vlarray.nrowsinbuf = 3 # Read some rows row = vlarray.read(0)[0] row2 = vlarray.read(2)[0] if common.verbose: print("Flavor:", vlarray.flavor) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row) nrows = 5 self.assertEqual(nrows, vlarray.nrows) if self.flavor == "numpy": self.assertEqual(type(row), np.ndarray) self.assertTrue( common.allequal( row, np.array([1, 2], dtype="int32"), self.flavor ) ) self.assertTrue( common.allequal(row2, np.array([], dtype="int32"), self.flavor) ) elif self.flavor == "python": self.assertEqual(row, [1, 2]) self.assertEqual(row2, []) self.assertEqual(len(row), 2) # Check filters: if self.compress != vlarray.filters.complevel and common.verbose: print("Error in compress. Class:", self.__class__.__name__) print("self, vlarray:", self.compress, vlarray.filters.complevel) self.assertEqual(vlarray.filters.complevel, self.compress) if self.compress > 0 and tb.which_lib_version(self.complib): self.assertEqual(vlarray.filters.complib, self.complib) if self.shuffle != vlarray.filters.shuffle and common.verbose: print("Error in shuffle. Class:", self.__class__.__name__) print("self, vlarray:", self.shuffle, vlarray.filters.shuffle) self.assertEqual(self.shuffle, vlarray.filters.shuffle) if self.bitshuffle != vlarray.filters.bitshuffle and common.verbose: print("Error in shuffle. Class:", self.__class__.__name__) print( "self, vlarray:", self.bitshuffle, vlarray.filters.bitshuffle ) self.assertEqual(self.shuffle, vlarray.filters.shuffle) if self.fletcher32 != vlarray.filters.fletcher32 and common.verbose: print("Error in fletcher32. Class:", self.__class__.__name__) print( "self, vlarray:", self.fletcher32, vlarray.filters.fletcher32 ) self.assertEqual(self.fletcher32, vlarray.filters.fletcher32) def test02a_getitem(self): """Checking vlarray __getitem__ (slices)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02a_getitem..." % self.__class__.__name__) # Create an instance of an HDF5 Table self.h5file = tb.open_file(self.h5fname, "r") vlarray = self.h5file.get_node("/vlarray1") rows = [[1, 2], [3, 4, 5], [], [6, 7, 8, 9], [10, 11, 12, 13, 14]] slices = [ slice(None, None, None), slice(1, 1, 1), slice(30, None, None), slice(0, None, None), slice(3, None, 1), slice(3, None, 2), slice(None, 1, None), slice(None, 2, 1), slice(None, 30, 2), slice(None, None, 1), slice(None, None, 2), slice(None, None, 3), ] for slc in slices: # Read the rows in slc rows2 = vlarray[slc] rows1 = rows[slc] rows1f = [] if common.verbose: print("Flavor:", vlarray.flavor) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Original rows ==>", rows1) print("Rows read in vlarray ==>", rows2) if self.flavor == "numpy": for val in rows1: rows1f.append(np.array(val, dtype="int32")) for i in range(len(rows1f)): self.assertTrue( common.allequal(rows2[i], rows1f[i], self.flavor) ) elif self.flavor == "python": self.assertEqual(rows2, rows1) def test02b_getitem(self): """Checking vlarray __getitem__ (scalars)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02b_getitem..." % self.__class__.__name__) if self.flavor != "numpy": # This test is only valid for NumPy return # Create an instance of an HDF5 Table self.h5file = tb.open_file(self.h5fname, "r") vlarray = self.h5file.get_node("/vlarray1") # Get a numpy array of objects rows = np.array(vlarray[:], dtype=object) for slc in [0, np.array(1), 2, np.array([3]), [4]]: # Read the rows in slc rows2 = vlarray[slc] rows1 = rows[slc] if common.verbose: print("Flavor:", vlarray.flavor) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Original rows ==>", rows1) print("Rows read in vlarray ==>", rows2) for i in range(len(rows1)): self.assertTrue( common.allequal(rows2[i], rows1[i], self.flavor) ) def test03_append(self): """Checking vlarray append.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_append..." % self.__class__.__name__) # Create an instance of an HDF5 Table self.h5file = tb.open_file(self.h5fname, "a") vlarray = self.h5file.get_node("/vlarray1") # Append a new row vlarray.append([7, 8, 9, 10]) # Choose a small value for buffer size vlarray.nrowsinbuf = 3 # Read some rows: row1 = vlarray[0] row2 = vlarray[2] row3 = vlarray[-1] if common.verbose: print("Flavor:", vlarray.flavor) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row1) nrows = 6 self.assertEqual(nrows, vlarray.nrows) if self.flavor == "numpy": self.assertEqual(type(row1), type(np.array([1, 2]))) self.assertTrue( common.allequal( row1, np.array([1, 2], dtype="int32"), self.flavor ) ) self.assertTrue( common.allequal(row2, np.array([], dtype="int32"), self.flavor) ) self.assertTrue( common.allequal( row3, np.array([7, 8, 9, 10], dtype="int32"), self.flavor ) ) elif self.flavor == "python": self.assertEqual(row1, [1, 2]) self.assertEqual(row2, []) self.assertEqual(row3, [7, 8, 9, 10]) self.assertEqual(len(row3), 4) def test04_get_row_size(self): """Checking get_row_size method.""" self.h5file = tb.open_file(self.h5fname, "a") vlarray = self.h5file.get_node("/vlarray1") self.assertEqual(vlarray.get_row_size(0), 2 * vlarray.atom.size) self.assertEqual(vlarray.get_row_size(1), 3 * vlarray.atom.size) self.assertEqual(vlarray.get_row_size(2), 0 * vlarray.atom.size) self.assertEqual(vlarray.get_row_size(3), 4 * vlarray.atom.size) self.assertEqual(vlarray.get_row_size(4), 5 * vlarray.atom.size) class BasicNumPyTestCase(BasicTestCase): flavor = "numpy" class BasicPythonTestCase(BasicTestCase): flavor = "python" class ZlibComprTestCase(BasicTestCase): compress = 1 complib = "zlib" @common.unittest.skipIf( not common.blosc_avail, "BLOSC compression library not available" ) class BloscComprTestCase(BasicTestCase): compress = 9 shuffle = 0 complib = "blosc" @common.unittest.skipIf( not common.blosc_avail, "BLOSC compression library not available" ) class BloscShuffleComprTestCase(BasicTestCase): compress = 6 shuffle = 1 complib = "blosc" @common.unittest.skipIf( not common.blosc_avail, "BLOSC compression library not available" ) class BloscBitShuffleComprTestCase(BasicTestCase): compress = 9 bitshuffle = 1 complib = "blosc" @common.unittest.skipIf( not common.blosc_avail, "BLOSC compression library not available" ) class BloscBloscLZComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:blosclz" @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 BloscLZ4ComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:lz4" @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 BloscLZ4HCComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:lz4hc" @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 BloscSnappyComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:snappy" @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 BloscZlibComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:zlib" @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 BloscZstdComprTestCase(BasicTestCase): compress = 9 shuffle = 1 complib = "blosc:zstd" @common.unittest.skipIf( not common.lzo_avail, "LZO compression library not available" ) class LZOComprTestCase(BasicTestCase): compress = 1 complib = "lzo" @common.unittest.skipIf( not common.bzip2_avail, "BZIP2 compression library not available" ) class Bzip2ComprTestCase(BasicTestCase): compress = 1 complib = "bzip2" class ShuffleComprTestCase(BasicTestCase): compress = 1 shuffle = 1 class TypesTestCase(common.TempFileMixin, common.PyTablesTestCase): open_mode = "w" compress = 0 complib = "zlib" # Default compression library def test01_StringAtom(self): """Checking vlarray with NumPy string atoms ('numpy' flavor)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_StringAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "stringAtom", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "numpy" vlarray.append(np.array(["1", "12", "123", "1234", "12345"])) vlarray.append(np.array(["1", "12345"])) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) np.testing.assert_array_equal( row[0], np.array(["1", "12", "123", "123", "123"], "S") ) np.testing.assert_array_equal(row[1], np.array(["1", "123"], "S")) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 2) def test01a_StringAtom(self): """Checking vlarray with NumPy string atoms ('numpy' flavor, strided)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01a_StringAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "stringAtom", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "numpy" vlarray.append(np.array(["1", "12", "123", "1234", "12345"][::2])) vlarray.append(np.array(["1", "12345", "2", "321"])[::3]) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) np.testing.assert_array_equal( row[0], np.array(["1", "123", "123"], "S") ) np.testing.assert_array_equal(row[1], np.array(["1", "321"], "S")) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test01a_2_StringAtom(self): """Checking vlarray with NumPy string atoms (NumPy flavor, no conv)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test01a_2_StringAtom..." % self.__class__.__name__ ) vlarray = self.h5file.create_vlarray( "/", "stringAtom", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "numpy" vlarray.append(np.array(["1", "12", "123", "123"])) vlarray.append(np.array(["1", "2", "321"])) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) np.testing.assert_array_equal( row[0], np.array(["1", "12", "123", "123"], "S") ) np.testing.assert_array_equal(row[1], np.array(["1", "2", "321"], "S")) self.assertEqual(len(row[0]), 4) self.assertEqual(len(row[1]), 3) def test01b_StringAtom(self): """Checking vlarray with NumPy string atoms (python flavor)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01b_StringAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "stringAtom2", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "python" vlarray.append(["1", "12", "123", "1234", "12345"]) vlarray.append(["1", "12345"]) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing String flavor") print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], [b"1", b"12", b"123", b"123", b"123"]) self.assertEqual(row[1], [b"1", b"123"]) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 2) def test01c_StringAtom(self): """Checking updating vlarray with NumPy string atoms ('numpy' flavor)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01c_StringAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "stringAtom", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "numpy" vlarray.append(np.array(["1", "12", "123", "1234", "12345"])) vlarray.append(np.array(["1", "12345"])) # Modify the rows vlarray[0] = np.array(["1", "123", "12", "", "12345"]) vlarray[1] = np.array(["44", "4"]) # This should work as well if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([b"1", b"123", b"12", b"", b"123"]) ) ) self.assertTrue( common.allequal(row[1], np.array(["44", "4"], dtype="S3")) ) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 2) def test01d_StringAtom(self): """Checking updating vlarray with string atoms (String flavor)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01d_StringAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "stringAtom2", atom=tb.StringAtom(itemsize=3), title="Ragged array of strings", ) vlarray.flavor = "python" vlarray.append(["1", "12", "123", "1234", "12345"]) vlarray.append(["1", "12345"]) # Modify the rows vlarray[0] = ["1", "123", "12", "", "12345"] vlarray[1] = ["44", "4"] if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing String flavor") print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], [b"1", b"123", b"12", b"", b"123"]) self.assertEqual(row[1], [b"44", b"4"]) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 2) def test02_BoolAtom(self): """Checking vlarray with boolean atoms.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_BoolAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "BoolAtom", atom=tb.BoolAtom(), title="Ragged array of Booleans", ) vlarray.append([1, 0, 3]) vlarray.append([1, 0]) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([1, 0, 1], dtype="bool")) ) self.assertTrue( common.allequal(row[1], np.array([1, 0], dtype="bool")) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test02b_BoolAtom(self): """Checking setting vlarray with boolean atoms.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02b_BoolAtom..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "BoolAtom", atom=tb.BoolAtom(), title="Ragged array of Booleans", ) vlarray.append([1, 0, 3]) vlarray.append([1, 0]) # Modify the rows vlarray[0] = (0, 1, 3) vlarray[1] = (0, 1) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([0, 1, 1], dtype="bool")) ) self.assertTrue( common.allequal(row[1], np.array([0, 1], dtype="bool")) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test03_IntAtom(self): """Checking vlarray with integer atoms.""" ttypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "int64", # "UInt64", # Unavailable in some platforms ] if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_IntAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([1, 2, 3]) vlarray.append([1, 0]) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([1, 2, 3], dtype=atype)) ) self.assertTrue( common.allequal(row[1], np.array([1, 0], dtype=atype)) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test03a_IntAtom(self): """Checking vlarray with integer atoms (byteorder swapped)""" ttypes = { "int8": np.int8, "uint8": np.uint8, "int16": np.int16, "uint16": np.uint16, "int32": np.int32, "uint32": np.uint32, "int64": np.int64, # "uint64": np.int64, # Unavailable in some platforms } if common.verbose: print("\n", "-=" * 30) print("Running %s.test03a_IntAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]) ) a0 = np.array([1, 2, 3], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray.append(a0) a1 = np.array([1, 0], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray.append(a1) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([1, 2, 3], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal(row[1], np.array([1, 0], dtype=ttypes[atype])) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test03b_IntAtom(self): """Checking updating vlarray with integer atoms.""" ttypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "int64", # "UInt64", # Unavailable in some platforms ] if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_IntAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([1, 2, 3]) vlarray.append([1, 0]) # Modify rows vlarray[0] = (3, 2, 1) vlarray[1] = (0, 1) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([3, 2, 1], dtype=atype)) ) self.assertTrue( common.allequal(row[1], np.array([0, 1], dtype=atype)) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test03c_IntAtom(self): """Checking updating vlarray with integer atoms (byteorder swapped)""" ttypes = { "int8": np.int8, "uint8": np.uint8, "int16": np.int16, "uint16": np.uint16, "int32": np.int32, "uint32": np.uint32, "int64": np.int64, # "uint64": np.int64, # Unavailable in some platforms } if common.verbose: print("\n", "-=" * 30) print("Running %s.test03c_IntAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]) ) a0 = np.array([1, 2, 3], dtype=atype) vlarray.append(a0) a1 = np.array([1, 0], dtype=atype) vlarray.append(a1) # Modify rows a0 = np.array([3, 2, 1], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray[0] = a0 a1 = np.array([0, 1], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray[1] = a1 if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([3, 2, 1], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal(row[1], np.array([0, 1], dtype=ttypes[atype])) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test03d_IntAtom(self): """Checking updating vlarray with integer atoms (another byteorder)""" ttypes = { "int8": np.int8, "uint8": np.uint8, "int16": np.int16, "uint16": np.uint16, "int32": np.int32, "uint32": np.uint32, "int64": np.int64, # "uint64": np.int64, # Unavailable in some platforms } if common.verbose: print("\n", "-=" * 30) print("Running %s.test03d_IntAtom..." % self.__class__.__name__) byteorder = {"little": "big", "big": "little"}[sys.byteorder] for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]), byteorder=byteorder, ) a0 = np.array([1, 2, 3], dtype=atype) vlarray.append(a0) a1 = np.array([1, 0], dtype=atype) vlarray.append(a1) # Modify rows a0 = np.array([3, 2, 1], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray[0] = a0 a1 = np.array([0, 1], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray[1] = a1 if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) byteorder2 = tb.utils.byteorders[row[0].dtype.byteorder] if byteorder2 != "irrelevant": self.assertEqual( tb.utils.byteorders[row[0].dtype.byteorder], sys.byteorder ) self.assertEqual(vlarray.byteorder, byteorder) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([3, 2, 1], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal(row[1], np.array([0, 1], dtype=ttypes[atype])) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04_FloatAtom(self): """Checking vlarray with floating point atoms.""" ttypes = [ "float32", "float64", ] for name in ("float16", "float96", "float128"): atomname = name.capitalize() + "Atom" if hasattr(tb, atomname): ttypes.append(name) if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_FloatAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([1.3, 2.2, 3.3]) vlarray.append([5.96, 0.597]) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([1.3, 2.2, 3.3], atype)) ) self.assertTrue( common.allequal(row[1], np.array([5.96, 0.597], atype)) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04a_FloatAtom(self): """Checking vlarray with float atoms (byteorder swapped)""" ttypes = { "float32": np.float32, "float64": np.float64, } if hasattr(tb, "Float16Atom"): ttypes["float16"] = np.float16 if hasattr(tb, "Float96Atom"): ttypes["float96"] = np.float96 if hasattr(tb, "Float128Atom"): ttypes["float128"] = np.float128 if common.verbose: print("\n", "-=" * 30) print("Running %s.test04a_FloatAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]) ) a0 = np.array([1.3, 2.2, 3.3], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray.append(a0) a1 = np.array([5.96, 0.597], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray.append(a1) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([1.3, 2.2, 3.3], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal( row[1], np.array([5.96, 0.597], dtype=ttypes[atype]) ) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04b_FloatAtom(self): """Checking updating vlarray with floating point atoms.""" ttypes = [ "float32", "float64", ] for name in ("float16", "float96", "float128"): atomname = name.capitalize() + "Atom" if hasattr(tb, atomname): ttypes.append(name) if common.verbose: print("\n", "-=" * 30) print("Running %s.test04b_FloatAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([1.3, 2.2, 3.3]) vlarray.append([5.96, 0.597]) # Modifiy some rows vlarray[0] = (4.3, 2.2, 4.3) vlarray[1] = (1.123, 1.1e-3) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal(row[0], np.array([4.3, 2.2, 4.3], atype)) ) self.assertTrue( common.allequal(row[1], np.array([1.123, 1.1e-3], atype)) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04c_FloatAtom(self): """Checking updating vlarray with float atoms (byteorder swapped)""" ttypes = { "float32": np.float32, "float64": np.float64, } if hasattr(tb, "Float16Atom"): ttypes["float16"] = np.float16 if hasattr(tb, "Float96Atom"): ttypes["float96"] = np.float96 if hasattr(tb, "Float128Atom"): ttypes["float128"] = np.float128 if common.verbose: print("\n", "-=" * 30) print("Running %s.test04c_FloatAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]) ) a0 = np.array([1.3, 2.2, 3.3], dtype=atype) vlarray.append(a0) a1 = np.array([1, 0], dtype=atype) vlarray.append(a1) # Modify rows a0 = np.array([4.3, 2.2, 4.3], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray[0] = a0 a1 = np.array([1.123, 1.1e-3], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray[1] = a1 if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([4.3, 2.2, 4.3], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal( row[1], np.array([1.123, 1.1e-3], dtype=ttypes[atype]) ) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04d_FloatAtom(self): """Checking updating vlarray with float atoms (another byteorder)""" ttypes = { "float32": np.float32, "float64": np.float64, } if hasattr(tb, "Float16Atom"): ttypes["float16"] = np.float16 if hasattr(tb, "Float96Atom"): ttypes["float96"] = np.float96 if hasattr(tb, "Float128Atom"): ttypes["float128"] = np.float128 if common.verbose: print("\n", "-=" * 30) print("Running %s.test04d_FloatAtom..." % self.__class__.__name__) byteorder = {"little": "big", "big": "little"}[sys.byteorder] for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(ttypes[atype]), byteorder=byteorder, ) a0 = np.array([1.3, 2.2, 3.3], dtype=atype) vlarray.append(a0) a1 = np.array([1, 0], dtype=atype) vlarray.append(a1) # Modify rows a0 = np.array([4.3, 2.2, 4.3], dtype=atype) a0 = a0.byteswap() a0 = a0.view(a0.dtype.newbyteorder()) vlarray[0] = a0 a1 = np.array([1.123, 1.1e-3], dtype=atype) a1 = a1.byteswap() a1 = a1.view(a1.dtype.newbyteorder()) vlarray[1] = a1 if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.byteorder, byteorder) self.assertEqual( tb.utils.byteorders[row[0].dtype.byteorder], sys.byteorder ) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([4.3, 2.2, 4.3], dtype=ttypes[atype]) ) ) self.assertTrue( common.allequal( row[1], np.array([1.123, 1.1e-3], dtype=ttypes[atype]) ) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04_ComplexAtom(self): """Checking vlarray with numerical complex atoms.""" ttypes = [ "complex64", "complex128", ] if hasattr(tb, "Complex192Atom"): ttypes.append("complex192") if hasattr(tb, "Complex256Atom"): ttypes.append("complex256") if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_ComplexAtom..." % self.__class__.__name__) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([(1.3 + 0j), (0 + 2.2j), (3.3 + 3.3j)]) vlarray.append([(0 - 5.96j), (0.597 + 0j)]) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([(1.3 + 0j), (0 + 2.2j), (3.3 + 3.3j)], atype), ) ) self.assertTrue( common.allequal( row[1], np.array([(0 - 5.96j), (0.597 + 0j)], atype) ) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test04b_ComplexAtom(self): """Checking modifying vlarray with numerical complex atoms.""" ttypes = [ "complex64", "complex128", ] if hasattr(tb, "Complex192Atom"): ttypes.append("complex192") if hasattr(tb, "Complex256Atom"): ttypes.append("complex256") if common.verbose: print("\n", "-=" * 30) print( "Running %s.test04b_ComplexAtom..." % self.__class__.__name__ ) for atype in ttypes: vlarray = self.h5file.create_vlarray( "/", atype, atom=tb.Atom.from_sctype(atype) ) vlarray.append([(1.3 + 0j), (0 + 2.2j), (3.3 + 3.3j)]) vlarray.append([(0 - 5.96j), (0.597 + 0j)]) # Modify the rows vlarray[0] = ((1.4 + 0j), (0 + 4.2j), (3.3 + 4.3j)) vlarray[1] = ((4 - 5.96j), (0.597 + 4j)) if self.reopen: name = vlarray._v_pathname self._reopen(mode="a") vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([(1.4 + 0j), (0 + 4.2j), (3.3 + 4.3j)], atype), ) ) self.assertTrue( common.allequal( row[1], np.array([(4 - 5.96j), (0.597 + 4j)], atype) ) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 2) def test05_VLStringAtom(self): """Checking vlarray with variable length strings.""" # Skip the test if the default encoding has been mangled. if sys.getdefaultencoding() != "ascii": return if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05_VLStringAtom..." % self.__class__.__name__ ) vlarray = self.h5file.create_vlarray( "/", "VLStringAtom", atom=tb.VLStringAtom() ) vlarray.append(b"asd") vlarray.append(b"asd\xe4") vlarray.append(b"aaana") vlarray.append(b"") # Check for ticket #62. self.assertRaises(TypeError, vlarray.append, [b"foo", b"bar"]) # `VLStringAtom` makes no encoding assumptions. See ticket #51. self.assertRaises(UnicodeEncodeError, vlarray.append, "asd\xe4") if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 4) self.assertEqual(row[0], b"asd") self.assertEqual(row[1], b"asd\xe4") self.assertEqual(row[2], b"aaana") self.assertEqual(row[3], b"") self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) self.assertEqual(len(row[3]), 0) def test05b_VLStringAtom(self): """Checking updating vlarray with variable length strings.""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05b_VLStringAtom..." % self.__class__.__name__ ) vlarray = self.h5file.create_vlarray( "/", "VLStringAtom", atom=tb.VLStringAtom() ) vlarray.append(b"asd") vlarray.append(b"aaana") # Modify values vlarray[0] = b"as4" vlarray[1] = b"aaanc" self.assertRaises(ValueError, vlarray.__setitem__, 1, b"shrt") self.assertRaises(ValueError, vlarray.__setitem__, 1, b"toolong") if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", repr(row[0])) print("Second row in vlarray ==>", repr(row[1])) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], b"as4") self.assertEqual(row[1], b"aaanc") self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 5) def test06a_Object(self): """Checking vlarray with object atoms.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test06a_Object..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "Object", atom=tb.ObjectAtom() ) vlarray.append( [[1, 2, 3], "aaa", "aaa\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbd"] ) vlarray.append([3, 4, C()]) vlarray.append(42) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual( row[0], [[1, 2, 3], "aaa", "aaa\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbd"], ) list1 = list(row[1]) obj = list1.pop() self.assertEqual(list1, [3, 4]) self.assertEqual(obj.c, C().c) self.assertEqual(row[2], 42) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 3) self.assertRaises(TypeError, len, row[2]) def test06b_Object(self): """Checking updating vlarray with object atoms.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test06b_Object..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "Object", atom=tb.ObjectAtom() ) # When updating an object, this seems to change the number # of bytes that pickle.dumps generates # vlarray.append( # ([1,2,3], "aaa", "aaa\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbd")) vlarray.append(([1, 2, 3], "aaa", "\xef\xbf\xbd\xef\xbf\xbd4")) # vlarray.append([3,4, C()]) vlarray.append([3, 4, [24]]) # Modify the rows # vlarray[0] = ([1,2,4], "aa4", "aaa\xef\xbf\xbd\xef\xbf\xbd4") vlarray[0] = ([1, 2, 4], "aa4", "\xef\xbf\xbd\xef\xbf\xbd5") # vlarray[1] = (3,4, C()) vlarray[1] = [4, 4, [24]] if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 2) self.assertEqual( row[0], ([1, 2, 4], "aa4", "\xef\xbf\xbd\xef\xbf\xbd5") ) list1 = list(row[1]) obj = list1.pop() self.assertEqual(list1, [4, 4]) # self.assertEqual(obj.c, C().c) self.assertEqual(obj, [24]) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 3) def test06c_Object(self): """Checking vlarray with object atoms (numpy arrays as values)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test06c_Object..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "Object", atom=tb.ObjectAtom() ) vlarray.append(np.array([[1, 2], [0, 4]], "i4")) vlarray.append(np.array([0, 1, 2, 3], "i8")) vlarray.append(np.array(42, "i1")) if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertTrue( common.allequal(row[0], np.array([[1, 2], [0, 4]], "i4")) ) self.assertTrue(common.allequal(row[1], np.array([0, 1, 2, 3], "i8"))) self.assertTrue(common.allequal(row[2], np.array(42, "i1"))) def test06d_Object(self): """Checking updating vlarray with object atoms (numpy arrays)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test06d_Object..." % self.__class__.__name__) vlarray = self.h5file.create_vlarray( "/", "Object", atom=tb.ObjectAtom() ) vlarray.append(np.array([[1, 2], [0, 4]], "i4")) vlarray.append(np.array([0, 1, 2, 3], "i8")) vlarray.append(np.array(42, "i1")) # Modify the rows. Since PyTables 2.2.1 we use a binary # pickle for arrays and ObjectAtoms, so the next should take # the same space than the above. vlarray[0] = np.array([[1, 0], [0, 4]], "i4") vlarray[1] = np.array([0, 1, 0, 3], "i8") vlarray[2] = np.array(22, "i1") if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertTrue( common.allequal(row[0], np.array([[1, 0], [0, 4]], "i4")) ) self.assertTrue(common.allequal(row[1], np.array([0, 1, 0, 3], "i8"))) self.assertTrue(common.allequal(row[2], np.array(22, "i1"))) def test07_VLUnicodeAtom(self): """Checking vlarray with variable length Unicode strings.""" # Skip the test if the default encoding has been mangled. if sys.getdefaultencoding() != "ascii": return if common.verbose: print("\n", "-=" * 30) print( "Running %s.test07_VLUnicodeAtom..." % self.__class__.__name__ ) vlarray = self.h5file.create_vlarray( "/", "VLUnicodeAtom", atom=tb.VLUnicodeAtom() ) vlarray.append("asd") vlarray.append("asd\u0140") vlarray.append("aaana") vlarray.append("") # Check for ticket #62. self.assertRaises(TypeError, vlarray.append, ["foo", "bar"]) # `VLUnicodeAtom` makes no encoding assumptions. self.assertRaises(UnicodeDecodeError, vlarray.append, "asd\xe4") if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 4) self.assertEqual(row[0], "asd") self.assertEqual(row[1], "asd\u0140") self.assertEqual(row[2], "aaana") self.assertEqual(row[3], "") self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) self.assertEqual(len(row[3]), 0) def test07b_VLUnicodeAtom(self): """Checking updating vlarray with variable length Unicode strings.""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test07b_VLUnicodeAtom..." % self.__class__.__name__ ) vlarray = self.h5file.create_vlarray( "/", "VLUnicodeAtom", atom=tb.VLUnicodeAtom() ) vlarray.append("asd") vlarray.append("aaan\xe4") # Modify values vlarray[0] = "as\xe4" vlarray[1] = "aaan\u0140" self.assertRaises(ValueError, vlarray.__setitem__, 1, "shrt") self.assertRaises(ValueError, vlarray.__setitem__, 1, "toolong") if self.reopen: name = vlarray._v_pathname self._reopen() vlarray = self.h5file.get_node(name) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", repr(row[0])) print("Second row in vlarray ==>", repr(row[1])) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], "as\xe4") self.assertEqual(row[1], "aaan\u0140") self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 5) class TypesReopenTestCase(TypesTestCase): title = "Reopen" reopen = True class TypesNoReopenTestCase(TypesTestCase): title = "No reopen" reopen = False class MDTypesTestCase(common.TempFileMixin, common.PyTablesTestCase): open_mode = "w" compress = 0 complib = "zlib" # Default compression library def setUp(self): super().setUp() self.rootgroup = self.h5file.root def test01_StringAtom(self): """Checking vlarray with MD NumPy string atoms.""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_StringAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "stringAtom", tb.StringAtom(itemsize=3, shape=(2,)), "Ragged array of strings", ) vlarray.append([["123", "45"], ["45", "123"]]) vlarray.append([["s", "abc"], ["abc", "f"], ["s", "ab"], ["ab", "f"]]) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) np.testing.assert_array_equal( row[0], np.array([["123", "45"], ["45", "123"]], "S") ) np.testing.assert_array_equal( row[1], np.array( [["s", "abc"], ["abc", "f"], ["s", "ab"], ["ab", "f"]], "S" ), ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 4) def test01b_StringAtom(self): """Checking vlarray with MD NumPy string atoms ('python' flavor)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test01b_StringAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "stringAtom", tb.StringAtom(itemsize=3, shape=(2,)), "Ragged array of strings", ) vlarray.flavor = "python" vlarray.append([["123", "45"], ["45", "123"]]) vlarray.append([["s", "abc"], ["abc", "f"], ["s", "ab"], ["ab", "f"]]) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], [[b"123", b"45"], [b"45", b"123"]]) self.assertEqual( row[1], [[b"s", b"abc"], [b"abc", b"f"], [b"s", b"ab"], [b"ab", b"f"]], ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 4) def test01c_StringAtom(self): """Checking vlarray with MD NumPy string atoms (with offset)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test01c_StringAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "stringAtom", tb.StringAtom(itemsize=3, shape=(2,)), "Ragged array of strings", ) vlarray.flavor = "python" a = np.array([["a", "b"], ["123", "45"], ["45", "123"]], dtype="S3") vlarray.append(a[1:]) a = np.array( [ ["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"], ["s", "ab"], ["ab", "f"], ] ) vlarray.append(a[2:]) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], [[b"123", b"45"], [b"45", b"123"]]) self.assertEqual( row[1], [[b"s", b"abc"], [b"abc", b"f"], [b"s", b"ab"], [b"ab", b"f"]], ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 4) def test01d_StringAtom(self): """Checking vlarray with MD NumPy string atoms (with stride)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test01d_StringAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "stringAtom", tb.StringAtom(itemsize=3, shape=(2,)), "Ragged array of strings", ) vlarray.flavor = "python" a = np.array([["a", "b"], ["123", "45"], ["45", "123"]], dtype="S3") vlarray.append(a[1::2]) a = np.array( [ ["s", "a"], ["ab", "f"], ["s", "abc"], ["abc", "f"], ["s", "ab"], ["ab", "f"], ] ) vlarray.append(a[::3]) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertEqual(row[0], [[b"123", b"45"]]) self.assertEqual(row[1], [[b"s", b"a"], [b"abc", b"f"]]) self.assertEqual(len(row[0]), 1) self.assertEqual(len(row[1]), 2) def test02_BoolAtom(self): """Checking vlarray with MD boolean atoms.""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_BoolAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "BoolAtom", tb.BoolAtom(shape=(3,)), "Ragged array of Booleans", ) vlarray.append([(1, 0, 3), (1, 1, 1), (0, 0, 0)]) vlarray.append([(1, 0, 0)]) # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([[1, 0, 1], [1, 1, 1], [0, 0, 0]], dtype="bool"), ) ) self.assertTrue( common.allequal(row[1], np.array([[1, 0, 0]], dtype="bool")) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 1) def test02b_BoolAtom(self): """Checking vlarray with MD boolean atoms (with offset)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test02b_BoolAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "BoolAtom", tb.BoolAtom(shape=(3,)), "Ragged array of Booleans", ) a = np.array( [(0, 0, 0), (1, 0, 3), (1, 1, 1), (0, 0, 0)], dtype="bool" ) vlarray.append(a[1:]) # Create an offset a = np.array([(1, 1, 1), (1, 0, 0)], dtype="bool") vlarray.append(a[1:]) # Create an offset # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([[1, 0, 1], [1, 1, 1], [0, 0, 0]], dtype="bool"), ) ) self.assertTrue( common.allequal(row[1], np.array([[1, 0, 0]], dtype="bool")) ) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 1) def test02c_BoolAtom(self): """Checking vlarray with MD boolean atoms (with strides)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test02c_BoolAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "BoolAtom", tb.BoolAtom(shape=(3,)), "Ragged array of Booleans", ) a = np.array( [(0, 0, 0), (1, 0, 3), (1, 1, 1), (0, 0, 0)], dtype="bool" ) vlarray.append(a[1::2]) # Create an strided array a = np.array([(1, 1, 1), (1, 0, 0), (0, 0, 0)], dtype="bool") vlarray.append(a[::2]) # Create an strided array # Read all the rows: row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([[1, 0, 1], [0, 0, 0]], dtype="bool") ) ) self.assertTrue( common.allequal( row[1], np.array([[1, 1, 1], [0, 0, 0]], dtype="bool") ) ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 2) def test03_IntAtom(self): """Checking vlarray with MD integer atoms.""" ttypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "int64", # "UInt64", # Unavailable in some platforms ] root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_IntAtom..." % self.__class__.__name__) # Create an string atom for atype in ttypes: vlarray = self.h5file.create_vlarray( root, atype, atom=tb.Atom.from_sctype(atype, (2, 3)) ) vlarray.append([np.ones((2, 3), atype), np.zeros((2, 3), atype)]) vlarray.append([np.ones((2, 3), atype) * 100]) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", repr(row[1])) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array([np.ones((2, 3)), np.zeros((2, 3))], atype), ) ) self.assertTrue( common.allequal( row[1], np.array([np.ones((2, 3)) * 100], atype) ) ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 1) def test04_FloatAtom(self): """Checking vlarray with MD floating point atoms.""" ttypes = [ "float32", "float64", "complex64", "complex128", ] for name in ("float16", "float96", "float128"): atomname = name.capitalize() + "Atom" if hasattr(tb, atomname): ttypes.append(name) for itemsize in (192, 256): atomname = "Complex%dAtom" % itemsize if hasattr(tb, atomname): ttypes.append("complex%d" % (itemsize)) root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_FloatAtom..." % self.__class__.__name__) # Create an string atom for atype in ttypes: vlarray = self.h5file.create_vlarray( root, atype, atom=tb.Atom.from_sctype(atype, (5, 2, 6)) ) vlarray.append( [np.ones((5, 2, 6), atype) * 1.3, np.zeros((5, 2, 6), atype)] ) vlarray.append([np.ones((5, 2, 6), atype) * 2.0e4]) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing type:", atype) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, 2) self.assertTrue( common.allequal( row[0], np.array( [np.ones((5, 2, 6)) * 1.3, np.zeros((5, 2, 6))], atype ), ) ) self.assertTrue( common.allequal( row[1], np.array([np.ones((5, 2, 6)) * 2.0e4], atype) ) ) self.assertEqual(len(row[0]), 2) self.assertEqual(len(row[1]), 1) class MDTypesNumPyTestCase(MDTypesTestCase): title = "MDTypes" class AppendShapeTestCase(common.TempFileMixin, common.PyTablesTestCase): open_mode = "w" def setUp(self): super().setUp() self.rootgroup = self.h5file.root def test00_difinputs(self): """Checking vlarray.append() with different inputs.""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test00_difinputs..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Int32Atom(), "Ragged array of ints" ) vlarray.flavor = "python" # Check different ways to input # All of the next should lead to the same rows vlarray.append((1, 2, 3)) # a tuple vlarray.append([1, 2, 3]) # a unique list vlarray.append(np.array([1, 2, 3], dtype="int32")) # and array if self.close: if common.verbose: print("(closing file version)") self._reopen() vlarray = self.h5file.root.vlarray # Read all the vlarray row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual(row[0], [1, 2, 3]) self.assertEqual(row[1], [1, 2, 3]) self.assertEqual(row[2], [1, 2, 3]) def test01_toomanydims(self): """Checking vlarray.append() with too many dimensions.""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_toomanydims..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.StringAtom(itemsize=3), "Ragged array of strings", ) # Adding an array with one dimensionality more than allowed with self.assertRaises(ValueError): vlarray.append([["123", "456", "3"]]) if self.close: if common.verbose: print("(closing file version)") self._reopen() vlarray = self.h5file.root.vlarray # Read all the rows (there should be none) row = vlarray.read() if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) self.assertEqual(vlarray.nrows, 0) def test02_zerodims(self): """Checking vlarray.append() with a zero-dimensional array""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_zerodims..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Int32Atom(), "Ragged array of ints" ) vlarray.append(np.zeros(dtype="int32", shape=(6, 0))) if self.close: if common.verbose: print("(closing file version)") self._reopen() vlarray = self.h5file.root.vlarray # Read the only row in vlarray row = vlarray.read(0)[0] if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", repr(row)) self.assertEqual(vlarray.nrows, 1) self.assertTrue( common.allequal(row, np.zeros(dtype="int32", shape=(0,))) ) self.assertEqual(len(row), 0) def test03a_cast(self): """Checking vlarray.append() with a casted array (upgrading case)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test03a_cast..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Int32Atom(), "Ragged array of ints" ) # This type has to be upgraded vlarray.append(np.array([1, 2], dtype="int16")) if self.close: if common.verbose: print("(closing file version)") self._reopen() vlarray = self.h5file.root.vlarray # Read the only row in vlarray row = vlarray.read(0)[0] if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", repr(row)) self.assertEqual(vlarray.nrows, 1) self.assertTrue(common.allequal(row, np.array([1, 2], dtype="int32"))) self.assertEqual(len(row), 2) def test03b_cast(self): """Checking vlarray.append() with a casted array (downgrading case)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test03b_cast..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Int32Atom(), "Ragged array of ints" ) # This type has to be downcasted vlarray.append(np.array([1, 2], dtype="float64")) if self.close: if common.verbose: print("(closing file version)") self._reopen() vlarray = self.h5file.root.vlarray # Read the only row in vlarray row = vlarray.read(0)[0] if common.verbose: print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", repr(row)) self.assertEqual(vlarray.nrows, 1) self.assertTrue(common.allequal(row, np.array([1, 2], dtype="int32"))) self.assertEqual(len(row), 2) class OpenAppendShapeTestCase(AppendShapeTestCase): close = 0 class CloseAppendShapeTestCase(AppendShapeTestCase): close = 1 class FlavorTestCase(common.TempFileMixin, common.PyTablesTestCase): open_mode = "w" compress = 0 complib = "zlib" # Default compression library def setUp(self): super().setUp() self.rootgroup = self.h5file.root def test01a_EmptyVLArray(self): """Checking empty vlarrays with different flavors (closing the file)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print( "Running %s.test01_EmptyVLArray..." % self.__class__.__name__ ) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Atom.from_kind("int", itemsize=4) ) vlarray.flavor = self.flavor self.h5file.close() self.h5file = tb.open_file(self.h5fname, "r") # Read all the rows (it should be empty): vlarray = self.h5file.root.vlarray row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row, repr(row)) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) # Check that the object read is effectively empty self.assertEqual(vlarray.nrows, 0) self.assertEqual(row, []) def test01b_EmptyVLArray(self): """Checking empty vlarrays with different flavors (no closing file)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print( "Running %s.test01_EmptyVLArray..." % self.__class__.__name__ ) # Create an string atom vlarray = self.h5file.create_vlarray( root, "vlarray", tb.Atom.from_kind("int", itemsize=4) ) vlarray.flavor = self.flavor # Read all the rows (it should be empty): row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) # Check that the object read is effectively empty self.assertEqual(vlarray.nrows, 0) self.assertEqual(row, []) def test02_BooleanAtom(self): """Checking vlarray with different flavors (boolean versions)""" root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_BoolAtom..." % self.__class__.__name__) # Create an string atom vlarray = self.h5file.create_vlarray(root, "Bool", tb.BoolAtom()) vlarray.flavor = self.flavor vlarray.append([1, 2, 3]) vlarray.append(()) # Empty row vlarray.append([100, 0]) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 0) self.assertEqual(len(row[2]), 2) if self.flavor == "python": arr1 = [1, 1, 1] arr2 = [] arr3 = [1, 0] elif self.flavor == "numpy": arr1 = np.array([1, 1, 1], dtype="bool") arr2 = np.array([], dtype="bool") arr3 = np.array([1, 0], dtype="bool") if self.flavor == "numpy": self.assertTrue(common.allequal(row[0], arr1, self.flavor)) self.assertTrue(common.allequal(row[1], arr2, self.flavor)) self.assertTrue(common.allequal(row[1], arr2, self.flavor)) else: # 'python' flavor self.assertEqual(row[0], arr1) self.assertEqual(row[1], arr2) self.assertEqual(row[2], arr3) def test03_IntAtom(self): """Checking vlarray with different flavors (integer versions)""" ttypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "int64", # Not checked because some platforms does not support it # "UInt64", ] root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_IntAtom..." % self.__class__.__name__) # Create an string atom for atype in ttypes: vlarray = self.h5file.create_vlarray( root, atype, tb.Atom.from_sctype(atype) ) vlarray.flavor = self.flavor vlarray.append([1, 2, 3]) vlarray.append(()) vlarray.append([100, 0]) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 0) self.assertEqual(len(row[2]), 2) if self.flavor == "python": arr1 = [1, 2, 3] arr2 = [] arr3 = [100, 0] elif self.flavor == "numpy": arr1 = np.array([1, 2, 3], dtype=atype) arr2 = np.array([], dtype=atype) arr3 = np.array([100, 0], dtype=atype) if self.flavor == "numpy": self.assertTrue(common.allequal(row[0], arr1, self.flavor)) self.assertTrue(common.allequal(row[1], arr2, self.flavor)) self.assertTrue(common.allequal(row[2], arr3, self.flavor)) else: # "python" flavor self.assertEqual(row[0], arr1) self.assertEqual(row[1], arr2) self.assertEqual(row[2], arr3) def test03b_IntAtom(self): """Checking vlarray flavors (integer versions and closed file)""" ttypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "int64", # Not checked because some platforms does not support it # "UInt64", ] root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_IntAtom..." % self.__class__.__name__) # Create an string atom for atype in ttypes: vlarray = self.h5file.create_vlarray( root, atype, tb.Atom.from_sctype(atype) ) vlarray.flavor = self.flavor vlarray.append([1, 2, 3]) vlarray.append(()) vlarray.append([100, 0]) self._reopen(mode="a") # open in "a"ppend mode root = self.h5file.root # Very important! vlarray = self.h5file.get_node(root, str(atype)) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 0) self.assertEqual(len(row[2]), 2) if self.flavor == "python": arr1 = [1, 2, 3] arr2 = [] arr3 = [100, 0] elif self.flavor == "numpy": arr1 = np.array([1, 2, 3], dtype=atype) arr2 = np.array([], dtype=atype) arr3 = np.array([100, 0], dtype=atype) if self.flavor == "numpy": self.assertTrue(common.allequal(row[0], arr1, self.flavor)) self.assertTrue(common.allequal(row[1], arr2, self.flavor)) self.assertTrue(common.allequal(row[2], arr3, self.flavor)) else: # Tuple or List flavors self.assertEqual(row[0], arr1) self.assertEqual(row[1], arr2) self.assertEqual(row[2], arr3) def test04_FloatAtom(self): """Checking vlarray with different flavors (floating point versions)""" ttypes = [ "float32", "float64", "complex64", "complex128", ] for name in ("float16", "float96", "float128"): atomname = name.capitalize() + "Atom" if hasattr(tb, atomname): ttypes.append(name) for itemsize in (192, 256): atomname = "Complex%dAtom" % itemsize if hasattr(tb, atomname): ttypes.append("complex%d" % (itemsize)) root = self.rootgroup if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_FloatAtom..." % self.__class__.__name__) # Create an string atom for atype in ttypes: vlarray = self.h5file.create_vlarray( root, atype, tb.Atom.from_sctype(atype) ) vlarray.flavor = self.flavor vlarray.append([1.3, 2.2, 3.3]) vlarray.append(()) vlarray.append([5.96, 0.597]) # Read all the rows: row = vlarray.read() if common.verbose: print("Testing flavor:", self.flavor) print("Object read:", row) print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) self.assertEqual(vlarray.nrows, 3) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 0) self.assertEqual(len(row[2]), 2) if self.flavor == "python": arr1 = list(np.array([1.3, 2.2, 3.3], atype)) arr2 = list(np.array([], atype)) arr3 = list(np.array([5.96, 0.597], atype)) elif self.flavor == "numpy": arr1 = np.array([1.3, 2.2, 3.3], dtype=atype) arr2 = np.array([], dtype=atype) arr3 = np.array([5.96, 0.597], dtype=atype) if self.flavor == "numpy": self.assertTrue(common.allequal(row[0], arr1, self.flavor)) self.assertTrue(common.allequal(row[1], arr2, self.flavor)) self.assertTrue(common.allequal(row[2], arr3, self.flavor)) else: # Tuple or List flavors self.assertEqual(row[0], arr1) self.assertEqual(row[1], arr2) self.assertEqual(row[2], arr3) class NumPyFlavorTestCase(FlavorTestCase): flavor = "numpy" class PythonFlavorTestCase(FlavorTestCase): flavor = "python" class ReadRangeTestCase(common.TempFileMixin, common.PyTablesTestCase): nrows = 100 mode = "w" compress = 0 complib = "zlib" # Default compression library def setUp(self): super().setUp() self.rootgroup = self.h5file.root self.populateFile() self._reopen() def populateFile(self): group = self.rootgroup filters = tb.Filters(complevel=self.compress, complib=self.complib) vlarray = self.h5file.create_vlarray( group, "vlarray", tb.Int32Atom(), "ragged array if ints", filters=filters, expectedrows=1000, ) # Fill it with 100 rows with variable length for i in range(self.nrows): vlarray.append(list(range(i))) def test01_start(self): """Checking reads with only a start value""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Read some rows: row = [] row.append(vlarray.read(0)[0]) row.append(vlarray.read(10)[0]) row.append(vlarray.read(99)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue(common.allequal(row[0], np.arange(0, dtype="int32"))) self.assertTrue(common.allequal(row[1], np.arange(10, dtype="int32"))) self.assertTrue(common.allequal(row[2], np.arange(99, dtype="int32"))) def test01b_start(self): """Checking reads with only a start value in a slice""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01b_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Read some rows: row = [] row.append(vlarray[0]) row.append(vlarray[10]) row.append(vlarray[99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue(common.allequal(row[0], np.arange(0, dtype="int32"))) self.assertTrue(common.allequal(row[1], np.arange(10, dtype="int32"))) self.assertTrue(common.allequal(row[2], np.arange(99, dtype="int32"))) def test01np_start(self): """Checking reads with only a start value in a slice (numpy indexes)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01np_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Read some rows: row = [] row.append(vlarray[np.int8(0)]) row.append(vlarray[np.int32(10)]) row.append(vlarray[np.int64(99)]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue(common.allequal(row[0], np.arange(0, dtype="int32"))) self.assertTrue(common.allequal(row[1], np.arange(10, dtype="int32"))) self.assertTrue(common.allequal(row[2], np.arange(99, dtype="int32"))) def test02_stop(self): """Checking reads with only a stop value""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_stop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray.read(stop=1)) row.append(vlarray.read(stop=10)) row.append(vlarray.read(stop=99)) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 1) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue( common.allequal(row[0][0], np.arange(0, dtype="int32")) ) for x in range(10): self.assertTrue( common.allequal(row[1][x], np.arange(x, dtype="int32")) ) for x in range(99): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test02b_stop(self): """Checking reads with only a stop value in a slice""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02b_stop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[:1]) row.append(vlarray[:10]) row.append(vlarray[:99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 1) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) for x in range(1): self.assertTrue( common.allequal(row[0][x], np.arange(0, dtype="int32")) ) for x in range(10): self.assertTrue( common.allequal(row[1][x], np.arange(x, dtype="int32")) ) for x in range(99): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test03_startstop(self): """Checking reads with a start and stop values""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_startstop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray.read(0, 10)) row.append(vlarray.read(5, 15)) row.append(vlarray.read(0, 100)) # read all the array if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 10) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 100) for x in range(0, 10): self.assertTrue( common.allequal(row[0][x], np.arange(x, dtype="int32")) ) for x in range(5, 15): self.assertTrue( common.allequal(row[1][x - 5], np.arange(x, dtype="int32")) ) for x in range(0, 100): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test03b_startstop(self): """Checking reads with a start and stop values in slices""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03b_startstop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[0:10]) row.append(vlarray[5:15]) row.append(vlarray[:]) # read all the array if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 10) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 100) for x in range(0, 10): self.assertTrue( common.allequal(row[0][x], np.arange(x, dtype="int32")) ) for x in range(5, 15): self.assertTrue( common.allequal(row[1][x - 5], np.arange(x, dtype="int32")) ) for x in range(0, 100): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test04_startstopstep(self): """Checking reads with a start, stop & step values""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test04_startstopstep..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray.read(0, 10, 2)) row.append(vlarray.read(5, 15, 3)) row.append(vlarray.read(0, 100, 20)) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test04np_startstopstep(self): """Checking reads with a start, stop & step values (numpy indices)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test04np_startstopstep..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray.read(np.int8(0), np.int8(10), np.int8(2))) row.append(vlarray.read(np.int8(5), np.int8(15), np.int8(3))) row.append(vlarray.read(np.int8(0), np.int8(100), np.int8(20))) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test04b_slices(self): """Checking reads with start, stop & step values in slices""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test04b_slices..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[0:10:2]) row.append(vlarray[5:15:3]) row.append(vlarray[0:100:20]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test04bnp_slices(self): """Checking reads with start, stop & step values in slices. (numpy indices) """ if common.verbose: print("\n", "-=" * 30) print("Running %s.test04bnp_slices..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[np.int16(0) : np.int16(10) : np.int32(2)]) row.append(vlarray[np.int16(5) : np.int16(15) : np.int64(3)]) row.append(vlarray[np.uint16(0) : np.int32(100) : np.int8(20)]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test05_out_of_range(self): """Checking out of range reads""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05_out_of_range..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) with self.assertRaises(IndexError): row = vlarray.read(1000)[0] print("row-->", row) class GetItemRangeTestCase(common.TempFileMixin, common.PyTablesTestCase): nrows = 100 open_mode = "w" compress = 0 complib = "zlib" # Default compression library def setUp(self): super().setUp() self.rootgroup = self.h5file.root self.populateFile() self._reopen() def populateFile(self): group = self.rootgroup filters = tb.Filters(complevel=self.compress, complib=self.complib) vlarray = self.h5file.create_vlarray( group, "vlarray", tb.Int32Atom(), "ragged array if ints", filters=filters, expectedrows=1000, ) # Fill it with 100 rows with variable length for i in range(self.nrows): vlarray.append(list(range(i))) def test01_start(self): """Checking reads with only a start value""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Read some rows: row = [] row.append(vlarray[0]) # rank-0 array should work as a regular index (see #303) row.append(vlarray[np.array(10)]) row.append(vlarray[99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue(common.allequal(row[0], np.arange(0, dtype="int32"))) self.assertTrue(common.allequal(row[1], np.arange(10, dtype="int32"))) self.assertTrue(common.allequal(row[2], np.arange(99, dtype="int32"))) def test01b_start(self): """Checking reads with only a start value in a slice""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01b_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Read some rows: row = [] row.append(vlarray[0]) row.append(vlarray[10]) row.append(vlarray[99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue(common.allequal(row[0], np.arange(0, dtype="int32"))) self.assertTrue(common.allequal(row[1], np.arange(10, dtype="int32"))) self.assertTrue(common.allequal(row[2], np.arange(99, dtype="int32"))) def test02_stop(self): """Checking reads with only a stop value""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_stop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[:1]) row.append(vlarray[:10]) row.append(vlarray[:99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("First row in vlarray ==>", row[0]) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 1) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue( common.allequal(row[0][0], np.arange(0, dtype="int32")) ) for x in range(10): self.assertTrue( common.allequal(row[1][x], np.arange(x, dtype="int32")) ) for x in range(99): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test02b_stop(self): """Checking reads with only a stop value in a slice""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02b_stop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[:1]) row.append(vlarray[:10]) row.append(vlarray[:99]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 1) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) for x in range(1): self.assertTrue( common.allequal(row[0][x], np.arange(0, dtype="int32")) ) for x in range(10): self.assertTrue( common.allequal(row[1][x], np.arange(x, dtype="int32")) ) for x in range(99): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test03_startstop(self): """Checking reads with a start and stop values""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_startstop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[0:10]) row.append(vlarray[5:15]) row.append(vlarray[0:100]) # read all the array if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 10) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 100) for x in range(0, 10): self.assertTrue( common.allequal(row[0][x], np.arange(x, dtype="int32")) ) for x in range(5, 15): self.assertTrue( common.allequal(row[1][x - 5], np.arange(x, dtype="int32")) ) for x in range(0, 100): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test03b_startstop(self): """Checking reads with a start and stop values in slices""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03b_startstop..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[0:10]) row.append(vlarray[5:15]) row.append(vlarray[:]) # read all the array if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 10) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 100) for x in range(0, 10): self.assertTrue( common.allequal(row[0][x], np.arange(x, dtype="int32")) ) for x in range(5, 15): self.assertTrue( common.allequal(row[1][x - 5], np.arange(x, dtype="int32")) ) for x in range(0, 100): self.assertTrue( common.allequal(row[2][x], np.arange(x, dtype="int32")) ) def test04_slices(self): """Checking reads with a start, stop & step values""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_slices..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[0:10:2]) row.append(vlarray[5:15:3]) row.append(vlarray[0:100:20]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test04bnp_slices(self): """Checking reads with start, stop & step values (numpy indices)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test04np_slices..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Choose a small value for buffer size vlarray._nrowsinbuf = 3 # Read some rows: row = [] row.append(vlarray[np.int8(0) : np.int8(10) : np.int8(2)]) row.append(vlarray[np.int8(5) : np.int8(15) : np.int8(3)]) row.append(vlarray[np.int8(0) : np.int8(100) : np.int8(20)]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 5) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) for x in range(0, 10, 2): self.assertTrue( common.allequal(row[0][x // 2], np.arange(x, dtype="int32")) ) for x in range(5, 15, 3): self.assertTrue( common.allequal( row[1][(x - 5) // 3], np.arange(x, dtype="int32") ) ) for x in range(0, 100, 20): self.assertTrue( common.allequal(row[2][x // 20], np.arange(x, dtype="int32")) ) def test05_out_of_range(self): """Checking out of range reads""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05_out_of_range..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) with self.assertRaises(IndexError): row = vlarray[1000] print("row-->", row) def test05np_out_of_range(self): """Checking out of range reads (numpy indexes)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05np_out_of_range..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) with self.assertRaises(IndexError): row = vlarray[np.int32(1000)] print("row-->", row) class SetRangeTestCase(common.TempFileMixin, common.PyTablesTestCase): nrows = 100 open_mode = "w" compress = 0 complib = "zlib" # Default compression library def setUp(self): super().setUp() self.rootgroup = self.h5file.root self.populateFile() self._reopen(mode="a") def populateFile(self): group = self.rootgroup filters = tb.Filters(complevel=self.compress, complib=self.complib) vlarray = self.h5file.create_vlarray( group, "vlarray", tb.Int32Atom(), "ragged array if ints", filters=filters, expectedrows=1000, ) # Fill it with 100 rows with variable length for i in range(self.nrows): vlarray.append(list(range(i))) def test01_start(self): """Checking updates that modifies a complete row""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[0] = vlarray[0] * 2 + 3 vlarray[10] = vlarray[10] * 2 + 3 vlarray[99] = vlarray[99] * 2 + 3 # Read some rows: row = [] row.append(vlarray.read(0)[0]) row.append(vlarray.read(10)[0]) row.append(vlarray.read(99)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue( common.allequal(row[0], np.arange(0, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(10, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[2], np.arange(99, dtype="int32") * 2 + 3) ) def test01np_start(self): """Checking updates that modifies a complete row""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01np_start..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[np.int8(0)] = vlarray[np.int16(0)] * 2 + 3 vlarray[np.int8(10)] = vlarray[np.int8(10)] * 2 + 3 vlarray[np.int32(99)] = vlarray[np.int64(99)] * 2 + 3 # Read some rows: row = [] row.append(vlarray.read(np.int8(0))[0]) row.append(vlarray.read(np.int8(10))[0]) row.append(vlarray.read(np.int8(99))[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 99) self.assertTrue( common.allequal(row[0], np.arange(0, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(10, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[2], np.arange(99, dtype="int32") * 2 + 3) ) def test02_partial(self): """Checking updates with only a part of a row""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_partial..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[0] = vlarray[0] * 2 + 3 vlarray[10] = vlarray[10] * 2 + 3 vlarray[96] = vlarray[99][3:] * 2 + 3 # Read some rows: row = [] row.append(vlarray.read(0)[0]) row.append(vlarray.read(10)[0]) row.append(vlarray.read(96)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 96) self.assertTrue( common.allequal(row[0], np.arange(0, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(10, dtype="int32") * 2 + 3) ) a = np.arange(3, 99, dtype="int32") a = a * 2 + 3 self.assertTrue(common.allequal(row[2], a)) def test03a_several_rows(self): """Checking updating several rows at once (slice style)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test03a_several_rows..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[3:6] = ( vlarray[3] * 2 + 3, vlarray[4] * 2 + 3, vlarray[5] * 2 + 3, ) # Read some rows: row = [] row.append(vlarray.read(3)[0]) row.append(vlarray.read(4)[0]) row.append(vlarray.read(5)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 3) self.assertEqual(len(row[1]), 4) self.assertEqual(len(row[2]), 5) self.assertTrue( common.allequal(row[0], np.arange(3, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(4, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[2], np.arange(5, dtype="int32") * 2 + 3) ) def test03b_several_rows(self): """Checking updating several rows at once (list style)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test03b_several_rows..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[[0, 10, 96]] = ( vlarray[0] * 2 + 3, vlarray[10] * 2 + 3, vlarray[96] * 2 + 3, ) # Read some rows: row = [] row.append(vlarray.read(0)[0]) row.append(vlarray.read(10)[0]) row.append(vlarray.read(96)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 96) self.assertTrue( common.allequal(row[0], np.arange(0, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(10, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[2], np.arange(96, dtype="int32") * 2 + 3) ) def test03c_several_rows(self): """Checking updating several rows at once (NumPy's where style)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test03c_several_rows..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray # Modify some rows: vlarray[(np.array([0, 10, 96]),)] = ( vlarray[0] * 2 + 3, vlarray[10] * 2 + 3, vlarray[96] * 2 + 3, ) # Read some rows: row = [] row.append(vlarray.read(0)[0]) row.append(vlarray.read(10)[0]) row.append(vlarray.read(96)[0]) if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) print("Second row in vlarray ==>", row[1]) self.assertEqual(vlarray.nrows, self.nrows) self.assertEqual(len(row[0]), 0) self.assertEqual(len(row[1]), 10) self.assertEqual(len(row[2]), 96) self.assertTrue( common.allequal(row[0], np.arange(0, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[1], np.arange(10, dtype="int32") * 2 + 3) ) self.assertTrue( common.allequal(row[2], np.arange(96, dtype="int32") * 2 + 3) ) def test04_out_of_range(self): """Checking out of range updates (first index)""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test04_out_of_range..." % self.__class__.__name__ ) vlarray = self.h5file.root.vlarray if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) with self.assertRaises(IndexError): vlarray[1000] = [1] def test05_value_error(self): """Checking out value errors""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test05_value_error..." % self.__class__.__name__) vlarray = self.h5file.root.vlarray if common.verbose: print("Nrows in", vlarray._v_pathname, ":", vlarray.nrows) with self.assertRaises(ValueError): vlarray[10] = [1] * 100 class CopyTestCase(common.TempFileMixin, common.PyTablesTestCase): close = True def test01a_copy(self): """Checking VLArray.copy() method.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01a_copy..." % self.__class__.__name__) # Create an Vlarray arr = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", arr, "title array1" ) array1.flavor = "python" array1.append([[2, 3]]) array1.append(()) # an empty row array1.append([[3, 457], [2, 4]]) 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-->", repr(array1)) print("array2-->", repr(array2)) print("array1[:]-->", repr(array1.read())) print("array2[:]-->", repr(array2.read())) print("attrs array1-->", repr(array1.attrs)) print("attrs array2-->", repr(array2.attrs)) # Check that all the 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.flavor, array2.flavor) self.assertEqual(array1.atom.dtype, array2.atom.dtype) self.assertEqual(repr(array1.atom), repr(array2.atom)) self.assertEqual(array1.title, array2.title) def test01b_copy(self): """Checking VLArray.copy() method (Pseudo-atom case)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01b_copy..." % self.__class__.__name__) # Create an Vlarray arr = tb.VLStringAtom() array1 = self.h5file.create_vlarray( self.h5file.root, "array1", arr, "title array1" ) array1.flavor = "python" array1.append(b"a string") array1.append(b"") # an empty row array1.append(b"another string") 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-->", repr(array1)) print("array2-->", repr(array2)) print("array1[:]-->", repr(array1.read())) print("array2[:]-->", repr(array2.read())) print("attrs array1-->", repr(array1.attrs)) print("attrs array2-->", repr(array2.attrs)) # Check that all the 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.flavor, array2.flavor) self.assertEqual(array1.atom.type, array2.atom.type) self.assertEqual(repr(array1.atom), repr(array2.atom)) self.assertEqual(array1.title, array2.title) def test02_copy(self): """Checking VLArray.copy() method (where specified)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test02_copy..." % self.__class__.__name__) # Create an VLArray arr = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", arr, "title array1" ) array1.flavor = "python" array1.append([[2, 3]]) array1.append(()) # an empty row array1.append([[3, 457], [2, 4]]) 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-->", repr(array1)) print("array2-->", repr(array2)) 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 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.flavor, array2.flavor) self.assertEqual(array1.atom.dtype, array2.atom.dtype) self.assertEqual(repr(array1.atom), repr(array1.atom)) self.assertEqual(array1.title, array2.title) def test03_copy(self): """Checking VLArray.copy() method ('python' flavor)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test03_copy..." % self.__class__.__name__) # Create an VLArray atom = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", atom, title="title array1" ) array1.flavor = "python" array1.append(((2, 3),)) array1.append(()) # an empty row array1.append(((3, 457), (2, 4))) 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)) # Assert other properties in array self.assertEqual(array1.nrows, array2.nrows) self.assertEqual(array1.shape, array2.shape) self.assertEqual(array1.flavor, array2.flavor) # Very important here self.assertEqual(array1.atom.dtype, array2.atom.dtype) self.assertEqual(repr(array1.atom), repr(array1.atom)) self.assertEqual(array1.title, array2.title) def test04_copy(self): """Checking VLArray.copy() method (checking title copying)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test04_copy..." % self.__class__.__name__) # Create an VLArray atom = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", atom=atom, title="title array1" ) array1.append(((2, 3),)) array1.append(()) # an empty row array1.append(((3, 457), (2, 4))) # 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 VLArray.copy() method (user attributes copied)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test05_copy..." % self.__class__.__name__) # Create an Array atom = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", atom=atom, title="title array1" ) array1.append(((2, 3),)) array1.append(()) # an empty row array1.append(((3, 457), (2, 4))) # 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 notest05b_copy(self): """Checking VLArray.copy() method (user attributes not copied)""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test05b_copy..." % self.__class__.__name__) # Create an VLArray atom = tb.Int16Atom(shape=2) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", atom=atom, title="title array1" ) array1.append(((2, 3),)) array1.append(()) # an empty row array1.append(((3, 457), (2, 4))) # 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(array2.attrs.attr1, None) self.assertEqual(array2.attrs.attr2, None) class CloseCopyTestCase(CopyTestCase): close = 1 class OpenCopyTestCase(CopyTestCase): close = 0 class CopyIndexTestCase(common.TempFileMixin, common.PyTablesTestCase): def test01_index(self): """Checking VLArray.copy() method with indexes.""" if common.verbose: print("\n", "-=" * 30) print("Running %s.test01_index..." % self.__class__.__name__) # Create an VLArray atom = tb.Int32Atom(shape=(2,)) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", atom, "t array1" ) array1.flavor = "python" # The next creates 20 rows of variable length r = [] for row in range(20): r.append([[row, row + 1]]) array1.append([row, row + 1]) if self.close: if common.verbose: print("(closing file version)") self._reopen(mode="a") array1 = self.h5file.root.array1 # Copy to another array array2 = array1.copy( "/", "array2", start=self.start, stop=self.stop, step=self.step ) r2 = r[self.start : self.stop : self.step] if common.verbose: print("r2-->", r2) print("array2-->", array2[:]) print("attrs array1-->", repr(array1.attrs)) print("attrs array2-->", repr(array2.attrs)) print("nrows in array2-->", array2.nrows) print("and it should be-->", len(r2)) # Check that all the elements are equal self.assertEqual(r2, array2[:]) # Assert the number of rows in array self.assertEqual(len(r2), array2.nrows) class CopyIndex1TestCase(CopyIndexTestCase): close = 0 start = 0 stop = 7 step = 1 class CopyIndex2TestCase(CopyIndexTestCase): close = 1 start = 0 stop = -1 step = 1 class CopyIndex3TestCase(CopyIndexTestCase): close = 0 start = 1 stop = 7 step = 1 class CopyIndex4TestCase(CopyIndexTestCase): close = 1 start = 0 stop = 6 step = 1 class CopyIndex5TestCase(CopyIndexTestCase): close = 0 start = 3 stop = 7 step = 1 class CopyIndex6TestCase(CopyIndexTestCase): close = 1 start = 3 stop = 6 step = 2 class CopyIndex7TestCase(CopyIndexTestCase): close = 0 start = 0 stop = 7 step = 10 class CopyIndex8TestCase(CopyIndexTestCase): close = 1 start = 6 stop = -1 # Negative values means starting from the end step = 1 class CopyIndex9TestCase(CopyIndexTestCase): close = 0 start = 3 stop = 4 step = 1 class CopyIndex10TestCase(CopyIndexTestCase): close = 1 start = 3 stop = 4 step = 2 class CopyIndex11TestCase(CopyIndexTestCase): close = 0 start = -3 stop = -1 step = 2 class CopyIndex12TestCase(CopyIndexTestCase): close = 1 start = -1 # Should point to the last element stop = None # None should mean the last element (including it) step = 1 class ChunkshapeTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() atom = tb.Int32Atom(shape=(2,)) self.h5file.create_vlarray( "/", "vlarray", atom=atom, title="t array1", chunkshape=13 ) def test00(self): """Test setting the chunkshape in a table (no reopen).""" vla = self.h5file.root.vlarray if common.verbose: print("chunkshape-->", vla.chunkshape) self.assertEqual(vla.chunkshape, (13,)) def test01(self): """Test setting the chunkshape in a table (reopen).""" self.h5file.close() self.h5file = tb.open_file(self.h5fname, "r") vla = self.h5file.root.vlarray if common.verbose: print("chunkshape-->", vla.chunkshape) self.assertEqual(vla.chunkshape, (13,)) class VLUEndianTestCase(common.PyTablesTestCase): def setUp(self): super().setUp() self.h5fname = common.test_filename("vlunicode_endian.h5") self.h5file = tb.open_file(self.h5fname) def tearDown(self): self.h5file.close() super().tearDown() def test(self): """Accessing ``vlunicode`` data of a different endianness.""" bedata = self.h5file.root.vlunicode_big[0] ledata = self.h5file.root.vlunicode_little[0] self.assertEqual(bedata, "para\u0140lel") self.assertEqual(ledata, "para\u0140lel") class TruncateTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() # Create an VLArray arr = tb.Int16Atom(dflt=3) array1 = self.h5file.create_vlarray( self.h5file.root, "array1", arr, "title array1" ) # Add a couple of rows array1.append(np.array([456, 2], dtype="int16")) array1.append(np.array([3], dtype="int16")) def test00_truncate(self): """Checking VLArray.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.assertEqual(array1.nrows, 0) self.assertEqual(array1[:], []) def test01_truncate(self): """Checking VLArray.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.assertEqual(array1.nrows, 1) self.assertTrue( common.allequal(array1[0], np.array([456, 2], dtype="int16")) ) def test02_truncate(self): """Checking VLArray.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.assertEqual(array1.nrows, 2) self.assertTrue( common.allequal(array1[0], np.array([456, 2], dtype="int16")) ) self.assertTrue( common.allequal(array1[1], np.array([3], dtype="int16")) ) def test03_truncate(self): """Checking VLArray.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[0], np.array([456, 2], dtype="int16")) ) self.assertTrue( common.allequal(array1[1], np.array([3], dtype="int16")) ) # Check that the added rows are empty self.assertTrue( common.allequal(array1[2], np.array([], dtype="int16")) ) self.assertTrue( common.allequal(array1[3], np.array([], dtype="int16")) ) class TruncateOpenTestCase(TruncateTestCase): close = 0 class TruncateCloseTestCase(TruncateTestCase): close = 1 class PointSelectionTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() # The next are valid selections for both NumPy and PyTables self.working_keyset = [ [], # empty list [2], # single-entry list [0, 2], # list [0, -2], # negative values ([0, 2],), # tuple of list np.array([], dtype="i4"), # empty array np.array([1], dtype="i4"), # single-entry array np.array([True, False, True]), # array of bools ] # The next are invalid selections for VLArrays self.not_working_keyset = [ [1, 2, 100], # coordinate 100 > len(vlarray) ([True, False, True],), # tuple of bools ] # Create a sample array arr1 = np.array([5, 6], dtype="i4") arr2 = np.array([5, 6, 7], dtype="i4") arr3 = np.array([5, 6, 9, 8], dtype="i4") self.nparr = np.array([arr1, arr2, arr3], dtype="object") # Create the VLArray self.vlarr = self.h5file.create_vlarray( self.h5file.root, "vlarray", tb.Int32Atom() ) self.vlarr.append(arr1) self.vlarr.append(arr2) self.vlarr.append(arr3) def test01a_read(self): """Test for point-selections (read, boolean keys).""" nparr = self.nparr vlarr = self.vlarr for key in self.working_keyset: if common.verbose: print("Selection to test:", repr(key)) a = nparr[key].tolist() b = vlarr[key] # if common.verbose: # print "NumPy selection:", a, type(a) # print "PyTables selection:", b, type(b) self.assertEqual( repr(a), repr(b), "NumPy array and PyTables selections does not match.", ) def test01b_read(self): """Test for point-selections (not working selections, read).""" vlarr = self.vlarr for key in self.not_working_keyset: if common.verbose: print("Selection to test:", key) self.assertRaises(IndexError, vlarr.__getitem__, key) class SizeInMemoryPropertyTestCase( common.TempFileMixin, common.PyTablesTestCase ): def create_array(self, atom, complevel): filters = tb.Filters(complevel=complevel, complib="blosc") self.array = self.h5file.create_vlarray( "/", "vlarray", atom=atom, filters=filters ) def test_zero_length(self): atom = tb.Int32Atom() complevel = 0 self.create_array(atom, complevel) self.assertEqual(self.array.size_in_memory, 0) def int_tests(self, complevel, flavor): atom = tb.Int32Atom() self.create_array(atom, complevel) self.array.flavor = flavor expected_size = 0 for i in range(10): row = np.arange((i + 1) * 10, dtype="i4") self.array.append(row) expected_size += row.nbytes return expected_size def test_numpy_int_numpy_flavor(self): complevel = 0 flavor = "numpy" expected_size = self.int_tests(complevel, flavor) self.assertEqual(self.array.size_in_memory, expected_size) # compression will have no effect, since this is uncompressed size def test_numpy_int_numpy_flavor_compressed(self): complevel = 1 flavor = "numpy" expected_size = self.int_tests(complevel, flavor) self.assertEqual(self.array.size_in_memory, expected_size) # flavor will have no effect on what's stored in HDF5 file def test_numpy_int_python_flavor(self): complevel = 0 flavor = "python" expected_size = self.int_tests(complevel, flavor) self.assertEqual(self.array.size_in_memory, expected_size) # this relies on knowledge of the implementation, so it's not # a great test def test_object_atom(self): atom = tb.ObjectAtom() complevel = 0 self.create_array(atom, complevel) obj = [1, 2, 3] for i in range(10): self.array.append(obj) pickle_array = atom.toarray(obj) expected_size = 10 * pickle_array.nbytes self.assertEqual(self.array.size_in_memory, expected_size) class SizeOnDiskPropertyTestCase( common.TempFileMixin, common.PyTablesTestCase ): def create_array(self, atom, complevel): filters = tb.Filters(complevel=complevel, complib="blosc") self.h5file.create_vlarray("/", "vlarray", atom, filters=filters) self.array = self.h5file.get_node("/", "vlarray") def test_not_implemented(self): atom = tb.IntAtom() complevel = 0 self.create_array(atom, complevel) self.assertRaises( NotImplementedError, getattr, self.array, "size_on_disk" ) class AccessClosedTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() self.array = self.h5file.create_vlarray( self.h5file.root, "array", atom=tb.StringAtom(8) ) self.array.append([str(i) for i in range(5, 5005, 100)]) 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, "xxxxxxxxx") class TestCreateVLArrayArgs(common.TempFileMixin, common.PyTablesTestCase): obj = np.array([1, 2, 3]) where = "/" name = "vlarray" atom = tb.Atom.from_dtype(obj.dtype) title = "title" filters = None expectedrows = None chunkshape = None byteorder = None createparents = False def test_positional_args_01(self): self.h5file.create_vlarray( self.where, self.name, self.atom, self.title, self.filters, self.expectedrows, ) self.h5file.close() self.h5file = tb.open_file(self.h5fname) ptarr = self.h5file.get_node(self.where, self.name) self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (0,)) self.assertEqual(ptarr.nrows, 0) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) def test_positional_args_02(self): ptarr = self.h5file.create_vlarray( self.where, self.name, self.atom, self.title, self.filters, self.expectedrows, ) ptarr.append(self.obj) self.h5file.close() self.h5file = tb.open_file(self.h5fname) ptarr = self.h5file.get_node(self.where, self.name) nparr = ptarr.read()[0] self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (1,)) self.assertEqual(ptarr[0].shape, self.obj.shape) self.assertEqual(ptarr.nrows, 1) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) self.assertTrue(common.allequal(self.obj, nparr)) def test_positional_args_obj(self): self.h5file.create_vlarray( self.where, self.name, None, self.title, self.filters, self.expectedrows, 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()[0] self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (1,)) self.assertEqual(ptarr[0].shape, self.obj.shape) self.assertEqual(ptarr.nrows, 1) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) self.assertTrue(common.allequal(self.obj, nparr)) def test_kwargs_obj(self): self.h5file.create_vlarray( self.where, self.name, title=self.title, 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()[0] self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (1,)) self.assertEqual(ptarr[0].shape, self.obj.shape) self.assertEqual(ptarr.nrows, 1) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) self.assertTrue(common.allequal(self.obj, nparr)) def test_kwargs_atom_01(self): ptarr = self.h5file.create_vlarray( self.where, self.name, title=self.title, atom=self.atom ) ptarr.append(self.obj) self.h5file.close() self.h5file = tb.open_file(self.h5fname) ptarr = self.h5file.get_node(self.where, self.name) nparr = ptarr.read()[0] self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (1,)) self.assertEqual(ptarr[0].shape, self.obj.shape) self.assertEqual(ptarr.nrows, 1) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) self.assertTrue(common.allequal(self.obj, nparr)) def test_kwargs_atom_02(self): ptarr = self.h5file.create_vlarray( self.where, self.name, title=self.title, atom=self.atom ) # ptarr.append(self.obj) self.h5file.close() self.h5file = tb.open_file(self.h5fname) ptarr = self.h5file.get_node(self.where, self.name) self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (0,)) self.assertEqual(ptarr.nrows, 0) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) def test_kwargs_obj_atom(self): ptarr = self.h5file.create_vlarray( self.where, self.name, title=self.title, 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()[0] self.assertEqual(ptarr.title, self.title) self.assertEqual(ptarr.shape, (1,)) self.assertEqual(ptarr[0].shape, self.obj.shape) self.assertEqual(ptarr.nrows, 1) self.assertEqual(ptarr.atom, self.atom) self.assertEqual(ptarr.atom.dtype, self.atom.dtype) 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_vlarray, self.where, self.name, title=self.title, obj=self.obj, atom=atom, ) def suite(): theSuite = common.unittest.TestSuite() niter = 1 for n in range(niter): theSuite.addTest(common.make_suite(BasicNumPyTestCase)) theSuite.addTest(common.make_suite(BasicPythonTestCase)) theSuite.addTest(common.make_suite(ZlibComprTestCase)) theSuite.addTest(common.make_suite(BloscComprTestCase)) theSuite.addTest(common.make_suite(BloscShuffleComprTestCase)) theSuite.addTest(common.make_suite(BloscBitShuffleComprTestCase)) theSuite.addTest(common.make_suite(BloscBloscLZComprTestCase)) theSuite.addTest(common.make_suite(BloscLZ4ComprTestCase)) theSuite.addTest(common.make_suite(BloscLZ4HCComprTestCase)) theSuite.addTest(common.make_suite(BloscSnappyComprTestCase)) theSuite.addTest(common.make_suite(BloscZlibComprTestCase)) theSuite.addTest(common.make_suite(BloscZstdComprTestCase)) theSuite.addTest(common.make_suite(LZOComprTestCase)) theSuite.addTest(common.make_suite(Bzip2ComprTestCase)) theSuite.addTest(common.make_suite(TypesReopenTestCase)) theSuite.addTest(common.make_suite(TypesNoReopenTestCase)) theSuite.addTest(common.make_suite(MDTypesNumPyTestCase)) theSuite.addTest(common.make_suite(OpenAppendShapeTestCase)) theSuite.addTest(common.make_suite(CloseAppendShapeTestCase)) theSuite.addTest(common.make_suite(PythonFlavorTestCase)) theSuite.addTest(common.make_suite(NumPyFlavorTestCase)) theSuite.addTest(common.make_suite(ReadRangeTestCase)) theSuite.addTest(common.make_suite(GetItemRangeTestCase)) theSuite.addTest(common.make_suite(SetRangeTestCase)) theSuite.addTest(common.make_suite(ShuffleComprTestCase)) 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(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(ChunkshapeTestCase)) theSuite.addTest(common.make_suite(VLUEndianTestCase)) theSuite.addTest(common.make_suite(TruncateOpenTestCase)) theSuite.addTest(common.make_suite(TruncateCloseTestCase)) theSuite.addTest(common.make_suite(PointSelectionTestCase)) theSuite.addTest(common.make_suite(SizeInMemoryPropertyTestCase)) theSuite.addTest(common.make_suite(SizeOnDiskPropertyTestCase)) theSuite.addTest(common.make_suite(AccessClosedTestCase)) theSuite.addTest(common.make_suite(TestCreateVLArrayArgs)) return theSuite if __name__ == "__main__": common.parse_argv(sys.argv) common.print_versions() common.unittest.main(defaultTest="suite")