Checking table read.
(self)
| 393 | self.assertEqual(expectedCol.type, col.type) |
| 394 | |
| 395 | def test01_readTable(self): |
| 396 | """Checking table read.""" |
| 397 | |
| 398 | if common.verbose: |
| 399 | print("\n", "-=" * 30) |
| 400 | print("Running %s.test01_readTable..." % self.__class__.__name__) |
| 401 | |
| 402 | # Create an instance of an HDF5 Table |
| 403 | self.h5file = tb.open_file(self.h5fname, "r") |
| 404 | table = self.h5file.get_node("/table0") |
| 405 | |
| 406 | # Choose a small value for buffer size |
| 407 | table.nrowsinbuf = 3 |
| 408 | # Read the records and select those with "var2" file less than 20 |
| 409 | result = [rec["var2"] for rec in table.iterrows() if rec["var2"] < 20] |
| 410 | if common.verbose: |
| 411 | print("Nrows in", table._v_pathname, ":", table.nrows) |
| 412 | print("Last record in table ==>", table[-1]) |
| 413 | print("Total selected records in table ==> ", len(result)) |
| 414 | nrows = self.expectedrows - 1 |
| 415 | rec = list(table.iterrows())[-1] |
| 416 | self.assertEqual( |
| 417 | (rec["var1"], rec["var2"], rec["var7"]), (b"0001", nrows, b"1") |
| 418 | ) |
| 419 | if isinstance(rec["var5"], np.ndarray): |
| 420 | self.assertTrue( |
| 421 | common.allequal( |
| 422 | rec["var5"], np.array((float(nrows),) * 4, np.float32) |
| 423 | ) |
| 424 | ) |
| 425 | else: |
| 426 | self.assertEqual(rec["var5"], float(nrows)) |
| 427 | if isinstance(rec["var9"], np.ndarray): |
| 428 | self.assertTrue( |
| 429 | common.allequal( |
| 430 | rec["var9"], |
| 431 | np.array( |
| 432 | [0.0 + float(nrows) * 1.0j, float(nrows) + 0.0j], |
| 433 | np.complex64, |
| 434 | ), |
| 435 | ) |
| 436 | ) |
| 437 | else: |
| 438 | self.assertEqual((rec["var9"]), float(nrows) + 0.0j) |
| 439 | self.assertEqual(len(result), 20) |
| 440 | |
| 441 | def test01a_fetch_all_fields(self): |
| 442 | """Checking table read (using Row.fetch_all_fields)""" |