()
| 130 | |
| 131 | |
| 132 | def test_runtime_stride_dim1(): |
| 133 | def data_gen(): |
| 134 | return [ |
| 135 | np.arange(12, dtype=np.float32).reshape(4, 3), |
| 136 | np.arange(20, dtype=np.float32).reshape(4, 5), |
| 137 | ] |
| 138 | |
| 139 | src = fn.external_source(data_gen) |
| 140 | strides = [np.array(x, dtype=np.int64) for x in [1, 2, -1, -2, -5]] |
| 141 | stride = fn.external_source(source=strides, batch=False, cycle=True) |
| 142 | pipe = index_pipe(src, lambda x: x[::stride]) |
| 143 | |
| 144 | j = 0 |
| 145 | for _ in range(4): |
| 146 | inp, cpu, gpu = tuple(out.as_cpu() for out in pipe.run()) |
| 147 | for i in range(len(inp)): |
| 148 | x = inp.at(i) |
| 149 | # fmt: off |
| 150 | ref = x[::strides[j]] |
| 151 | # fmt: on |
| 152 | assert np.array_equal(ref, cpu.at(i)) |
| 153 | assert np.array_equal(ref, gpu.at(i)) |
| 154 | j = (j + 1) % len(strides) |
| 155 | |
| 156 | |
| 157 | def test_runtime_stride_dim2(): |
nothing calls this directly
no test coverage detected