| 194 | @raises(RuntimeError, "Expected data numpy size ") |
| 195 | @params(True, False) |
| 196 | def test_malformed_data(truncate): |
| 197 | def malformed_source(): |
| 198 | data = np.empty(100, dtype=np.uint8) |
| 199 | buff = io.BytesIO() |
| 200 | np.save(buff, data, allow_pickle=False) |
| 201 | buff.seek(0) |
| 202 | raw = np.frombuffer(buff.read(), dtype=np.uint8) |
| 203 | if truncate: |
| 204 | raw = raw[:-100] # Truncate the data |
| 205 | else: |
| 206 | raw = np.concatenate((raw, np.empty(100, dtype=np.uint8))) # Add extra data |
| 207 | yield (raw,) |
| 208 | |
| 209 | @pipeline_def(batch_size=1, num_threads=1) |
| 210 | def pipe(): |
| 211 | encoded_npy = fn.external_source( |
| 212 | source=malformed_source(), num_outputs=1, batch=False, ndim=1, dtype=DALIDataType.UINT8 |
| 213 | ) |
| 214 | decoded = fn.decoders.numpy(encoded_npy) |
| 215 | return decoded |
| 216 | |
| 217 | p = pipe() |
| 218 | p.build() |
| 219 | p.run() |