(self)
| 77 | self.assertTupleEqual(d["image"].shape, expected_shape) |
| 78 | |
| 79 | def test_set_data(self): |
| 80 | data_list1 = list(range(10)) |
| 81 | |
| 82 | transform = Compose([Lambda(func=lambda x: np.array([x * 10])), RandLambda(func=lambda x: x + 1)]) |
| 83 | |
| 84 | dataset = CacheDataset( |
| 85 | data=data_list1, |
| 86 | transform=transform, |
| 87 | cache_rate=1.0, |
| 88 | num_workers=4, |
| 89 | progress=True, |
| 90 | copy_cache=not sys.platform == "linux", |
| 91 | ) |
| 92 | |
| 93 | num_workers = 2 if sys.platform == "linux" else 0 |
| 94 | dataloader = DataLoader(dataset=dataset, num_workers=num_workers, batch_size=1) |
| 95 | for i, d in enumerate(dataloader): |
| 96 | np.testing.assert_allclose([[data_list1[i] * 10 + 1]], d) |
| 97 | # simulate another epoch, the cache content should not be modified |
| 98 | for i, d in enumerate(dataloader): |
| 99 | np.testing.assert_allclose([[data_list1[i] * 10 + 1]], d) |
| 100 | |
| 101 | # update the datalist and fill the cache content |
| 102 | data_list2 = list(range(-10, 0)) |
| 103 | dataset.set_data(data=data_list2) |
| 104 | # rerun with updated cache content |
| 105 | for i, d in enumerate(dataloader): |
| 106 | np.testing.assert_allclose([[data_list2[i] * 10 + 1]], d) |
| 107 | |
| 108 | |
| 109 | class _StatefulTransform(Transform, ThreadUnsafe): |
nothing calls this directly
no test coverage detected