File: C:/Users/fred/anaconda3/Lib/site-packages/tables/tests/test_tablesMD.py
import sys
import numpy as np
import tables as tb
from tables.tests import common
# It is important that columns are ordered according to their names
# to ease the comparison with structured arrays.
# Test Record class
class Record(tb.IsDescription):
var0 = tb.StringCol(itemsize=4, dflt=b"", shape=2) # 4-char str array
var1 = tb.StringCol(itemsize=4, dflt=[b"abcd", b"efgh"], shape=(2, 2))
var1_ = tb.IntCol(dflt=((1, 1),), shape=2) # integer array
var2 = tb.IntCol(dflt=((1, 1), (1, 1)), shape=(2, 2)) # integer array
var3 = tb.Int16Col(dflt=2) # short integer
var4 = tb.FloatCol(dflt=3.1) # double (double-precision)
var5 = tb.Float32Col(dflt=4.2) # float (single-precision)
var6 = tb.UInt16Col(dflt=5) # unsigned short integer
var7 = tb.StringCol(itemsize=1, dflt=b"e") # 1-character String
# Dictionary definition
RecordDescriptionDict = {
'var0': tb.StringCol(itemsize=4, dflt=b"", shape=2), # 4-char str array
'var1': tb.StringCol(itemsize=4, dflt=[b"abcd", b"efgh"], shape=(2, 2)),
# 'var0': StringCol(itemsize=4, shape=2), # 4-character String
# 'var1': StringCol(itemsize=4, shape=(2,2)), # 4-character String
'var1_': tb.IntCol(shape=2), # integer array
'var2': tb.IntCol(shape=(2, 2)), # integer array
'var3': tb.Int16Col(), # short integer
'var4': tb.FloatCol(), # double (double-precision)
'var5': tb.Float32Col(), # float (single-precision)
'var6': tb.Int16Col(), # unsigned short integer
'var7': tb.StringCol(itemsize=1), # 1-character String
}
# Record class with numpy dtypes (mixed shapes is checked here)
class RecordDT(tb.IsDescription):
var0 = tb.Col.from_dtype(np.dtype("2S4"), dflt=b"") # shape in dtype
var1 = tb.Col.from_dtype(np.dtype(("S4", (
2, 2))), dflt=[b"abcd", b"efgh"]) # shape is a mix
var1_ = tb.Col.from_dtype(
np.dtype("2i4"), dflt=((1, 1),)) # shape in dtype
var2 = tb.Col.from_sctype("i4", shape=(
2, 2), dflt=((1, 1), (1, 1))) # shape is a mix
var3 = tb.Col.from_dtype(np.dtype("i2"), dflt=2)
var4 = tb.Col.from_dtype(np.dtype("2f8"), dflt=3.1)
var5 = tb.Col.from_dtype(np.dtype("f4"), dflt=4.2)
var6 = tb.Col.from_dtype(np.dtype("()u2"), dflt=5)
var7 = tb.Col.from_dtype(np.dtype("S1"), dflt=b"e") # no shape
class BasicTestCase(common.TempFileMixin, common.PyTablesTestCase):
# file = "test.h5"
open_mode = "w"
title = "This is the table title"
expectedrows = 100
appendrows = 20
compress = 0
complib = "zlib" # Default compression library
record = Record
recarrayinit = 0
maxshort = 1 << 15
def setUp(self):
super().setUp()
# Create an instance of an HDF5 Table
self.rootgroup = self.h5file.root
self.populateFile()
self.h5file.close()
def initRecArray(self):
record = self.recordtemplate
row = record[0]
buflist = []
# Fill the recarray
for i in range(self.expectedrows):
tmplist = []
# Both forms (list or chararray) work
var0 = ['%04d' % (self.expectedrows - i)] * 2
tmplist.append(var0)
var1 = [['%04d' % (self.expectedrows - i)] * 2] * 2
tmplist.append(var1)
var1_ = (i, 1)
tmplist.append(var1_)
var2 = ((i, 1), (1, 1)) # *-*
tmplist.append(var2)
var3 = i % self.maxshort
tmplist.append(var3)
if isinstance(row['var4'], np.ndarray):
tmplist.append([float(i), float(i * i)])
else:
tmplist.append(float(i))
if isinstance(row['var5'], np.ndarray):
tmplist.append(np.array((float(i),)*4))
else:
tmplist.append(float(i))
# var6 will be like var3 but byteswaped
tmplist.append(((var3 >> 8) & 0xff) + ((var3 << 8) & 0xff00))
var7 = var1[0][0][-1]
tmplist.append(var7)
buflist.append(tuple(tmplist))
self.record = np.rec.array(buflist, dtype=record.dtype,
shape=self.expectedrows)
def populateFile(self):
group = self.rootgroup
if self.recarrayinit:
# Initialize a starting buffer, if any
self.initRecArray()
for j in range(3):
# Create a table
filters = tb.Filters(complevel=self.compress, complib=self.complib)
if j < 2:
byteorder = sys.byteorder
else:
# table2 will be byteswapped
byteorder = {"little": "big", "big": "little"}[sys.byteorder]
table = self.h5file.create_table(group, 'table'+str(j),
self.record,
title=self.title,
filters=filters,
expectedrows=self.expectedrows,
byteorder=byteorder)
if not self.recarrayinit:
# Get the row object associated with the new table
row = table.row
# Fill the table
for i in range(self.expectedrows):
s = '%04d' % (self.expectedrows - i)
row['var0'] = s.encode('ascii')
row['var1'] = s.encode('ascii')
row['var7'] = s[-1].encode('ascii')
row['var1_'] = (i, 1)
row['var2'] = ((i, 1), (1, 1)) # *-*
row['var3'] = i % self.maxshort
if isinstance(row['var4'], np.ndarray):
row['var4'] = [float(i), float(i * i)]
else:
row['var4'] = float(i)
if isinstance(row['var5'], np.ndarray):
row['var5'] = np.array((float(i),)*4)
else:
row['var5'] = float(i)
# var6 will be like var3 but byteswaped
row['var6'] = (((row['var3'] >> 8) & 0xff) +
((row['var3'] << 8) & 0xff00))
row.append()
# Flush the buffer for this table
table.flush()
# Create a new group (descendant of group)
group2 = self.h5file.create_group(group, 'group'+str(j))
# Iterate over this new group (group2)
group = group2
def test00_description(self):
"""Checking table description and descriptive fields."""
self.h5file = tb.open_file(self.h5fname)
tbl = self.h5file.get_node('/table0')
desc = tbl.description
if isinstance(self.record, dict):
columns = self.record
elif isinstance(self.record, np.ndarray):
descr, _ = tb.description.descr_from_dtype(self.record.dtype)
columns = descr._v_colobjects
elif isinstance(self.record, np.dtype):
descr, _ = tb.description.descr_from_dtype(self.record)
columns = descr._v_colobjects
else:
# This is an ordinary description.
columns = self.record.columns
# Check table and description attributes at the same time.
# These checks are only valid for non-nested tb.
# Column names.
expectedNames = ['var0', 'var1', 'var1_', 'var2', 'var3', 'var4',
'var5', 'var6', 'var7']
self.assertEqual(expectedNames, list(tbl.colnames))
self.assertEqual(expectedNames, list(desc._v_names))
# Column types.
expectedTypes = [columns[colname].dtype
for colname in expectedNames]
self.assertEqual(expectedTypes,
[tbl.coldtypes[v] for v in expectedNames])
self.assertEqual(expectedTypes,
[desc._v_dtypes[v] for v in expectedNames])
# Column string types.
expectedTypes = [columns[colname].type
for colname in expectedNames]
self.assertEqual(expectedTypes,
[tbl.coltypes[v] for v in expectedNames])
self.assertEqual(expectedTypes,
[desc._v_types[v] for v in expectedNames])
# Column defaults.
for v in expectedNames:
if common.verbose:
print("dflt-->", columns[v].dflt)
print("coldflts-->", tbl.coldflts[v])
print("desc.dflts-->", desc._v_dflts[v])
self.assertTrue(common.areArraysEqual(tbl.coldflts[v],
columns[v].dflt))
self.assertTrue(common.areArraysEqual(desc._v_dflts[v],
columns[v].dflt))
# Column path names.
self.assertEqual(expectedNames, list(desc._v_pathnames))
# Column objects.
for colName in expectedNames:
expectedCol = columns[colName]
col = desc._v_colobjects[colName]
self.assertEqual(expectedCol.dtype, col.dtype)
self.assertEqual(expectedCol.type, col.type)
def test01_readTable(self):
"""Checking table read and cuts."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_readTable..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
table = self.h5file.get_node("/table0")
# Choose a small value for buffer size
table.nrowsinbuf = 3
# Read the records and select those with "var2" file less than 20
result = [r['var2'][0][0] for r in table.iterrows()
if r['var2'][0][0] < 20]
if common.verbose:
print("Table:", repr(table))
print("Nrows in", table._v_pathname, ":", table.nrows)
print("Last record in table ==>", table[-1])
print("Total selected records in table ==> ", len(result))
nrows = self.expectedrows - 1
r = [r for r in table.iterrows() if r['var2'][0][0] < 20][-1]
self.assertEqual((
r['var0'][0],
r['var1'][0][0],
r['var1_'][0],
r['var2'][0][0],
r['var7']
), (b"0001", b"0001", nrows, nrows, b"1"))
if isinstance(r['var5'], np.ndarray):
self.assertTrue(common.allequal(
r['var5'], np.array((nrows,)*4, np.float32)))
else:
self.assertEqual(r['var5'], float(nrows))
self.assertEqual(len(result), 20)
def test01b_readTable(self):
"""Checking table read and cuts (multidimensional columns case)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01b_readTable..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
table = self.h5file.get_node("/table0")
# Choose a small value for buffer size
table.nrowsinbuf = 3
# Read the records and select those with "var2" file less than 20
result1 = [r['var5'] for r in table.iterrows() if r['var2'][0][0] < 20]
if common.verbose:
print("Nrows in", table._v_pathname, ":", table.nrows)
print("Last record in table ==>", table[-1])
print("Total selected records in table ==> ", len(result1))
nrows = table.nrows
result2 = [r for r in table.iterrows() if r['var2'][0][0] < 20][-1]
if isinstance(result2['var5'], np.ndarray):
self.assertTrue(common.allequal(
result1[0], np.array((float(0),) * 4, np.float32)))
self.assertTrue(common.allequal(
result1[1], np.array((float(1),) * 4, np.float32)))
self.assertTrue(common.allequal(
result1[2], np.array((float(2),) * 4, np.float32)))
self.assertTrue(common.allequal(
result1[3], np.array((float(3),) * 4, np.float32)))
self.assertTrue(common.allequal(
result1[10], np.array((float(10),) * 4, np.float32)))
self.assertTrue(common.allequal(
result2['var5'], np.array((float(nrows - 1),) * 4, np.float32)
))
else:
self.assertEqual(result2['var5'], float(nrows - 1))
self.assertEqual(len(result1), 20)
# Read the records and select those with "var2" file less than 20
result1 = [r['var1'] for r in table.iterrows() if r['var2'][0][0] < 20]
result2 = [r for r in table.iterrows() if r['var2'][0][0] < 20][-1]
if result2['var1'].dtype.char == "S":
a = np.array([['%04d' % (self.expectedrows - 0)]*2]*2, 'S')
self.assertTrue(common.allequal(result1[0], a))
a = np.array([['%04d' % (self.expectedrows - 1)]*2]*2, 'S')
self.assertTrue(common.allequal(result1[1], a))
a = np.array([['%04d' % (self.expectedrows - 2)]*2]*2, 'S')
self.assertTrue(common.allequal(result1[2], a))
a = np.array([['%04d' % (self.expectedrows - 3)]*2]*2, 'S')
self.assertTrue(common.allequal(result1[3], a))
a = np.array([['%04d' % (self.expectedrows - 10)]*2]*2, 'S')
self.assertTrue(common.allequal(result1[10], a))
a = np.array([['%04d' % (1)]*2]*2, 'S')
self.assertTrue(common.allequal(result2['var1'], a))
else:
self.assertEqual(result1['var1'], "0001")
self.assertEqual(len(result1), 20)
def test01c_readTable(self):
"""Checking shape of multidimensional columns."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01c_readTable..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
table = self.h5file.get_node("/table0")
if common.verbose:
print("var2 col shape:", table.cols.var2.shape)
print("Should be:", table.cols.var2[:].shape)
self.assertEqual(table.cols.var2.shape, table.cols.var2[:].shape)
def test02_AppendRows(self):
"""Checking whether appending record rows works or not."""
# Now, open it, but in "append" mode
self.h5file = tb.open_file(self.h5fname, mode="a")
self.rootgroup = self.h5file.root
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_AppendRows..." % self.__class__.__name__)
# Get a table
table = self.h5file.get_node("/group0/table1")
# Get their row object
row = table.row
if common.verbose:
print("Nrows in old", table._v_pathname, ":", table.nrows)
print("Record Format ==>", table.description._v_nested_formats)
print("Record Size ==>", table.rowsize)
# Append some rows
for i in range(self.appendrows):
s = '%04d' % (self.appendrows - i)
row['var0'] = s.encode('ascii')
row['var1'] = s.encode('ascii')
row['var7'] = s[-1].encode('ascii')
row['var1_'] = (i, 1)
row['var2'] = ((i, 1), (1, 1)) # *-*
row['var3'] = i % self.maxshort
if isinstance(row['var4'], np.ndarray):
row['var4'] = [float(i), float(i * i)]
else:
row['var4'] = float(i)
if isinstance(row['var5'], np.ndarray):
row['var5'] = np.array((float(i),)*4)
else:
row['var5'] = float(i)
row.append()
# Flush the buffer for this table and read it
table.flush()
result = [r['var2'][0][0] for r in table.iterrows()
if r['var2'][0][0] < 20]
row = [r for r in table.iterrows() if r['var2'][0][0] < 20][-1]
nrows = self.appendrows - 1
self.assertEqual((
row['var0'][0],
row['var1'][0][0],
row['var1_'][0],
row['var2'][0][0],
row['var7']),
(b"0001", b"0001", nrows, nrows, b"1"))
if isinstance(row['var5'], np.ndarray):
self.assertTrue(common.allequal(
row['var5'], np.array((float(nrows),) * 4, np.float32)))
else:
self.assertEqual(row['var5'], float(nrows))
if self.appendrows <= 20:
add = self.appendrows
else:
add = 20
self.assertEqual(len(result), 20 + add) # because we appended new rows
# del table
# CAVEAT: The next test only works for tables with rows < 2**15
def test03_endianess(self):
"""Checking if table is endianess aware."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_endianess..." % self.__class__.__name__)
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
table = self.h5file.get_node("/group0/group1/table2")
# Read the records and select the ones with "var3" column less than 20
result = [r['var2'] for r in table.iterrows() if r['var3'] < 20]
if common.verbose:
print("Nrows in", table._v_pathname, ":", table.nrows)
print("On-disk byteorder ==>", table.byteorder)
print("Last record in table ==>", table[-1])
print("Total selected records in table ==>", len(result))
nrows = self.expectedrows - 1
r = list(table.iterrows())[-1]
self.assertEqual((r['var1'][0][0], r['var3']), (b"0001", nrows))
self.assertEqual(len(result), 20)
class BasicWriteTestCase(BasicTestCase):
title = "BasicWrite"
class DictWriteTestCase(BasicTestCase):
# This checks also unidimensional arrays as columns
title = "DictWrite"
record = RecordDescriptionDict
nrows = 21
nrowsinbuf = 3 # Choose a small value for the buffer size
start = 0
stop = 10
step = 3
class RecordDTWriteTestCase(BasicTestCase):
title = "RecordDTWriteTestCase"
record = RecordDT
# Pure NumPy dtype
class NumPyDTWriteTestCase(BasicTestCase):
title = "NumPyDTWriteTestCase"
record = np.dtype("(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,2f8,f4,i2,S1")
record.names = 'var0,var1,var1_,var2,var3,var4,var5,var6,var7'.split(',')
class RecArrayOneWriteTestCase(BasicTestCase):
title = "RecArrayOneWrite"
record = np.rec.array(
None,
formats="(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,2f8,f4,i2,S1",
names='var0,var1,var1_,var2,var3,var4,var5,var6,var7',
shape=0)
class RecArrayTwoWriteTestCase(BasicTestCase):
title = "RecArrayTwoWrite"
expectedrows = 100
recarrayinit = 1
recordtemplate = np.rec.array(
None,
formats="(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,f8,f4,i2,S1",
names='var0,var1,var1_,var2,var3,var4,var5,var6,var7',
shape=1)
class RecArrayThreeWriteTestCase(BasicTestCase):
title = "RecArrayThreeWrite"
expectedrows = 100
recarrayinit = 1
recordtemplate = np.rec.array(
None,
formats="(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,2f8,4f4,i2,S1",
names='var0,var1,var1_,var2,var3,var4,var5,var6,var7',
shape=1)
class RecArrayAlignedWriteTestCase(BasicTestCase):
title = "RecArrayThreeWrite"
expectedrows = 100
recarrayinit = 1
recordtemplate = np.rec.array(
None,
formats="(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,2f8,4f4,i2,S1",
names='var0,var1,var1_,var2,var3,var4,var5,var6,var7',
shape=1, aligned=True)
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class CompressBloscTablesTestCase(BasicTestCase):
title = "CompressBloscTables"
compress = 1
complib = "blosc"
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class CompressLZOTablesTestCase(BasicTestCase):
title = "CompressLZOTables"
compress = 1
complib = "lzo"
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class CompressBzip2TablesTestCase(BasicTestCase):
title = "CompressBzip2Tables"
compress = 1
complib = "bzip2"
class CompressZLIBTablesTestCase(BasicTestCase):
title = "CompressOneTables"
compress = 1
complib = "zlib"
class CompressTwoTablesTestCase(BasicTestCase):
title = "CompressTwoTables"
compress = 1
# This checks also unidimensional arrays as columns
record = RecordDescriptionDict
class BigTablesTestCase(BasicTestCase):
title = "BigTables"
# 10000 rows takes much more time than we can afford for tests
# reducing to 1000 would be more than enough
# F. Alted 2004-01-19
# expectedrows = 10000
# appendrows = 1000
expectedrows = 1000
appendrows = 100
class BasicRangeTestCase(common.TempFileMixin, common.PyTablesTestCase):
# file = "test.h5"
open_mode = "w"
title = "This is the table title"
record = Record
maxshort = 1 << 15
expectedrows = 100
compress = 0
# Default values
nrows = 20
nrowsinbuf = 3 # Choose a small value for the buffer size
start = 1
stop = nrows
checkrecarray = 0
checkgetCol = 0
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
for j in range(3):
# Create a table
table = self.h5file.create_table(
group, 'table'+str(j), self.record, title=self.title,
filters=tb.Filters(self.compress),
expectedrows=self.expectedrows)
# Get the row object associated with the new table
row = table.row
# Fill the table
for i in range(self.expectedrows):
row['var1'] = '%04d' % (self.expectedrows - i)
row['var7'] = row['var1'][0][0][-1]
row['var2'] = i
row['var3'] = i % self.maxshort
if isinstance(row['var4'], np.ndarray):
row['var4'] = [float(i), float(i * i)]
else:
row['var4'] = float(i)
if isinstance(row['var5'], np.ndarray):
row['var5'] = np.array((float(i),)*4)
else:
row['var5'] = float(i)
# var6 will be like var3 but byteswaped
row['var6'] = (((row['var3'] >> 8) & 0xff) +
((row['var3'] << 8) & 0xff00))
row.append()
# Flush the buffer for this table
table.flush()
# Create a new group (descendant of group)
group2 = self.h5file.create_group(group, 'group'+str(j))
# Iterate over this new group (group2)
group = group2
def check_range(self):
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
table = self.h5file.get_node("/table0")
table.nrowsinbuf = self.nrowsinbuf
resrange = slice(self.start, self.stop, self.step).indices(table.nrows)
reslength = len(list(range(*resrange)))
if self.checkrecarray:
recarray = table.read(self.start, self.stop, self.step)
result = []
for nrec in range(len(recarray)):
if recarray['var2'][nrec][0][0] < self.nrows and 0 < self.step:
result.append(recarray['var2'][nrec][0][0])
elif (recarray['var2'][nrec][0][0] > self.nrows and
0 > self.step):
result.append(recarray['var2'][nrec][0][0])
elif self.checkgetCol:
column = table.read(self.start, self.stop, self.step, 'var2')
result = []
for nrec in range(len(column)):
if column[nrec][0][0] < self.nrows and 0 < self.step: # *-*
result.append(column[nrec][0][0]) # *-*
elif column[nrec][0][0] > self.nrows and 0 > self.step: # *-*
result.append(column[nrec][0][0]) # *-*
else:
if 0 < self.step:
result = [
r['var2'][0][0] for r in table.iterrows(self.start,
self.stop,
self.step)
if r['var2'][0][0] < self.nrows
]
elif 0 > self.step:
result = [
r['var2'][0][0] for r in table.iterrows(self.start,
self.stop,
self.step)
if r['var2'][0][0] > self.nrows
]
if self.start < 0:
startr = self.expectedrows + self.start
else:
startr = self.start
if self.stop is None:
if self.checkrecarray or self.checkgetCol:
# data read using the read method
stopr = startr + 1
else:
# data read using the iterrows method
stopr = self.nrows
elif self.stop < 0:
stopr = self.expectedrows + self.stop
else:
stopr = self.stop
if self.nrows < stopr:
stopr = self.nrows
if common.verbose:
print("Nrows in", table._v_pathname, ":", table.nrows)
if reslength:
if self.checkrecarray:
print("Last record *read* in recarray ==>", recarray[-1])
elif self.checkgetCol:
print("Last value *read* in getCol ==>", column[-1])
else:
r = list(
table.iterrows(self.start, self.stop, self.step))[-1]
print("Last record *read* in table range ==>", r)
print("Total number of selected records ==>", len(result))
print("Selected records:\n", result)
print("Selected records should look like:\n",
list(range(startr, stopr, self.step)))
print("start, stop, step ==>", startr, stopr, self.step)
self.assertEqual(result, list(range(startr, stopr, self.step)))
if not (self.checkrecarray or self.checkgetCol):
if startr < stopr and 0 < self.step:
r = [r['var2'] for r in table.iterrows(self.start, self.stop,
self.step)
if r['var2'][0][0] < self.nrows][-1]
if self.nrows > self.expectedrows:
self.assertEqual(
r[0][0],
list(range(self.start, self.stop, self.step))[-1])
else:
self.assertEqual(r[0][0],
list(range(startr, stopr, self.step))[-1])
elif startr > stopr and 0 > self.step:
r = [r['var2'] for r in table.iterrows(self.start, self.stop,
self.step)
if r['var2'][0][0] > self.nrows][0]
if self.nrows < self.expectedrows:
self.assertEqual(
r[0][0],
list(range(self.start, self.stop or -1, self.step))[0])
else:
self.assertEqual(
r[0][0],
list(range(startr, stopr or -1, self.step))[0])
# Close the file
self.h5file.close()
def test01_range(self):
"""Checking ranges in table iterators (case1)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_range..." % self.__class__.__name__)
# Case where step < nrowsinbuf < 2 * step
self.nrows = 21
self.nrowsinbuf = 3
self.start = 0
self.stop = self.expectedrows
self.step = 2
self.check_range()
def test01a_range(self):
"""Checking ranges in table iterators (case1)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_range..." % self.__class__.__name__)
# Case where step < nrowsinbuf < 2 * step
self.nrows = 21
self.nrowsinbuf = 3
self.start = self.expectedrows - 1
self.stop = None
self.step = -2
self.check_range()
def test02_range(self):
"""Checking ranges in table iterators (case2)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_range..." % self.__class__.__name__)
# Case where step < nrowsinbuf < 10 * step
self.nrows = 21
self.nrowsinbuf = 31
self.start = 11
self.stop = self.expectedrows
self.step = 3
self.check_range()
def test03_range(self):
"""Checking ranges in table iterators (case3)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_range..." % self.__class__.__name__)
# Case where step < nrowsinbuf < 1.1 * step
self.nrows = self.expectedrows
self.nrowsinbuf = 11 # Choose a small value for the buffer size
self.start = 0
self.stop = self.expectedrows
self.step = 10
self.check_range()
def test04_range(self):
"""Checking ranges in table iterators (case4)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_range..." % self.__class__.__name__)
# Case where step == nrowsinbuf
self.nrows = self.expectedrows
self.nrowsinbuf = 11 # Choose a small value for the buffer size
self.start = 1
self.stop = self.expectedrows
self.step = 11
self.check_range()
def test05_range(self):
"""Checking ranges in table iterators (case5)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_range..." % self.__class__.__name__)
# Case where step > 1.1 * nrowsinbuf
self.nrows = 21
self.nrowsinbuf = 10 # Choose a small value for the buffer size
self.start = 1
self.stop = self.expectedrows
self.step = 11
self.check_range()
def test06_range(self):
"""Checking ranges in table iterators (case6)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test06_range..." % self.__class__.__name__)
# Case where step > 3 * nrowsinbuf
self.nrows = 3
self.nrowsinbuf = 3 # Choose a small value for the buffer size
self.start = 2
self.stop = self.expectedrows
self.step = 10
self.check_range()
def test07_range(self):
"""Checking ranges in table iterators (case7)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test07_range..." % self.__class__.__name__)
# Case where start == stop
self.nrows = 2
self.nrowsinbuf = 3 # Choose a small value for the buffer size
self.start = self.nrows
self.stop = self.nrows
self.step = 10
self.check_range()
def test08_range(self):
"""Checking ranges in table iterators (case8)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08_range..." % self.__class__.__name__)
# Case where start > stop
self.nrows = 2
self.nrowsinbuf = 3 # Choose a small value for the buffer size
self.start = self.nrows + 1
self.stop = self.nrows
self.step = 1
self.check_range()
def test09_range(self):
"""Checking ranges in table iterators (case9)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test09_range..." % self.__class__.__name__)
# Case where stop = None
self.nrows = 100
self.nrowsinbuf = 3 # Choose a small value for the buffer size
self.start = 1
self.stop = 2
self.step = 1
self.check_range()
def test10_range(self):
"""Checking ranges in table iterators (case10)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10_range..." % self.__class__.__name__)
# Case where start < 0 and stop = 0
self.nrows = self.expectedrows
self.nrowsinbuf = 5 # Choose a small value for the buffer size
self.start = -6
self.startr = self.expectedrows + self.start
self.stop = 0
self.stopr = self.expectedrows + self.stop
self.step = 2
self.check_range()
def test11_range(self):
"""Checking ranges in table iterators (case11)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test11_range..." % self.__class__.__name__)
# Case where start < 0 and stop < 0
self.nrows = self.expectedrows
self.nrowsinbuf = 5 # Choose a small value for the buffer size
self.start = -6
self.startr = self.expectedrows + self.start
self.stop = -2
self.stopr = self.expectedrows + self.stop
self.step = 1
self.check_range()
def test12_range(self):
"""Checking ranges in table iterators (case12)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test12_range..." % self.__class__.__name__)
# Case where start < 0 and stop < 0 and start > stop
self.nrows = self.expectedrows
self.nrowsinbuf = 5 # Choose a small value for the buffer size
self.start = -1
self.startr = self.expectedrows + self.start
self.stop = -2
self.stopr = self.expectedrows + self.stop
self.step = 1
self.check_range()
def test13_range(self):
"""Checking ranges in table iterators (case13)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test13_range..." % self.__class__.__name__)
# Case where step < 0
self.step = -11
try:
self.check_range()
except ValueError:
if common.verbose:
(type, value, traceback) = sys.exc_info()
print("\nGreat!, the next ValueError was catched!")
self.h5file.close()
# else:
# self.fail("expected a ValueError")
# Case where step == 0
self.step = 0
try:
self.check_range()
except ValueError:
if common.verbose:
(type, value, traceback) = sys.exc_info()
print("\nGreat!, the next ValueError was catched!")
self.h5file.close()
# else:
# self.fail("expected a ValueError")
class IterRangeTestCase(BasicRangeTestCase):
pass
class RecArrayRangeTestCase(BasicRangeTestCase):
checkrecarray = 1
class GetColRangeTestCase(BasicRangeTestCase):
checkgetCol = 1
def test01_nonexistentField(self):
"""Checking non-existing Field in getCol method """
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_nonexistentField..." %
self.__class__.__name__)
# Create an instance of an HDF5 Table
self.h5file = tb.open_file(self.h5fname, "r")
self.root = self.h5file.root
table = self.h5file.get_node("/table0")
with self.assertRaises(KeyError):
table.read(field='non-existent-column')
class Rec(tb.IsDescription):
col1 = tb.IntCol(pos=1, shape=(2,))
col2 = tb.StringCol(itemsize=3, pos=2, shape=(3,))
col3 = tb.FloatCol(pos=3, shape=(3, 2))
class RecArrayIO(common.TempFileMixin, common.PyTablesTestCase):
def test00(self):
"""Checking saving a normal recarray"""
# Create a recarray
intlist1 = [[456, 23]*3]*2
intlist2 = np.array([[2, 2]*3]*2, dtype=int)
arrlist1 = [['dbe']*2]*3
arrlist2 = [['de']*2]*3
floatlist1 = [[1.2, 2.3]*3]*4
floatlist2 = np.array([[4.5, 2.4]*3]*4)
b = [(intlist1, arrlist1, floatlist1), (
intlist2, arrlist2, floatlist2)]
r = np.rec.array(b, formats='(2,6)i4,(3,2)S3,(4,6)f8',
names='col1,col2,col3')
# Save it in a table:
self.h5file.create_table(self.h5file.root, 'recarray', r)
# Read it again
r2 = self.h5file.root.recarray.read()
self.assertEqual(r.tobytes(), r2.tobytes())
def test01(self):
"""Checking saving a recarray with an offset in its buffer"""
# Create a recarray
intlist1 = [[456, 23]*3]*2
intlist2 = np.array([[2, 2]*3]*2, dtype=int)
arrlist1 = [['dbe']*2]*3
arrlist2 = [['de']*2]*3
floatlist1 = [[1.2, 2.3]*3]*4
floatlist2 = np.array([[4.5, 2.4]*3]*4)
b = [(intlist1, arrlist1, floatlist1), (
intlist2, arrlist2, floatlist2)]
r = np.rec.array(b, formats='(2,6)i4,(3,2)S3,(4,6)f8',
names='col1,col2,col3')
# Get a view of the recarray
r1 = r[1:]
# Save it in a table:
self.h5file.create_table(self.h5file.root, 'recarray', r1)
# Read it again
r2 = self.h5file.root.recarray.read()
self.assertEqual(r1.tobytes(), r2.tobytes())
def test02(self):
"""Checking saving a slice of a large recarray"""
# Create a recarray
intlist1 = [[[23, 24, 35]*6]*6]
intlist2 = np.array([[[2, 3, 4]*6]*6], dtype=int)
arrlist1 = [['dbe']*2]*3
arrlist2 = [['de']*2]*3
floatlist1 = [[1.2, 2.3]*3]*4
floatlist2 = np.array([[4.5, 2.4]*3]*4)
b = [(intlist1, arrlist1, floatlist1), (
intlist2, arrlist2, floatlist2)]
r = np.rec.array(b * 300, formats='(1,6,18)i4,(3,2)S3,(4,6)f8',
names='col1,col2,col3')
# Get a slice of recarray
r1 = r[290:292]
# Save it in a table:
self.h5file.create_table(self.h5file.root, 'recarray', r1)
# Read it again
r2 = self.h5file.root.recarray.read()
self.assertEqual(r1.tobytes(), r2.tobytes())
def test03(self):
"""Checking saving a slice of a strided recarray"""
# Create a recarray
intlist1 = [[[23, 24, 35]*6]*6]
intlist2 = np.array([[[2, 3, 4]*6]*6], dtype=int)
arrlist1 = [['dbe']*2]*3
arrlist2 = [['de']*2]*3
floatlist1 = [[1.2, 2.3]*3]*4
floatlist2 = np.array([[4.5, 2.4]*3]*4)
b = [(intlist1, arrlist1, floatlist1), (
intlist2, arrlist2, floatlist2)]
r = np.rec.array(b * 300, formats='(1,6,18)i4,(3,2)S3,(4,6)f8',
names='col1,col2,col3', shape=600)
# Get a strided recarray
r2 = r[::2]
# Get a slice
r1 = r2[148:]
# Save it in a table:
self.h5file.create_table(self.h5file.root, 'recarray', r1)
# Read it again
r2 = self.h5file.root.recarray.read()
self.assertEqual(r1.tobytes(), r2.tobytes())
def test08a(self):
"""Checking modifying one column (single column version, list)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08a..." % self.__class__.__name__)
# Create a new table:
table = self.h5file.create_table(self.h5file.root, 'recarray', Rec)
# Append new rows
s0, s1, s2, s3 = ['dbe']*3, ['ded']*3, ['db1']*3, ['de1']*3
f0, f1, f2, f3 = [[1.2]*2]*3, [[1.3]*2]*3, [[1.4]*2]*3, [[1.5]*2]*3
r = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1)],
formats="(2,)i4,(3,)S3,(3,2)f8")
table.append(r)
table.append([([457, 458], s2, f2), ([5, 6], s3, f3)])
# Modify just one existing column
table.cols.col1[1:] = [[[2, 3], [3, 4], [4, 5]]]
# Create the modified recarray
r1 = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1),
([3, 4], s2, f2), ([4, 5], s3, f3)],
formats="(2,)i4,(3,)S3,(3,2)f8",
names="col1,col2,col3")
# Read the modified table
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test08b(self):
"""Checking modifying one column (single column version, recarray)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08b..." % self.__class__.__name__)
# Create a new table:
table = self.h5file.create_table(self.h5file.root, 'recarray', Rec)
# Append new rows
s0, s1, s2, s3 = ['dbe']*3, ['ded']*3, ['db1']*3, ['de1']*3
f0, f1, f2, f3 = [[1.2]*2]*3, [[1.3]*2]*3, [[1.4]*2]*3, [[1.5]*2]*3
r = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1)],
formats="(2,)i4,(3,)S3,(3,2)f8")
table.append(r)
table.append([([457, 458], s2, f2), ([5, 6], s3, f3)])
# Modify just one existing column
columns = np.rec.fromarrays(
np.array([[[2, 3], [3, 4], [4, 5]]]), formats="i4")
table.modify_columns(start=1, columns=columns, names=["col1"])
# Create the modified recarray
r1 = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1),
([3, 4], s2, f2), ([4, 5], s3, f3)],
formats="(2,)i4,(3,)S3,(3,2)f8",
names="col1,col2,col3")
# Read the modified table
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test08b2(self):
"""Checking modifying one column (single column version, recarray,
modify_column)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08b2..." % self.__class__.__name__)
# Create a new table:
table = self.h5file.create_table(self.h5file.root, 'recarray', Rec)
# Append new rows
s0, s1, s2, s3 = ['dbe']*3, ['ded']*3, ['db1']*3, ['de1']*3
f0, f1, f2, f3 = [[1.2]*2]*3, [[1.3]*2]*3, [[1.4]*2]*3, [[1.5]*2]*3
r = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1)],
formats="(2,)i4,(3,)S3,(3,2)f8")
table.append(r)
table.append([([457, 458], s2, f2), ([5, 6], s3, f3)])
# Modify just one existing column
columns = np.rec.fromarrays(
np.array([[[2, 3], [3, 4], [4, 5]]]), formats="i4")
table.modify_column(start=1, column=columns, colname="col1")
# Create the modified recarray
r1 = np.rec.array([([456, 457], s0, f0), ([2, 3], s1, f1),
([3, 4], s2, f2), ([4, 5], s3, f3)],
formats="(2,)i4,(3,)S3,(3,2)f8",
names="col1,col2,col3")
# Read the modified table
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
class DefaultValues(common.TempFileMixin, common.PyTablesTestCase):
def test00(self):
"""Checking saving a Table MD with default values"""
# Create a table
table = self.h5file.create_table(self.h5file.root, 'table', Record)
# Take a number of records a bit large
# nrows = int(table.nrowsinbuf * 1.1)
nrows = 5 # for test
# Fill the table with nrows records
for i in range(nrows):
if i == 3 or i == 4:
table.row['var2'] = ((2, 2), (2, 2)) # *-*
# This injects the row values.
table.row.append()
# We need to flush the buffers in table in order to get an
# accurate number of records on it.
table.flush()
# Create a recarray with the same default values
buffer = [(
["\x00"]*2, # just "" does not initialize the buffer properly
[["abcd", "efgh"]]*2,
(1, 1),
((1, 1), (1, 1)),
2, 3.1, 4.2, 5, "e")]
r = np.rec.array(
buffer * nrows,
formats='(2,)S4,(2,2)S4,(2,)i4,(2,2)i4,i2,f8,f4,u2,S1',
names=['var0', 'var1', 'var1_', 'var2', 'var3', 'var4', 'var5',
'var6', 'var7']) # *-*
# Assign the value exceptions
r["var2"][3] = ((2, 2), (2, 2)) # *-*
r["var2"][4] = ((2, 2), (2, 2)) # *-*
# Read the table in another recarray
r2 = table.read()
# This generates too much output. Activate only when
# self.nrowsinbuf is very small (<10)
if common.verbose and 1:
print("Table values:")
print(r2)
print("Record values:")
print(r)
# Both checks do work, however, tobytes() seems more stringent.
self.assertEqual(r.tobytes(), r2.tobytes())
# self.assertTrue(common.areArraysEqual(r,r2))
class RecordT(tb.IsDescription):
var0 = tb.IntCol(dflt=1, shape=()) # native int
var1 = tb.IntCol(dflt=[1], shape=(1,)) # 1-D int (one element)
var2_s = tb.IntCol(dflt=[1, 1], shape=2) # 1-D int (two elements)
var2 = tb.IntCol(dflt=[1, 1], shape=(2,)) # 1-D int (two elements)
var3 = tb.IntCol(dflt=[[0, 0], [1, 1]], shape=(2, 2)) # 2-D int
class ShapeTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
self.populateFile()
def populateFile(self):
table = self.h5file.create_table(self.h5file.root, 'table', RecordT)
row = table.row
# Fill the table with some rows with default values
for i in range(1):
row.append()
# Flush the buffer for this table
table.flush()
def test00(self):
"""Checking scalar shapes"""
if self.reopen:
self._reopen()
table = self.h5file.root.table
if common.verbose:
print("The values look like:", table.cols.var0[:])
print("They should look like:", [1])
# The real check
self.assertEqual(table.cols.var0[:].tolist(), [1])
def test01(self):
"""Checking undimensional (one element) shapes"""
if self.reopen:
self._reopen()
table = self.h5file.root.table
if common.verbose:
print("The values look like:", table.cols.var1[:])
print("They should look like:", [[1]])
# The real check
self.assertEqual(table.cols.var1[:].tolist(), [[1]])
def test02(self):
"""Checking undimensional (two elements) shapes"""
if self.reopen:
self._reopen()
table = self.h5file.root.table
if common.verbose:
print("The values look like:", table.cols.var2[:])
print("They should look like:", [[1, 1]])
# The real check
self.assertEqual(table.cols.var2[:].tolist(), [[1, 1]])
self.assertEqual(table.cols.var2_s[:].tolist(), [[1, 1]])
def test03(self):
"""Checking bidimensional shapes"""
if self.reopen:
self._reopen()
table = self.h5file.root.table
if common.verbose:
print("The values look like:", table.cols.var3[:])
print("They should look like:", [[[0, 0], [1, 1]]])
# The real check
self.assertEqual(table.cols.var3[:].tolist(), [[[0, 0], [1, 1]]])
class ShapeTestCase1(ShapeTestCase):
reopen = 0
class ShapeTestCase2(ShapeTestCase):
reopen = 1
class SetItemTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
# Create a new table:
self.table = self.h5file.create_table(self.h5file.root,
'recarray', Rec)
self.table.nrowsinbuf = self.buffersize # set buffer value
def test01(self):
"""Checking modifying one table row with __setitem__"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing row
table[2] = (456, 'db2', 1.2)
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(456, 'db2', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test01b(self):
"""Checking modifying one table row with __setitem__ (long index)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing row
table[2] = (456, 'db2', 1.2)
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(456, 'db2', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test02(self):
"""Modifying one row, with a step (__setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
rows = np.rec.array([(457, 'db1', 1.2)], formats=formats)
table[1:3:2] = rows
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(457, 'db1', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test03(self):
"""Checking modifying several rows at once (__setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
rows = np.rec.array(
[(457, 'db1', 1.2), (5, 'de1', 1.3)], formats=formats)
# table.modify_rows(start=1, rows=rows)
table[1:3] = rows
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(5, 'de1', 1.3), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test04(self):
"""Modifying several rows at once, with a step (__setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
rows = np.rec.array([(457, 'db1', 1.2), (6, 'de2', 1.3)],
formats=formats)
# table[1:4:2] = rows
table[1::2] = rows
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(457, 'db1', 1.2), (6, 'de2', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test05(self):
"""Checking modifying one column (single element, __setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
table.cols.col1[1] = -1
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (-1, 'ded', 1.3),
(457, 'db1', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test06a(self):
"""Checking modifying one column (several elements, __setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
table.cols.col1[1:4] = [(2, 2), (3, 3), (4, 4)]
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(3, 'db1', 1.2), (4, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test06b(self):
"""Checking modifying one column (iterator, __setitem__)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
with self.assertRaises(NotImplementedError):
for row in table.iterrows():
row['col1'] = row.nrow + 1
row.append()
table.flush()
def test07(self):
"""Modifying one column (several elements, __setitem__, step)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (1, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
table.cols.col1[1:4:2] = [(2, 2), (3, 3)]
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(457, 'db1', 1.2), (3, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test08(self):
"""Modifying one column (one element, __setitem__, step)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
table.cols.col1[1:4:3] = [(2, 2)]
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(457, 'db1', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test09(self):
"""Modifying beyond the table extend (__setitem__, step)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Try to modify beyond the extent
# This will silently exclude the non-fitting rows
rows = np.rec.array([(457, 'db1', 1.2), (6, 'de2', 1.3)],
formats=formats)
table[1::2] = rows
# How it should look like
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(457, 'db1', 1.2), (6, 'de2', 1.3)],
formats=formats)
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
class SetItemTestCase1(SetItemTestCase):
reopen = 0
buffersize = 1
class SetItemTestCase2(SetItemTestCase):
reopen = 1
buffersize = 2
class SetItemTestCase3(SetItemTestCase):
reopen = 0
buffersize = 1000
class SetItemTestCase4(SetItemTestCase):
reopen = 1
buffersize = 1000
class UpdateRowTestCase(common.TempFileMixin, common.PyTablesTestCase):
def setUp(self):
super().setUp()
# Create a new table:
self.table = self.h5file.create_table(self.h5file.root,
'recarray', Rec)
self.table.nrowsinbuf = self.buffersize # set buffer value
def test01(self):
"""Checking modifying one table row with Row.update"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)],
formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing row
for row in table.iterrows(2, 3):
(row['col1'], row['col2'], row['col3']) = [456, 'db2', 1.2]
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(456, 'db2', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test02(self):
"""Modifying one row, with a step (Row.update)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
for row in table.iterrows(1, 3, 2):
if row.nrow == 1:
(row['col1'], row['col2'], row['col3']) = (457, 'db1', 1.2)
elif row.nrow == 3:
(row['col1'], row['col2'], row['col3']) = (6, 'de2', 1.3)
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(457, 'db1', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test03(self):
"""Checking modifying several rows at once (Row.update)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
for row in table.iterrows(1, 3):
if row.nrow == 1:
(row['col1'], row['col2'], row['col3']) = (457, 'db1', 1.2)
elif row.nrow == 2:
(row['col1'], row['col2'], row['col3']) = (5, 'de1', 1.3)
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(5, 'de1', 1.3), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test04(self):
"""Modifying several rows at once, with a step (Row.update)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify two existing rows
for row in table.iterrows(1, stop=4, step=2):
if row.nrow == 1:
(row['col1'], row['col2'], row['col3']) = (457, 'db1', 1.2)
elif row.nrow == 3:
(row['col1'], row['col2'], row['col3']) = (6, 'de2', 1.3)
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (457, 'db1', 1.2),
(457, 'db1', 1.2), (6, 'de2', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test05(self):
"""Checking modifying one column (single element, Row.update)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
for row in table.iterrows(1, 2):
row['col1'] = -1
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (-1, 'ded', 1.3),
(457, 'db1', 1.2), (5, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test06(self):
"""Checking modifying one column (several elements, Row.update)"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just one existing column
for row in table.iterrows(1, 4):
row['col1'] = row.nrow + 1
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ded', 1.3),
(3, 'db1', 1.2), (4, 'de1', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test07(self):
"""Modifying values from a selection"""
table = self.table
formats = table.description._v_nested_formats
# append new rows
r = np.rec.array([(456, 'dbe', 1.2), (1, 'ded', 1.3)], formats=formats)
table.append(r)
table.append([(457, 'db1', 1.2), (5, 'de1', 1.3)])
# Modify just rows with col1 < 456
for row in table.iterrows():
if row['col1'][0] < 456:
row['col1'] = 2
row['col2'] = 'ada'
row.update()
# Create the modified recarray
r1 = np.rec.array([(456, 'dbe', 1.2), (2, 'ada', 1.3),
(457, 'db1', 1.2), (2, 'ada', 1.3)],
formats=formats,
names="col1,col2,col3")
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, 4)
def test08(self):
"""Modifying a large table (Row.update)"""
table = self.table
formats = table.description._v_nested_formats
nrows = 100
# append new rows
row = table.row
for i in range(nrows):
row['col1'] = i-1
row['col2'] = 'a'+str(i-1)
row['col3'] = -1.0
row.append()
table.flush()
# Modify all the rows
for row in table.iterrows():
row['col1'] = row.nrow
row['col2'] = 'b'+str(row.nrow)
row['col3'] = 0.0
row.update()
# Create the modified recarray
r1 = np.rec.array(
None, shape=nrows, formats=formats, names="col1,col2,col3")
for i in range(nrows):
r1['col1'][i] = i
r1['col2'][i] = 'b'+str(i)
r1['col3'][i] = 0.0
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, nrows)
def test08b(self):
"""Setting values on a large table without calling Row.update"""
table = self.table
formats = table.description._v_nested_formats
nrows = 100
# append new rows
row = table.row
for i in range(nrows):
row['col1'] = i-1
row['col2'] = 'a'+str(i-1)
row['col3'] = -1.0
row.append()
table.flush()
# Modify all the rows (actually don't)
for row in table.iterrows():
row['col1'] = row.nrow
row['col2'] = 'b'+str(row.nrow)
row['col3'] = 0.0
# row.update()
# Create the modified recarray
r1 = np.rec.array(
None, shape=nrows, formats=formats, names="col1,col2,col3")
for i in range(nrows):
r1['col1'][i] = i-1
r1['col2'][i] = 'a'+str(i-1)
r1['col3'][i] = -1.0
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, nrows)
def test09(self):
"""Modifying selected values on a large table"""
table = self.table
formats = table.description._v_nested_formats
nrows = 100
# append new rows
row = table.row
for i in range(nrows):
row['col1'] = i-1
row['col2'] = 'a'+str(i-1)
row['col3'] = -1.0
row.append()
table.flush()
# Modify selected rows
for row in table.iterrows():
if row['col1'][0] > nrows-3:
row['col1'] = row.nrow
row['col2'] = 'b'+str(row.nrow)
row['col3'] = 0.0
row.update()
# Create the modified recarray
r1 = np.rec.array(
None, shape=nrows, formats=formats, names="col1,col2,col3")
for i in range(nrows):
r1['col1'][i] = i-1
r1['col2'][i] = 'a'+str(i-1)
r1['col3'][i] = -1.0
# modify just the last line
r1['col1'][i] = i
r1['col2'][i] = 'b'+str(i)
r1['col3'][i] = 0.0
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, nrows)
def test09b(self):
"""Modifying selected values on a large table (alternate values)"""
table = self.table
formats = table.description._v_nested_formats
nrows = 100
# append new rows
row = table.row
for i in range(nrows):
row['col1'] = i-1
row['col2'] = 'a'+str(i-1)
row['col3'] = -1.0
row.append()
table.flush()
# Modify selected rows
for row in table.iterrows(step=10):
row['col1'] = row.nrow
row['col2'] = 'b'+str(row.nrow)
row['col3'] = 0.0
row.update()
# Create the modified recarray
r1 = np.rec.array(
None, shape=nrows, formats=formats, names="col1,col2,col3")
for i in range(nrows):
if i % 10 > 0:
r1['col1'][i] = i-1
r1['col2'][i] = 'a'+str(i-1)
r1['col3'][i] = -1.0
else:
r1['col1'][i] = i
r1['col2'][i] = 'b'+str(i)
r1['col3'][i] = 0.0
# Read the modified table
if self.reopen:
self._reopen()
table = self.h5file.root.recarray
table.nrowsinbuf = self.buffersize # set buffer value
r2 = table.read()
if common.verbose:
print("Original table-->", repr(r2))
print("Should look like-->", repr(r1))
self.assertEqual(r1.tobytes(), r2.tobytes())
self.assertEqual(table.nrows, nrows)
class UpdateRowTestCase1(UpdateRowTestCase):
reopen = 0
buffersize = 1
class UpdateRowTestCase2(UpdateRowTestCase):
reopen = 1
buffersize = 2
class UpdateRowTestCase3(UpdateRowTestCase):
reopen = 0
buffersize = 1000
class UpdateRowTestCase4(UpdateRowTestCase):
reopen = 1
buffersize = 1000
def suite():
theSuite = common.unittest.TestSuite()
niter = 1
# common.heavy = 1 # Uncomment this only for testing purposes
for n in range(niter):
theSuite.addTest(common.make_suite(BasicWriteTestCase))
theSuite.addTest(common.make_suite(DictWriteTestCase))
theSuite.addTest(common.make_suite(RecordDTWriteTestCase))
theSuite.addTest(common.make_suite(NumPyDTWriteTestCase))
theSuite.addTest(common.make_suite(RecArrayOneWriteTestCase))
theSuite.addTest(common.make_suite(RecArrayTwoWriteTestCase))
theSuite.addTest(common.make_suite(RecArrayThreeWriteTestCase))
theSuite.addTest(
common.make_suite(RecArrayAlignedWriteTestCase))
theSuite.addTest(common.make_suite(CompressZLIBTablesTestCase))
theSuite.addTest(common.make_suite(CompressTwoTablesTestCase))
theSuite.addTest(common.make_suite(IterRangeTestCase))
theSuite.addTest(common.make_suite(RecArrayRangeTestCase))
theSuite.addTest(common.make_suite(GetColRangeTestCase))
theSuite.addTest(common.make_suite(DefaultValues))
theSuite.addTest(common.make_suite(RecArrayIO))
theSuite.addTest(common.make_suite(ShapeTestCase1))
theSuite.addTest(common.make_suite(ShapeTestCase2))
theSuite.addTest(common.make_suite(SetItemTestCase1))
theSuite.addTest(common.make_suite(SetItemTestCase2))
theSuite.addTest(common.make_suite(SetItemTestCase3))
theSuite.addTest(common.make_suite(SetItemTestCase4))
theSuite.addTest(common.make_suite(UpdateRowTestCase1))
theSuite.addTest(common.make_suite(UpdateRowTestCase2))
theSuite.addTest(common.make_suite(UpdateRowTestCase3))
theSuite.addTest(common.make_suite(UpdateRowTestCase4))
theSuite.addTest(
common.make_suite(CompressBloscTablesTestCase))
theSuite.addTest(common.make_suite(CompressLZOTablesTestCase))
if common.heavy:
theSuite.addTest(
common.make_suite(CompressBzip2TablesTestCase))
theSuite.addTest(common.make_suite(BigTablesTestCase))
return theSuite
if __name__ == '__main__':
common.parse_argv(sys.argv)
common.print_versions()
common.unittest.main(defaultTest='suite')