tests.workflow_zero_shot_od_test

  1import logging
  2
  3import fiftyone as fo
  4import pytest
  5from fiftyone.utils.huggingface import load_from_hub
  6
  7import config.config
  8from main import (
  9    configure_logging,
 10    workflow_ensemble_selection,
 11    workflow_zero_shot_object_detection,
 12)
 13
 14
 15@pytest.fixture(autouse=True)
 16def deactivate_wandb_sync():
 17    config.config.WANDB_ACTIVE = False
 18
 19
 20@pytest.fixture(autouse=True)
 21def setup_logging():
 22    configure_logging()
 23
 24
 25@pytest.fixture
 26def dataset_v51():
 27    """Fixture to load a FiftyOne dataset from the hub."""
 28    dataset_name_hub = "Voxel51/fisheye8k"
 29    dataset_name = "fisheye8k_zero_shot_test"
 30    try:
 31        dataset = load_from_hub(
 32            repo_id=dataset_name_hub, max_samples=2, name=dataset_name
 33        )
 34    except:
 35        dataset = fo.load_dataset(dataset_name)
 36    dataset.persistent = True
 37    return dataset
 38
 39
 40def test_zero_shot_inference(dataset_v51):
 41    """Tests zero-shot object detection workflow by running inference on a dataset and validating detection outputs."""
 42
 43    config = {
 44        "n_post_processing_worker_per_inference_worker": 1,
 45        "n_worker_dataloader": 1,
 46        "prefetch_factor_dataloader": 1,
 47        "hf_models_zeroshot_objectdetection": {
 48            "omlab/omdet-turbo-swin-tiny-hf": {
 49                "batch_size": 2,  # Test 2
 50                "n_dataset_chunks": 1,
 51            },
 52            "IDEA-Research/grounding-dino-tiny": {
 53                "batch_size": 1,
 54                "n_dataset_chunks": 1,
 55            },
 56            "google/owlvit-large-patch14": {
 57                "batch_size": 1,
 58                "n_dataset_chunks": 1,
 59            },
 60            "google/owlv2-base-patch16-finetuned": {
 61                "batch_size": 1,
 62                "n_dataset_chunks": 2,  # Test 2
 63            },
 64            "google/owlv2-large-patch14-ensemble": {
 65                "batch_size": 1,
 66                "n_dataset_chunks": 1,
 67            },
 68        },
 69        "detection_threshold": 0.2,
 70        "object_classes": [
 71            "skater",
 72            "child",
 73            "bicycle",
 74            "bicyclist",
 75            "cyclist",
 76            "bike",
 77            "rider",
 78            "motorcycle",
 79            "motorcyclist",
 80            "pedestrian",
 81            "person",
 82            "walker",
 83            "jogger",
 84            "runner",
 85            "skateboarder",
 86            "scooter",
 87            "vehicle",
 88            "car",
 89            "bus",
 90            "truck",
 91            "taxi",
 92            "van",
 93            "pickup truck",
 94            "trailer",
 95            "emergency vehicle",
 96            "delivery driver",
 97        ],
 98    }
 99
