chore: enable checking of styles in examples, tests

examples/asynchronous_api/run.py

@@ -7,16 +7,19 @@
 import sys
 
 import cv2
+
 from model_api.models import DetectionModel
 
 
 def main():
     if len(sys.argv) != 2:
-        raise RuntimeError(f"Usage: {sys.argv[0]} <path_to_image>")
+        usage_message = f"Usage: {sys.argv[0]} <path_to_image>"
+        raise RuntimeError(usage_message)
 
     image = cv2.cvtColor(cv2.imread(sys.argv[1]), cv2.COLOR_BGR2RGB)
     if image is None:
-        raise RuntimeError("Failed to read the image")
+        error_message = f"Failed to read the image: {sys.argv[1]}"
+        raise RuntimeError(error_message)
 
     # Create Object Detection model using mode name and download from Open Model Zoo
     # Replace numpy preprocessing and embed it directly into a model graph to speed up inference

examples/serving_api/run.py

@@ -7,20 +7,24 @@
 import sys
 
 import cv2
+
 from model_api.models import DetectionModel
 
 
 def main():
     if len(sys.argv) != 2:
-        raise RuntimeError(f"Usage: {sys.argv[0]} <path_to_image>")
+        usage_message = f"Usage: {sys.argv[0]} <path_to_image>"
+        raise RuntimeError(usage_message)
 
     image = cv2.cvtColor(cv2.imread(sys.argv[1]), cv2.COLOR_BGR2RGB)
     if image is None:
-        raise RuntimeError("Failed to read the image")
+        error_message = f"Failed to read the image: {sys.argv[1]}"
+        raise RuntimeError(error_message)
 
     # Create Object Detection model specifying the OVMS server URL
     model = DetectionModel.create_model(
-        "localhost:8000/v2/models/ssd_mobilenet_v1_fpn_coco", model_type="ssd"
+        "localhost:8000/v2/models/ssd_mobilenet_v1_fpn_coco",
+        model_type="ssd",
     )
     detections = model(image)
     print(f"Detection results: {detections}")

examples/synchronous_api/run.py

@@ -7,17 +7,20 @@
 import sys
 
 import cv2
-from model_api.models import ClassificationModel, DetectionModel, SegmentationModel
 from PIL import Image
 
+from model_api.models import ClassificationModel, DetectionModel, SegmentationModel
+
 
 def main():
     if len(sys.argv) != 2:
-        raise RuntimeError(f"Usage: {sys.argv[0]} <path_to_image>")
+        usage_message = f"Usage: {sys.argv[0]} <path_to_image>"
+        raise RuntimeError(usage_message)
 
     image = cv2.cvtColor(cv2.imread(sys.argv[1]), cv2.COLOR_BGR2RGB)
     if image is None:
-        raise RuntimeError("Failed to read the image")
+        error_message = f"Failed to read the image: {sys.argv[1]}"
+        raise RuntimeError(error_message)
 
     # Create Image Classification model using mode name and download from Open Model Zoo
     efficientnet_b0 = ClassificationModel.create_model("efficientnet-b0-pytorch")
@@ -37,7 +40,7 @@ def main():
 
     # Instantiate from a local model (downloaded previously)
     ssd_mobilenet_fpn_local = DetectionModel.create_model(
-        "tmp/public/ssd_mobilenet_v1_fpn_coco/FP16/ssd_mobilenet_v1_fpn_coco.xml"
+        "tmp/public/ssd_mobilenet_v1_fpn_coco/FP16/ssd_mobilenet_v1_fpn_coco.xml",
     )
     detections = ssd_mobilenet_fpn_local(image)
     print(f"Detection results for local: {detections}")

examples/visual_prompting/run.py

@@ -6,15 +6,17 @@
 
 import argparse
 import colorsys
+from itertools import starmap
 
 import cv2
 import numpy as np
+
 from model_api.models import Model, Prompt, SAMVisualPrompter
 
 
 def get_colors(n: int):
     HSV_tuples = [(x / n, 0.5, 0.5) for x in range(n)]
-    RGB_tuples = map(lambda x: colorsys.hsv_to_rgb(*x), HSV_tuples)
+    RGB_tuples = starmap(colorsys.hsv_to_rgb, HSV_tuples)
     return (np.array(list(RGB_tuples)) * 255).astype(np.uint8)
 
 
