test: anomalib models tests

src/model_api/models/image_model.py

@@ -121,10 +121,12 @@ def parameters(cls) -> dict[str, Any]:
                 "orig_height": NumericalValue(
                     int,
                     description="Model input height before embedding processing",
+                    default_value=None,
                 ),
                 "orig_width": NumericalValue(
                     int,
                     description="Model input width before embedding processing",
+                    default_value=None,
                 ),
                 "pad_value": NumericalValue(
                     int,

src/model_api/models/types.py

@@ -62,6 +62,10 @@ def __init__(
         self.value_type = value_type
 
     def from_str(self, value: str) -> Any:
+        if not value and self.default_value is not None:
+            return self.default_value
+        if not value and self.default_value is None:
+            return None
         return self.value_type(value)
 
     def validate(self, value):

tests/accuracy/prepare_data.py

@@ -53,6 +53,21 @@ async def download_otx_model(client, otx_models_dir, model_name, format="xml"):
         )
 
 
+async def download_anomalib_model(client, models_dir, model_name):
+    await asyncio.gather(
+        stream_file(
+            client,
+            f"https://storage.geti.intel.com/geti_predict/test/anomalib_models/{model_name}/openvino.xml",
+            f"{models_dir}/{model_name}.xml",
+        ),
+        stream_file(
+            client,
+            f"https://storage.geti.intel.com/geti_predict/test/anomalib_models/{model_name}/openvino.bin",
+            f"{models_dir}/{model_name}.bin",
+        ),
+    )
+
+
 async def main():
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -66,6 +81,8 @@ async def main():
 
     otx_models_dir = args.data_dir / "otx_models"
     otx_models_dir.mkdir(parents=True, exist_ok=True)
+    anomalib_models_dir = args.data_dir / "anomalib_models"
+    anomalib_models_dir.mkdir(parents=True, exist_ok=True)
     async with httpx.AsyncClient(timeout=20.0) as client:
         await asyncio.gather(
             download_images(args.data_dir),
@@ -138,6 +155,9 @@ async def main():
             download_otx_model(client, otx_models_dir, "sam_vit_b_zsl_decoder"),
             download_otx_model(client, otx_models_dir, "rtmpose_tiny"),
             download_otx_model(client, otx_models_dir, "segnext_t_tiling"),
+            download_anomalib_model(client, anomalib_models_dir, "padim"),
+            download_anomalib_model(client, anomalib_models_dir, "stfpm"),
+            download_anomalib_model(client, anomalib_models_dir, "uflow"),
         )
 
 

tests/accuracy/public_scope.json

@@ -441,5 +441,41 @@
         ]
       }
     ]
+  },
+  {
+    "name": "anomalib_models/padim.xml",
+    "type": "AnomalyDetection",
+    "test_data": [
+      {
+        "image": "coco128/images/train2017/000000000074.jpg",
+        "reference": [
+          "anomaly_map min:153 max:255;pred_score:1.0;pred_label:True;pred_mask min:1 max:1;"
+        ]
+      }
+    ]
+  },
+  {
+    "name": "anomalib_models/stfpm.xml",
+    "type": "AnomalyDetection",
+    "test_data": [
+      {
+        "image": "coco128/images/train2017/000000000074.jpg",
+        "reference": [
+          "anomaly_map min:127 max:255;pred_score:1.0;pred_label:True;pred_mask min:0 max:1;"
+        ]
+      }
+    ]
+  },
+  {
+    "name": "anomalib_models/uflow.xml",
+    "type": "AnomalyDetection",
+    "test_data": [
+      {
+        "image": "coco128/images/train2017/000000000074.jpg",
+        "reference": [
+          "anomaly_map min:191 max:191;pred_score:1.0;pred_label:True;pred_mask min:1 max:1;"
+        ]
+      }
+    ]
   }
 ]

tests/accuracy/test_accuracy.py

@@ -95,12 +95,14 @@ def create_models(model_type, model_path, download_dir, force_onnx_adapter=False
         model_type.create_model(model_path, device="CPU", download_dir=download_dir),
     ]
     if model_path.endswith(".xml"):
-        wrapper_type = model_type.get_model_class(
-            create_core().read_model(model_path).get_rt_info(["model_info", "model_type"]).astype(str),
-        )
-        model = wrapper_type(OpenvinoAdapter(create_core(), model_path, device="CPU"))
-        model.load()
-        models.append(model)
+        model = create_core().read_model(model_path)
+        if model.has_rt_info(["model_info", "model_type"]):
+            wrapper_type = model_type.get_model_class(
+                create_core().read_model(model_path).get_rt_info(["model_info", "model_type"]).astype(str),
+            )
+            model = wrapper_type(OpenvinoAdapter(create_core(), model_path, device="CPU"))
+            model.load()
+            models.append(model)
     return models
 
 

