✨Add MyPy

.pre-commit-config.yaml

@@ -23,6 +23,13 @@ repos:
       # Run the formatter
       - id: ruff-format
 
+  # python static type checking
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: "v1.11.2"
+    hooks:
+      - id: mypy
+        additional_dependencies: [types-PyYAML, types-setuptools]
+
   - repo: https://github.com/pre-commit/mirrors-prettier
     rev: v4.0.0-alpha.8
     hooks:

docs/source/conf.py

@@ -42,7 +42,7 @@
 ]
 
 templates_path = ["_templates"]
-exclude_patterns = []
+exclude_patterns: list[str] = []
 
 # Automatic exclusion of prompts from the copies
 # https://sphinx-copybutton.readthedocs.io/en/latest/use.html#automatic-exclusion-of-prompts-from-the-copies

examples/python/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

model_api/python/model_api/adapters/inference_adapter.py

@@ -3,8 +3,9 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-import abc
+from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
+from typing import Any, Dict, List, Set, Tuple
 
 
 @dataclass
@@ -17,30 +18,37 @@ class Metadata:
     meta: dict = field(default_factory=dict)
 
 
-class InferenceAdapter(abc.ABC):
-    """An abstract Model Adapter with the following interface:
-
-    - Reading the model from disk or other place
-    - Loading the model to the device
-    - Accessing the information about inputs/outputs
-    - The model reshaping
-    - Synchronous model inference
-    - Asynchronous model inference
+class InferenceAdapter(ABC):
+    """
+    An abstract Model Adapter with the following interface:
+
+        - Reading the model from disk or other place
+        - Loading the model to the device
+        - Accessing the information about inputs/outputs
+        - The model reshaping
+        - Synchronous model inference
+        - Asynchronous model inference
     """
 
     precisions = ("FP32", "I32", "FP16", "I16", "I8", "U8")
 
-    @abc.abstractmethod
-    def __init__(self):
-        """An abstract Model Adapter constructor.
+    @abstractmethod
+    def __init__(self) -> None:
+        """
+        An abstract Model Adapter constructor.
         Reads the model from disk or other place.
         """
+        self.model: Any
 
-    @abc.abstractmethod
+    @abstractmethod
     def load_model(self):
         """Loads the model on the device."""
 
-    @abc.abstractmethod
+    @abstractmethod
+    def get_model(self):
+        """Get the model."""
+
+    @abstractmethod
     def get_input_layers(self):
         """Gets the names of model inputs and for each one creates the Metadata structure,
            which contains the information about the input shape, layout, precision
@@ -50,7 +58,7 @@ def get_input_layers(self):
             - the dict containing Metadata for all inputs
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def get_output_layers(self):
         """Gets the names of model outputs and for each one creates the Metadata structure,
            which contains the information about the output shape, layout, precision
@@ -60,7 +68,7 @@ def get_output_layers(self):
             - the dict containing Metadata for all outputs
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def reshape_model(self, new_shape):
         """Reshapes the model inputs to fit the new input shape.
 
@@ -74,7 +82,7 @@ def reshape_model(self, new_shape):
                 }
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def infer_sync(self, dict_data):
         """Performs the synchronous model inference. The infer is a blocking method.
 
@@ -95,9 +103,10 @@ def infer_sync(self, dict_data):
                 }
         """
 
-    @abc.abstractmethod
-    def infer_async(self, dict_data, callback_fn, callback_data):
-        """Performs the asynchronous model inference and sets
+    @abstractmethod
+    def infer_async(self, dict_data, callback_data):
+        """
+        Performs the asynchronous model inference and sets
         the callback for inference completion. Also, it should
         define get_raw_result() function, which handles the result
         of inference from the model.
@@ -109,11 +118,10 @@ def infer_async(self, dict_data, callback_fn, callback_data):
                     'input_layer_name_2': data_2,
                     ...
                 }
-            - callback_fn: the callback function, which is defined outside the adapter
             - callback_data: the data for callback, that will be taken after the model inference is ended
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def is_ready(self):
         """In case of asynchronous execution checks if one can submit input data
         to the model for inference, or all infer requests are busy.