@@ -28,7 +30,8 @@ def main():
 
     image = cv2.cvtColor(cv2.imread(args.image), cv2.COLOR_BGR2RGB)
     if image is None:
-        raise RuntimeError("Failed to read the image")
+        error_message = f"Failed to read the image: {args.image}"
+        raise RuntimeError(error_message)
 
     encoder = Model.create_model(args.encoder_path)
     decoder = Model.create_model(args.decoder_path)

examples/zsl_visual_prompting/run.py

@@ -6,15 +6,17 @@
 
 import argparse
 import colorsys
+from itertools import starmap
 
 import cv2
 import numpy as np
+
 from model_api.models import Model, Prompt, SAMLearnableVisualPrompter
 
 
 def get_colors(n: int):
     HSV_tuples = [(x / n, 0.5, 0.5) for x in range(n)]
-    RGB_tuples = map(lambda x: colorsys.hsv_to_rgb(*x), HSV_tuples)
+    RGB_tuples = starmap(colorsys.hsv_to_rgb, HSV_tuples)
     return (np.array(list(RGB_tuples)) * 255).astype(np.uint8)
 
 
@@ -30,16 +32,20 @@ def main():
 
     image = cv2.cvtColor(cv2.imread(args.image_source), cv2.COLOR_BGR2RGB)
     if image is None:
-        raise RuntimeError("Failed to read the source image")
+        error_message = f"Failed to read the source image: {args.image_source}"
+        raise RuntimeError(error_message)
 
     image_target = cv2.cvtColor(cv2.imread(args.image_target), cv2.COLOR_BGR2RGB)
     if image_target is None:
-        raise RuntimeError("Failed to read the target image")
+        error_message = f"Failed to read the target image: {args.image_target}"
+        raise RuntimeError(error_message)
 
     encoder = Model.create_model(args.encoder_path)
     decoder = Model.create_model(args.decoder_path)
     zsl_sam_prompter = SAMLearnableVisualPrompter(
-        encoder, decoder, threshold=args.threshold
+        encoder,
+        decoder,
+        threshold=args.threshold,
     )
 
     all_prompts = []
@@ -61,7 +67,11 @@ def main():
             image_target = cv2.addWeighted(image_target, 0.2, masked_img, 0.8, 0)
             print(f"Reference point: {prompt_point}, point score: {confidence:.3f}")
             cv2.circle(
-                image_target, prompt_point, radius=0, color=(0, 0, 255), thickness=5
+                image_target,
+                prompt_point,
+                radius=0,
+                color=(0, 0, 255),
+                thickness=5,
             )
 
     cv2.imwrite("zsl_sam_result.jpg", cv2.cvtColor(image_target, cv2.COLOR_RGB2BGR))

pyproject.toml

@@ -217,9 +217,7 @@ exclude = [
     "buck-out",
     "build",
     "dist",
-    "examples",
     "node_modules",
-    "tests",
     "venv",
 ]
 

tests/accuracy/conftest.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 import json
+from pathlib import Path
 
 import pytest
 
@@ -28,5 +29,5 @@ def pytest_runtest_makereport(item, call):
 
     if result.when == "call":
         test_results = item.config.test_results
-        with open("test_scope.json", "w") as outfile:
+        with Path("test_scope.json").open("w") as outfile:
             json.dump(test_results, outfile, indent=4)

tests/accuracy/prepare_data.py

@@ -18,15 +18,15 @@ async def download_images(data_dir):
         with ZipFile(BytesIO(archive.content)) as zfile:
             zfile.extractall(data_dir)
         image = await client.get(
-            "https://raw.githubusercontent.com/Shenggan/BCCD_Dataset/master/BCCD/JPEGImages/BloodImage_00007.jpg"
+            "https://raw.githubusercontent.com/Shenggan/BCCD_Dataset/master/BCCD/JPEGImages/BloodImage_00007.jpg",
         )
-        with open(data_dir / "BloodImage_00007.jpg", "wb") as im:
+        with Path(data_dir / "BloodImage_00007.jpg").open("wb") as im:
             im.write(image.content)
 
 
 async def stream_file(client, url, filename):
     async with client.stream("GET", url) as stream:
