()
| 155 | |
| 156 | |
| 157 | def test_runtime_stride_dim2(): |
| 158 | def data_gen(): |
| 159 | return [ |
| 160 | np.arange(12, dtype=np.float32).reshape(4, 3), |
| 161 | np.arange(20, dtype=np.float32).reshape(4, 5), |
| 162 | ] |
| 163 | |
| 164 | src = fn.external_source(data_gen) |
| 165 | strides = [np.array(x, dtype=np.int64) for x in [1, 2, -1, -2, -5]] |
| 166 | stride = fn.external_source(source=strides, batch=False, cycle=True) |
| 167 | pipe = index_pipe(src, lambda x: x[:, ::stride]) |
| 168 | |
| 169 | j = 0 |
| 170 | for _ in range(4): |
| 171 | inp, cpu, gpu = tuple(out.as_cpu() for out in pipe.run()) |
| 172 | for i in range(len(inp)): |
| 173 | x = inp.at(i) |
| 174 | # fmt: off |
| 175 | ref = x[:, ::strides[j]] |
| 176 | # fmt: on |
| 177 | assert np.array_equal(ref, cpu.at(i)) |
| 178 | assert np.array_equal(ref, gpu.at(i)) |
| 179 | j = (j + 1) % len(strides) |
| 180 | |
| 181 | |
| 182 | def test_new_axis(): |
nothing calls this directly
no test coverage detected