@@ -123,23 +131,23 @@ def is_ready(self):
                 submitted to the model for inference or not
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def await_all(self):
         """In case of asynchronous execution waits the completion of all
         busy infer requests.
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def await_any(self):
         """In case of asynchronous execution waits the completion of any
         busy infer request until it becomes available for the data submission.
         """
 
-    @abc.abstractmethod
+    @abstractmethod
     def get_rt_info(self, path):
         """Forwards to openvino.Model.get_rt_info(path)"""
 
-    @abc.abstractmethod
+    @abstractmethod
     def embed_preprocessing(
         self,
         layout,

model_api/python/model_api/adapters/onnx_adapter.py

@@ -162,6 +162,10 @@ def embed_preprocessing(
             reversed(preproc_funcs),
         )
 
+    def get_model(self):
+        """Return the reference to the ONNXRuntime session."""
+        return self.session
+
     def reshape_model(self, new_shape):
         raise NotImplementedError
 

model_api/python/model_api/adapters/ovms_adapter.py

@@ -87,6 +87,10 @@ def is_ready(self):
     def load_model(self):
         pass
 
+    def get_model(self):
+        """Return the reference to the GrpcClient."""
+        return self.client
+
     def await_all(self):
         pass
 

model_api/python/model_api/adapters/utils.py

@@ -7,6 +7,7 @@
 
 import math
 from functools import partial
+from typing import Callable, Optional
 
 import cv2
 import numpy as np
@@ -509,7 +510,7 @@ def crop_resize_ocv(image, size):
     return cv2.resize(cropped_frame, size)
 
 
-RESIZE_TYPES = {
+RESIZE_TYPES: dict[str, Callable] = {
     "crop": crop_resize_ocv,
     "standard": resize_image_ocv,
     "fit_to_window": resize_image_with_aspect_ocv,

model_api/python/model_api/models/action_classification.py

@@ -64,6 +64,14 @@ def __init__(
         self.image_blob_names = self._get_inputs()
         self.image_blob_name = self.image_blob_names[0]
         self.nscthw_layout = "NSCTHW" in self.inputs[self.image_blob_name].layout
+        self.labels: list[str]
+        self.path_to_labels: str
+        self.mean_values: list[int | float]
+        self.pad_value: int
+        self.resize_type: str
+        self.reverse_input_channels: bool
+        self.scale_values: list[int | float]
+
         if self.nscthw_layout:
             self.n, self.s, self.c, self.t, self.h, self.w = self.inputs[self.image_blob_name].shape
         else:
@@ -118,7 +126,7 @@ def parameters(cls) -> dict[str, Any]:
         )
         return parameters
 
-    def _get_inputs(self) -> tuple[list[str], list[str]]:
+    def _get_inputs(self) -> list[str]:
         """Defines the model inputs for images and additional info.
 
         Raises:

model_api/python/model_api/models/anomaly.py

@@ -14,6 +14,8 @@
 import cv2
 import numpy as np
 
+from model_api.adapters.inference_adapter import InferenceAdapter
+
 from .image_model import ImageModel
 from .types import ListValue, NumericalValue, StringValue
 from .utils import AnomalyResult
@@ -22,7 +24,9 @@
 class AnomalyDetection(ImageModel):
     __model__ = "AnomalyDetection"
 
-    def __init__(self, inference_adapter, configuration=dict(), preload=False):
+    def __init__(
+        self, inference_adapter: InferenceAdapter, configuration: dict = dict(), preload: bool = False
+    ) -> None:
         super().__init__(inference_adapter, configuration, preload)
         self._check_io_number(1, 1)
         self.normalization_scale: float
@@ -31,7 +35,7 @@ def __init__(self, inference_adapter, configuration=dict(), preload=False):
         self.task: str
         self.labels: list[str]
 