100    dataset_info = {
101        "name": "fisheye8k_zero_shot_test",
102        "v51_type": "FiftyOneDataset",
103        "splits": [],
104    }
105
106    fields_of_interest_include = "pred_zsod_"
107    dataset_fields = dataset_v51.get_field_schema()
108    # Cleanup fields prior to run
109    for field in dataset_fields:
110        if fields_of_interest_include in field:
111            dataset_v51.delete_sample_field(field)
112            logging.info(f"Removed dataset field {field} prior to test.")
113
114    workflow_zero_shot_object_detection(dataset_v51, dataset_info, config)
115
116    # Get fields of dataset after run
117    dataset_fields = dataset_v51.get_field_schema()
118    logging.info(f"Dataset fields after run: {dataset_fields}")
119    fields_of_interest = [
120        field for field in dataset_fields if fields_of_interest_include in field
121    ]
122
123    logging.info(f"Zero Shot fields found: {fields_of_interest}")
124    assert (
125        len(fields_of_interest) > 0
126    ), f"No fields that include '{fields_of_interest_include}' found"
127
128    n_detections_found = dict.fromkeys(fields_of_interest, 0)
129
130    for sample in dataset_v51.iter_samples(progress=True, autosave=False):
131        for field in fields_of_interest:
132            detections = sample[field].detections
133            logging.info(detections)
134            n_detections_found[field] += len(detections)
135            for detection in detections:
136                # Test if detection label is in allowed classes
137                assert detection.label in config["object_classes"], (
138                    f"Detection label '{detection.label}' not in configured object classes: "
139                    f"{config['object_classes']}"
140                )
141
142                # Test bbox coordinates (x, y, width, height)
143                bbox = detection.bounding_box
144                assert (
145                    0 <= bbox[0] <= 1 and 0 <= bbox[1] <= 1
146                ), f"Bounding box coordinates (x={bbox[0]}, y={bbox[1]}) must be between 0 and 1"
147                assert (
148                    0 <= bbox[2] <= 1 and 0 <= bbox[3] <= 1
149                ), f"Bounding box dimensions (w={bbox[2]}, h={bbox[3]}) must be between 0 and 1"
150                assert (
151                    bbox[0] + bbox[2] <= 1
152                ), f"Bounding box extends beyond right image border: x({bbox[0]}) + width({bbox[2]}) > 1"
153                assert (
154                    bbox[1] + bbox[3] <= 1
155                ), f"Bounding box extends beyond bottom image border: y({bbox[1]}) + height({bbox[3]}) > 1"
156
157                # Test bbox area is reasonable (not too small or large)
158                bbox_area = bbox[2] * bbox[3]
159                assert (
160                    0 <= bbox_area <= 1
161                ), f"Bounding box area ({bbox_area}) is outside of range [0, 1]"
162
163    # Check if all models detected something at all
164    for field, n_detections in n_detections_found.items():
165        assert n_detections > 0, (
166            f"No detections found for model {field}. "
167            f"Expected at least 1 detection, got {n_detections}"
168        )
169        logging.info(f"Found {n_detections} detections for model {field}")
170
171
172def test_ensemble_selection(dataset_v51):
173    """Integration test for the workflow_ensemble_selection function that verifies detection tagging and unique instance counting."""
174
175    dataset_info = {
176        "name": "fisheye8k_zero_shot_test",
177        "v51_type": "FiftyOneDataset",
178        "splits": [],
179    }
180
181    run_config = {
182        "field_includes": "pred_zsod_",  # V51 field used for detections, "pred_zsod_" default for zero-shot object detection models
183        "agreement_threshold": 3,  # Threshold for n models necessary for agreement between models
184        "iou_threshold": 0.5,  # Threshold for IoU between bboxes to consider them as overlapping
185        "max_bbox_size": 0.01,  # Value between [0,1] for the max size of considered bboxes
186        "positive_classes": [  # Classes to consider, must be subset of available classes in the detections. Example for Vulnerable Road Users.
187            "skater",
188            "child",
189            "bicycle",
190            "bicyclist",
191            "cyclist",
192            "bike",
193            "rider",
194            "motorcycle",
195            "motorcyclist",
196            "pedestrian",
197            "person",
198            "walker",
199            "jogger",
200            "runner",
201            "skateboarder",
202            "scooter",
203            "delivery driver",
204        ],
205    }
206
207    workflow_ensemble_selection(
208        dataset_v51, dataset_info, run_config, wandb_activate=False
209    )
210
211    fields_of_interest_include = "pred_zsod_"
212    dataset_fields = dataset_v51.get_field_schema()
213    fields_of_interest = [
214        field for field in dataset_fields if fields_of_interest_include in field
215    ]
216
217    n_unique_field = "n_unique_ensemble_selection"
218    for sample in dataset_v51.iter_samples(progress=True, autosave=False):
219        # Check if field was populated with unique instances
220        n_unique = sample[n_unique_field]
221        assert n_unique > 0, f"No unique instances found in sample {sample}"
222        n_samples_with_tags = 0
223        # Check if detections were tagged
224        for field in fields_of_interest:
225            detections = sample[field].detections
226            for detection in detections:
227                tags = detection.tags
228                if len(tags) > 0:
229                    n_samples_with_tags += 1
230        assert n_samples_with_tags > 0, "No samples with tags detected"