-        with open(filename, "wb") as file:
+        with Path(filename).open("wb") as file:
             async for data in stream.aiter_bytes():
                 file.write(data)
 
@@ -74,11 +74,17 @@ async def main():
             download_otx_model(client, otx_models_dir, "mlc_efficient_v2s_voc"),
             download_otx_model(client, otx_models_dir, "det_mobilenetv2_atss_bccd"),
             download_otx_model(
-                client, otx_models_dir, "det_mobilenetv2_atss_bccd_onnx", "onnx"
+                client,
+                otx_models_dir,
+                "det_mobilenetv2_atss_bccd_onnx",
+                "onnx",
             ),
             download_otx_model(client, otx_models_dir, "cls_mobilenetv3_large_cars"),
             download_otx_model(
-                client, otx_models_dir, "cls_mobilenetv3_large_cars", "onnx"
+                client,
+                otx_models_dir,
+                "cls_mobilenetv3_large_cars",
+                "onnx",
             ),
             download_otx_model(client, otx_models_dir, "cls_efficient_b0_cars"),
             download_otx_model(client, otx_models_dir, "cls_efficient_v2s_cars"),
@@ -88,7 +94,9 @@ async def main():
             download_otx_model(client, otx_models_dir, "Lite-hrnet-s_mod2", "onnx"),
             download_otx_model(client, otx_models_dir, "Lite-hrnet-x-mod3"),
             download_otx_model(
-                client, otx_models_dir, "is_efficientnetb2b_maskrcnn_coco_reduced"
+                client,
+                otx_models_dir,
+                "is_efficientnetb2b_maskrcnn_coco_reduced",
             ),
             download_otx_model(
                 client,
@@ -97,15 +105,21 @@ async def main():
                 "onnx",
             ),
             download_otx_model(
-                client, otx_models_dir, "is_resnet50_maskrcnn_coco_reduced"
+                client,
+                otx_models_dir,
+                "is_resnet50_maskrcnn_coco_reduced",
             ),
             download_otx_model(client, otx_models_dir, "mobilenet_v3_large_hc_cf"),
             download_otx_model(
-                client, otx_models_dir, "classification_model_with_xai_head"
+                client,
+                otx_models_dir,
+                "classification_model_with_xai_head",
             ),
             download_otx_model(client, otx_models_dir, "detection_model_with_xai_head"),
             download_otx_model(
-                client, otx_models_dir, "segmentation_model_with_xai_head"
+                client,
+                otx_models_dir,
+                "segmentation_model_with_xai_head",
             ),
             download_otx_model(client, otx_models_dir, "maskrcnn_model_with_xai_head"),
             download_otx_model(client, otx_models_dir, "maskrcnn_xai_tiling"),
@@ -114,7 +128,9 @@ async def main():
             download_otx_model(client, otx_models_dir, "anomaly_stfpm_bottle_mvtec"),
             download_otx_model(client, otx_models_dir, "deit-tiny"),
             download_otx_model(
-                client, otx_models_dir, "cls_efficient_b0_shuffled_outputs"
+                client,
+                otx_models_dir,
+                "cls_efficient_b0_shuffled_outputs",
             ),
             download_otx_model(client, otx_models_dir, "action_cls_xd3_kinetic"),
             download_otx_model(client, otx_models_dir, "sam_vit_b_zsl_encoder"),

tests/accuracy/test_YOLOv8.py

@@ -13,10 +13,11 @@
 import pytest
 import torch
 import ultralytics
-from model_api.models import YOLOv5
 from ultralytics.data import utils
 from ultralytics.models import yolo
 
+from model_api.models import YOLOv5
+
 
 def _init_predictor(yolo):
     yolo.predict(np.empty([1, 1, 3], np.uint8))
@@ -26,15 +27,17 @@ def _init_predictor(yolo):
 def _cached_alignment(pt):
     export_dir = Path(
         ultralytics.YOLO(
-            Path(os.environ["DATA"]) / "ultralytics" / pt, "detect"
-        ).export(format="openvino", half=True)
+            Path(os.environ["DATA"]) / "ultralytics" / pt,
+            "detect",
+        ).export(format="openvino", half=True),
     )
     impl_wrapper = YOLOv5.create_model(export_dir / (pt.stem + ".xml"), device="CPU")
     ref_wrapper = ultralytics.YOLO(export_dir, "detect")
     ref_wrapper.overrides["imgsz"] = (impl_wrapper.w, impl_wrapper.h)
     _init_predictor(ref_wrapper)
     ref_wrapper.predictor.model.ov_compiled_model = ov.Core().compile_model(
-        ref_wrapper.predictor.model.ov_model, "CPU"
+        ref_wrapper.predictor.model.ov_model,
+        "CPU",
     )
     ref_dir = export_dir / "ref"
     ref_dir.mkdir(exist_ok=True)
