(img_name, output_type, dtype, precision)
| 490 | ("cat-3449999_640_grayscale_8bit", types.ANY_DATA, types.UINT8, 8), |
| 491 | ) |
| 492 | def test_image_decoder_lossless_jpeg(img_name, output_type, dtype, precision): |
| 493 | device_id = 0 |
| 494 | if not is_nvjpeg_lossless_supported(device_id=device_id): |
| 495 | raise SkipTest("NVJPEG lossless supported on SM60+ capable devices only") |
| 496 | |
| 497 | data_dir = os.path.join(test_data_root, "db/single/jpeg_lossless/0") |
| 498 | ref_data_dir = os.path.join(test_data_root, "db/single/reference/jpeg_lossless") |
| 499 | |
| 500 | @pipeline_def(batch_size=1, device_id=device_id, num_threads=1) |
| 501 | def pipe(file): |
| 502 | encoded, _ = fn.readers.file(files=[file]) |
| 503 | decoded = fn.experimental.decoders.image( |
| 504 | encoded, device="mixed", dtype=dtype, output_type=output_type |
| 505 | ) |
| 506 | return decoded |
| 507 | |
| 508 | p = pipe(data_dir + f"/{img_name}.jpg") |
| 509 | (out,) = p.run() |
| 510 | result = np.array(out[0].as_cpu()) |
| 511 | |
| 512 | ref = np.load(ref_data_dir + f"/{img_name}.npy") |
| 513 | kwargs = {} |
| 514 | np_dtype = types.to_numpy_type(dtype) |
| 515 | max_val = np_dtype(1.0) if dtype == types.FLOAT else np.iinfo(np_dtype).max |
| 516 | need_scaling = max_val != np_dtype(2**precision - 1) |
| 517 | if need_scaling: |
| 518 | # numpy 2.x computes this division as float32 while numpy 1.x as float64 |
| 519 | # so we need to cast max_val to python float to get the same results |
| 520 | multiplier = float(max_val) / float(2**precision - 1) |
| 521 | ref = ref * multiplier |
| 522 | if dtype != types.FLOAT: |
| 523 | kwargs["atol"] = 0.5 # possible rounding error |
| 524 | np.testing.assert_allclose(ref, result, **kwargs) |
| 525 | |
| 526 | |
| 527 | def test_image_decoder_lossless_jpeg_cpu_not_supported(): |
nothing calls this directly
no test coverage detected