tests.workflow_class_mapping_test

  1import fiftyone as fo
  2import pytest
  3import config.config
  4from fiftyone import ViewField as F
  5from fiftyone.utils.huggingface import load_from_hub
  6from main import workflow_class_mapping
  7from utils.dataset_loader import load_dataset_info, _post_process_dataset
  8import logging
  9from utils.logging import configure_logging
 10
 11@pytest.fixture(autouse=True)
 12def setup_logging():
 13    configure_logging()
 14
 15@pytest.fixture(autouse=True)
 16def deactivate_wandb_sync():
 17    config.config.WANDB_ACTIVE = False
 18
 19@pytest.fixture
 20def dataset_v51():
 21    """Fixture to load a FiftyOne dataset from the hub with one image to test the Class Mapping workflow."""
 22    dataset_name_hub = "Abeyankar/class_mapping_test_dataset"
 23
 24    dataset = load_from_hub(dataset_name_hub, overwrite=True)
 25    dataset = _post_process_dataset(dataset)
 26
 27    return dataset
 28
 29@pytest.fixture
 30def dataset_v51_2():
 31    """Fixture to load a FiftyOne dataset from the hub with one image from the target dataset."""
 32    dataset_name_hub2 = "Abeyankar/class_mapping_target_test_dataset3"
 33
 34    dataset2 = load_from_hub(dataset_name_hub2, overwrite=True)
 35    dataset2 = _post_process_dataset(dataset2)
 36
 37    return dataset2
 38
 39@pytest.fixture
 40def dataset_v51_3():
 41    """Fixture to load a FiftyOne dataset from the hub with one image from the source dataset."""
 42    dataset_name_hub3 = "Abeyankar/class_mapping_source_test_dataset"
 43
 44    dataset3 = load_from_hub(dataset_name_hub3, overwrite=True)
 45    dataset3 = _post_process_dataset(dataset3)
 46
 47    return dataset3
 48
 49def test_class_mapping(dataset_v51,dataset_v51_2, dataset_v51_3):
 50    """Test Class mapping workflow on the selected sample from source dataset, verifying that each model has added its specific tag"""
 51    # Get the first sample from the dataset.
 52    sample = dataset_v51.first()
 53    assert sample is not None, "Target sample not found in dataset"
 54
 55    logging.info("\nBefore workflow:")
 56    if hasattr(sample, 'ground_truth') and sample["ground_truth"].detections:
 57        logging.info(f"Total number of detections in Sample: {len(sample['ground_truth'].detections)}")
 58
 59    dataset_info = load_dataset_info("fisheye8k")  # Use loader for actual dataset
 60    dataset_info["name"] = "fisheye8k_v51_cm_test"  # Update with test name for local tests where both exist
 61
 62    # Define a simplified local version of WORKFLOWS
 63    config = {
 64        # get the source and target dataset names from datasets.yaml
 65        "dataset_source": "cm_test_source",
 66        "dataset_target": "cm_test_target",
 67        "hf_models_zeroshot_classification": [
 68            "Salesforce/blip2-itm-vit-g",
 69        ],
 70        "thresholds": {
 71            "confidence": 0.2
 72        },
 73        "candidate_labels": {
 74            #Target class(Generalized class) : Source classes(granular categories)
 75            "Car": ["car", "van", "pickup"],
 76            "Truck": ["truck", "pickup"],
 77            #One_to_one_mapping
 78            "Bike" : ["motorbike/cycler"]
 79            #Can add other class mappings in here
 80        },
 81        "change_labels": False
 82    }
 83
 84    models = config["hf_models_zeroshot_classification"]
 85
 86    workflow_class_mapping(dataset_v51, dataset_info, config, wandb_activate=False, test_dataset_source=dataset_v51_3, test_dataset_target=dataset_v51_2)
 87
 88    logging.info("\nAfter workflow:")
 89    if hasattr(sample, "ground_truth") and sample["ground_truth"].detections:
 90        logging.info(f"Total number of detections in Sample: {len(sample.ground_truth.detections)}")
 91
 92    # Gather all tags from all detections in the updated sample
 93    found_tags = set()
 94    if hasattr(sample, "ground_truth") and sample["ground_truth"].detections:
 95        for detection in sample["ground_truth"].detections:
 96            if hasattr(detection, 'tags') and detection.tags:
 97                found_tags.update(detection.tags)
 98
 99    logging.info("\nAll found tags: %s", found_tags)
100
101    # Identify which models did not add tags
102    missing_models = []
103    for model in models:
104        if not any(f"new_class_{model}" in tag for tag in found_tags):
105            missing_models.append(model)
106
107    if missing_models:
108        logging.warning("\nMissing tags for models: %s", missing_models)
109
110    # Validate that each expected model name appears in at least one tag.
111    for model in models:
112        assert any(f"new_class_{model}" in tag for tag in found_tags), (
113            f"Tag for model {model} not found in detections"
114        )