-    def postprocess(self, outputs: dict[str, np.ndarray], meta: dict[str, Any]):
+    def postprocess(self, outputs: dict[str, np.ndarray], meta: dict[str, Any]) -> AnomalyResult:
         """Post-processes the outputs and returns the results.
 
         Args:

model_api/python/model_api/models/classification.py

@@ -14,6 +14,8 @@
 from openvino.runtime import Model, Type
 from openvino.runtime import opset10 as opset
 
+from model_api.adapters.inference_adapter import InferenceAdapter
+
 from .image_model import ImageModel
 from .types import BooleanValue, ListValue, NumericalValue, StringValue
 from .utils import ClassificationResult
@@ -22,13 +24,24 @@
 class ClassificationModel(ImageModel):
     __model__ = "Classification"
 
-    def __init__(self, inference_adapter, configuration=dict(), preload=False):
+    def __init__(self, inference_adapter: InferenceAdapter, configuration: dict = dict(), preload: bool = False):
         super().__init__(inference_adapter, configuration, preload=False)
+        self.topk: int
+        self.labels: list[str]
+        self.path_to_labels: str
+        self.multilabel: bool
+        self.hierarchical: bool
+        self.hierarchical_config: str
+        self.confidence_threshold: float
+        self.output_raw_scores: bool
+        self.hierarchical_postproc: str
+        self.labels_resolver: GreedyLabelsResolver | ProbabilisticLabelsResolver
+
         self._check_io_number(1, (1, 2, 3, 4, 5))
         if self.path_to_labels:
             self.labels = self._load_labels(self.path_to_labels)
         if len(self.outputs) == 1:
-            self._verify_signle_output()
+            self._verify_single_output()
 
         self.raw_scores_name = _raw_scores_name
         if self.hierarchical:
@@ -99,7 +112,7 @@ def _load_labels(self, labels_file):
                 labels.append(s[(begin_idx + 1) : end_idx])
         return labels
 
-    def _verify_signle_output(self):
+    def _verify_single_output(self):
         layer_name = next(iter(self.outputs))
         layer_shape = self.outputs[layer_name].shape
 
@@ -197,7 +210,7 @@ def get_saliency_maps(self, outputs: dict) -> np.ndarray:
         if not self.hierarchical:
             return saliency_maps
 
-        reordered_saliency_maps = [[] for _ in range(len(saliency_maps))]
+        reordered_saliency_maps: list[list[np.ndarray]] = [[] for _ in range(len(saliency_maps))]
         model_classes = self.hierarchical_info["cls_heads_info"]["class_to_group_idx"]
         label_to_model_out_idx = {lbl: i for i, lbl in enumerate(model_classes.keys())}
         for batch in range(len(saliency_maps)):
@@ -279,7 +292,7 @@ def get_multiclass_predictions(self, outputs):
         return list(zip(indicesTensor, labels, scoresTensor))
 
 
-def addOrFindSoftmaxAndTopkOutputs(inference_adapter, topk, output_raw_scores):
+def addOrFindSoftmaxAndTopkOutputs(inference_adapter: InferenceAdapter, topk: int, output_raw_scores: bool):
     softmaxNode = None
     for i in range(len(inference_adapter.model.outputs)):
         output_node = inference_adapter.model.get_output_op(i).input(0).get_source_output().get_node()

model_api/python/model_api/models/detection_model.py

@@ -18,6 +18,8 @@ class DetectionModel(ImageModel):
     The `postprocess` method must be implemented in a specific inherited wrapper.
     """
 
+    __model__ = "DetectionModel"
+
     def __init__(self, inference_adapter, configuration=dict(), preload=False):
         """Detection Model constructor
 

model_api/python/model_api/models/image_model.py

@@ -30,6 +30,8 @@ class ImageModel(Model):
         input_transform (InputTransform): instance of the `InputTransform` for image normalization
     """
 
+    __model__ = "ImageModel"
+
     def __init__(self, inference_adapter, configuration=dict(), preload=False):
         """Image model constructor