| 283 | |
| 284 | @params(True, False) |
| 285 | def test_coco_include_crowd(include_iscrowd): |
| 286 | @pipeline_def(batch_size=1, device_id=0, num_threads=4) |
| 287 | def coco_pipe(include_iscrowd): |
| 288 | _, boxes, _, image_ids = fn.readers.coco( |
| 289 | file_root=file_root, |
| 290 | annotations_file=train_annotations, |
| 291 | image_ids=True, |
| 292 | include_iscrowd=include_iscrowd, |
| 293 | ) |
| 294 | return boxes, image_ids |
| 295 | |
| 296 | annotations = None |
| 297 | with open(train_annotations) as file: |
| 298 | annotations = json.load(file) |
| 299 | |
| 300 | pipe = coco_pipe(include_iscrowd=include_iscrowd) |
| 301 | number_of_samples = pipe.epoch_size() |
| 302 | for k in number_of_samples: |
| 303 | # there is only one reader |
| 304 | number_of_samples = number_of_samples[k] |
| 305 | break |
| 306 | |
| 307 | anno_mapping = {} |
| 308 | for elm in annotations["annotations"]: |
| 309 | image_id = elm["image_id"] |
| 310 | if not anno_mapping.get(image_id): |
| 311 | anno_mapping[image_id] = {"bbox": [], "iscrowd": []} |
| 312 | anno_mapping[image_id]["bbox"].append(elm["bbox"]) |
| 313 | anno_mapping[image_id]["iscrowd"].append(elm["iscrowd"]) |
| 314 | |
| 315 | all_iscrowd = [] |
| 316 | for _ in range(number_of_samples): |
| 317 | boxes, image_ids = pipe.run() |
| 318 | image_ids = int(image_ids.as_array().item()) |
| 319 | boxes = boxes.as_array()[0] |
| 320 | anno = anno_mapping[image_ids] |
| 321 | idx = 0 |
| 322 | # it assumes that the coco reader reads annotations at the order of appearance inside JSON |
| 323 | all_iscrowd += anno["iscrowd"] |
| 324 | for j, iscrowd in enumerate(anno["iscrowd"]): |
| 325 | if include_iscrowd or iscrowd == 0: |
| 326 | assert np.all(boxes[idx] == np.array(anno["bbox"][j])) |
| 327 | idx += 1 |
| 328 | assert any(all_iscrowd), "At least one annotation should include `iscrowd=1`" |
| 329 | |
| 330 | |
| 331 | def test_coco_empty_annotations_pix(): |