(test_fun, path, img_type, batch, device, threads, validation_fun)
| 151 | |
| 152 | |
| 153 | def run_decode_fused(test_fun, path, img_type, batch, device, threads, validation_fun): |
| 154 | data_path = os.path.join(test_data_root, path, img_type) |
| 155 | pipe = test_fun( |
| 156 | data_path=data_path, |
| 157 | batch_size=batch, |
| 158 | num_threads=threads, |
| 159 | device_id=0, |
| 160 | device=device, |
| 161 | prefetch_queue_depth=1, |
| 162 | ) |
| 163 | idxs = [i for i in range(batch)] |
| 164 | iters = math.ceil(pipe.epoch_size("Reader") / batch) |
| 165 | for it in range(iters): |
| 166 | out_1, out_2 = pipe.run() |
| 167 | for sample_idx, img_1, img_2 in zip(idxs, out_1, out_2): |
| 168 | arr_1 = to_array(img_1) |
| 169 | arr_2 = to_array(img_2) |
| 170 | is_ok = validation_fun(arr_1, arr_2) |
| 171 | if not is_ok: |
| 172 | dump_as_core_artifacts( |
| 173 | img_1.source_info(), arr_1, arr_2, iter=it, sample_idx=sample_idx |
| 174 | ) |
| 175 | assert is_ok, ( |
| 176 | f"{validation_fun.__name__}\n" |
| 177 | + f"image: {img_1.source_info()} iter: {it} sample_idx: {sample_idx}" |
| 178 | ) |
| 179 | |
| 180 | |
| 181 | def test_image_decoder_fused(): |
nothing calls this directly
no test coverage detected