@@ -60,9 +63,8 @@ def _impaths():
         }
     )
     if not impaths:
-        raise RuntimeError(
-            f"{Path(os.environ['DATA']) / 'coco128/images/train2017/'} is empty"
-        )
+        error_message = f"{Path(os.environ['DATA']) / 'coco128/images/train2017/'} is empty"
+        raise RuntimeError(error_message)
     return impaths
 
 
@@ -77,7 +79,7 @@ def test_alignment(impath, pt):
     pred_scores = impl_preds.scores.astype(np.float32)
     pred_labels = impl_preds.labels
     ref_predictions = ref_wrapper.predict(im)
-    assert 1 == len(ref_predictions)
+    assert len(ref_predictions) == 1
     ref_boxes = ref_predictions[0].boxes.data.numpy()
     if 0 == len(pred_boxes) == len(ref_boxes):
         return  # np.isclose() doesn't work for empty arrays
@@ -86,7 +88,7 @@ def test_alignment(impath, pt):
     assert np.isclose(pred_boxes, ref_boxes[:, :4], 0, 1).all()
     assert np.isclose(pred_scores, ref_boxes[:, 4], 0.0, 0.02).all()
     assert (pred_labels == ref_boxes[:, 5]).all()
-    with open(ref_dir / impath.with_suffix(".txt").name, "w") as file:
+    with Path.open(ref_dir / impath.with_suffix(".txt").name, "w") as file:
         print(impl_preds, end="", file=file)
 
 
@@ -108,20 +110,22 @@ def evaluate(self, wrapper):
         )
         self.init_metrics(
             types.SimpleNamespace(
-                names={idx: label for idx, label in enumerate(wrapper.labels)}
-            )
+                names=dict(enumerate(wrapper.labels)),
+            ),
         )
         for batch in dataloader:
             im = cv2.imread(batch["im_file"][0])
             result = wrapper(im)
             bboxes = torch.from_numpy(result.bboxes) / torch.tile(
-                torch.tensor([im.shape[1], im.shape[0]], dtype=torch.float32), (1, 2)
+                torch.tensor([im.shape[1], im.shape[0]], dtype=torch.float32),
+                (1, 2),
             )
             scores = torch.from_numpy(result.scores)
             labels = torch.from_numpy(result.labels)
 
             pred = torch.cat(
-                [bboxes, scores[:, None], labels[:, None].float()], dim=1
+                [bboxes, scores[:, None], labels[:, None].float()],
+                dim=1,
             ).unsqueeze(0)
             if not pred.numel():
                 pred = pred.view(1, 0, 6)
@@ -130,7 +134,7 @@ def evaluate(self, wrapper):
 
 
 @pytest.mark.parametrize(
-    "pt,ref_mAP50_95",
+    ("pt", "ref_mAP50_95"),
     [
         (
             Path("yolov8n.pt"),
@@ -146,13 +150,12 @@ def test_metric(pt, ref_mAP50_95):
     mAP50_95 = Metrics().evaluate(
         YOLOv5.create_model(
             ultralytics.YOLO(
-                Path(os.environ["DATA"]) / "ultralytics" / pt, "detect"
+                Path(os.environ["DATA"]) / "ultralytics" / pt,
+                "detect",
             ).export(format="openvino", half=True)
             / pt.with_suffix(".xml"),
             device="CPU",
             configuration={"confidence_threshold": 0.001},
-        )
+        ),
     )["metrics/mAP50-95(B)"]
-    assert (
-        abs(mAP50_95 - ref_mAP50_95) <= 0.01 * ref_mAP50_95 or mAP50_95 >= ref_mAP50_95
-    )
+    assert abs(mAP50_95 - ref_mAP50_95) <= 0.01 * ref_mAP50_95 or mAP50_95 >= ref_mAP50_95