tests/functional/test_save.py

@@ -70,3 +70,45 @@ def test_onnx_save(tmp_path, data):
     assert type(cls_model) is type(deserialized)
     for attr in cls_model.parameters():
         assert getattr(cls_model, attr) == getattr(deserialized, attr)
+
+
+def test_padim_save(tmp_path, data):
+    padim_model = Model.create_model(
+        Path(data) / "anomalib_models/padim.xml",
+    )
+    xml_path = str(tmp_path / "a.xml")
+    padim_model.save(xml_path)
+    deserialized = Model.create_model(xml_path)
+
+    assert not deserialized.get_model().get_rt_info(["model_info", "embedded_processing"]).astype(bool)
+    assert type(padim_model) is type(deserialized)
+    for attr in padim_model.parameters():
+        assert getattr(padim_model, attr) == getattr(deserialized, attr)
+
+
+def test_stfpm_save(tmp_path, data):
+    stfpm_model = Model.create_model(
+        Path(data) / "anomalib_models/stfpm.xml",
+    )
+    xml_path = str(tmp_path / "a.xml")
+    stfpm_model.save(xml_path)
+    deserialized = Model.create_model(xml_path)
+
+    assert not deserialized.get_model().get_rt_info(["model_info", "embedded_processing"]).astype(bool)
+    assert type(stfpm_model) is type(deserialized)
+    for attr in stfpm_model.parameters():
+        assert getattr(stfpm_model, attr) == getattr(deserialized, attr)
+
+
+def test_uflow_save(tmp_path, data):
+    uflow_model = Model.create_model(
+        Path(data) / "anomalib_models/uflow.xml",
+    )
+    xml_path = str(tmp_path / "a.xml")
+    uflow_model.save(xml_path)
+    deserialized = Model.create_model(xml_path)
+
+    assert not deserialized.get_model().get_rt_info(["model_info", "embedded_processing"]).astype(bool)
+    assert type(uflow_model) is type(deserialized)
+    for attr in uflow_model.parameters():
+        assert getattr(uflow_model, attr) == getattr(deserialized, attr)

tests/precommit/prepare_data.py

@@ -28,16 +28,30 @@ def retrieve_otx_model(data_dir, model_name, format="xml"):
         )
 
 
+def retrieve_anomalib_model(data_dir, model_name, format="xml"):
+    destination_folder = Path(data_dir) / "anomalib_models"
+    destination_folder.mkdir(parents=True, exist_ok=True)
+    urlretrieve(
+        f"https://storage.geti.intel.com/geti_predict/test/anomalib_models/{model_name}/openvino.xml",
+        destination_folder / f"{model_name}.xml",
+    )
+    urlretrieve(
+        f"https://storage.geti.intel.com/geti_predict/test/anomalib_models/{model_name}/openvino.bin",
+        f"{destination_folder}/{model_name}.bin",
+    )
+
+
 def prepare_model(
     data_dir="./data",
     public_scope=Path(__file__).resolve().parent / "public_scope.json",
 ):
     # TODO refactor this test so that it does not use eval
     # flake8: noqa: F401
-    from model_api.models import ClassificationModel, DetectionModel, SegmentationModel
+    from model_api.models import AnomalyDetection, ClassificationModel, DetectionModel, SegmentationModel
 
     # Mapping of model type strings to actual classes for security
     MODEL_TYPE_MAPPING = {
+        "AnomalyDetection": AnomalyDetection,
         "ClassificationModel": ClassificationModel,
         "DetectionModel": DetectionModel,
         "SegmentationModel": SegmentationModel,
@@ -91,3 +105,6 @@ def prepare_data(data_dir="./data"):
     retrieve_otx_model(args.data_dir, "Lite-hrnet-18_mod2")
     retrieve_otx_model(args.data_dir, "tinynet_imagenet")
     retrieve_otx_model(args.data_dir, "cls_mobilenetv3_large_cars", "onnx")
+    retrieve_anomalib_model(args.data_dir, "padim")
+    retrieve_anomalib_model(args.data_dir, "stfpm")
+    retrieve_anomalib_model(args.data_dir, "uflow")

tests/precommit/public_scope.json

@@ -14,5 +14,17 @@
   {
     "name": "otx_models/tinynet_imagenet.xml",
     "type": "ClassificationModel"
+  },
+  {
+    "name": "anomalib_models/padim.xml",
+    "type": "AnomalyDetection"
+  },
+  {
+    "name": "anomalib_models/stfpm.xml",
+    "type": "AnomalyDetection"
+  },
+  {
+    "name": "anomalib_models/uflow.xml",
+    "type": "AnomalyDetection"
   }
 ]