Function _test_stitching

dali/test/python/operator_2/test_resize.py:664–746

(backend, dim, channel_first, dtype, interp)

def _test_stitching(backend, dim, channel_first, dtype, interp):
    batch_size = 1 if dim == 3 else 10
    pipe = dali.pipeline.Pipeline(
        batch_size=batch_size, num_threads=1, device_id=0, seed=1234, prefetch_queue_depth=1
    )
    with pipe:
        if dim == 2:
            files, labels = dali.fn.readers.caffe(path=db_2d_folder, random_shuffle=True)
            images_cpu = dali.fn.decoders.image(files, device="cpu")
        else:
            images_cpu = dali.fn.external_source(source=random_3d_loader(batch_size), layout="DHWC")

        images_hwc = images_cpu if backend_device(backend) == "cpu" else images_cpu.gpu()

        if channel_first:
            images = dali.fn.transpose(
                images_hwc, perm=[3, 0, 1, 2] if dim == 3 else [2, 0, 1], transpose_layout=True
            )
        else:
            images = images_hwc

        out_size_full = [32, 32, 32] if dim == 3 else [160, 160]
        out_size_half = [x // 2 for x in out_size_full]

        roi_start = [0] * dim
        roi_end = [1] * dim

        resized = resize_op(backend)(
            images, dtype=dtype, min_filter=interp, mag_filter=interp, size=out_size_full
        )

        outputs = [resized]

        for z in range(dim - 1):
            if dim == 3:
                roi_start[0] = z * 0.5
                roi_end[0] = (z + 1) * 0.5
            for y in [0, 1]:
                roi_start[-2] = y * 0.5
                roi_end[-2] = (y + 1) * 0.5
                for x in [0, 1]:
                    roi_start[-1] = x * 0.5
                    roi_end[-1] = (x + 1) * 0.5

                    part = fn.resize(
                        images,
                        dtype=dtype,
                        interp_type=interp,
                        size=out_size_half,
                        roi_start=roi_start,
                        roi_end=roi_end,
                        roi_relative=True,
                    )
                    outputs.append(part)

        pipe.set_outputs(*outputs)

    for iter in range(1):