add mypy

Signed-off-by: Ashwin Vaidya <[email protected]>

Author: ashwinvaidya17

.pre-commit-config.yaml

@@ -13,6 +13,13 @@ repos:
       - id: debug-statements
       - id: detect-private-key
 
+  # 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

@@ -14,9 +14,9 @@
  limitations under the License.
 """
 
-import abc
+from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Dict, List, Set, Tuple
+from typing import Any, Dict, List, Set, Tuple
 
 
 @dataclass
@@ -29,7 +29,7 @@ class Metadata:
     meta: Dict = field(default_factory=dict)
 
 
-class InferenceAdapter(metaclass=abc.ABCMeta):
+class InferenceAdapter(ABC):
     """
     An abstract Model Adapter with the following interface:
 
@@ -43,20 +43,25 @@ class InferenceAdapter(metaclass=abc.ABCMeta):
 
     precisions = ("FP32", "I32", "FP16", "I16", "I8", "U8")
 
-    @abc.abstractmethod
-    def __init__(self):
+    @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,
@@ -67,7 +72,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,
@@ -78,7 +83,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.
@@ -93,7 +98,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.
@@ -115,8 +120,8 @@ def infer_sync(self, dict_data):
                 }
         """
 
-    @abc.abstractmethod
-    def infer_async(self, dict_data, callback_fn, callback_data):
+    @abstractmethod
+    def infer_async(self, dict_data, callback_data):
         """
         Performs the asynchronous model inference and sets
         the callback for inference completion. Also, it should
@@ -130,11 +135,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
@@ -145,27 +149,27 @@ 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/utils.py

@@ -16,7 +16,7 @@
 
 import math
 from functools import partial
-from typing import Optional
+from typing import Callable, Optional
 
 import cv2
 import numpy as np
@@ -492,7 +492,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

@@ -19,6 +19,7 @@
 from typing import TYPE_CHECKING, Any
 
 import numpy as np
+
 from model_api.adapters.utils import RESIZE_TYPES, InputTransform
 
 from .model import Model
@@ -74,6 +75,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
@@ -129,7 +138,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

@@ -24,6 +24,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
@@ -32,7 +34,7 @@
 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
@@ -41,7 +43,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

@@ -25,6 +25,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
@@ -33,13 +35,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 1 == len(self.outputs):
-            self._verify_signle_output()
+            self._verify_single_output()
 
         self.raw_scores_name = _raw_scores_name
         if self.hierarchical:
@@ -108,7 +121,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
 
@@ -207,7 +220,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)):
@@ -296,7 +309,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 = (

model_api/python/model_api/models/model.py

@@ -16,6 +16,7 @@
 
 import logging as log
 import re
+from abc import ABC
 from contextlib import contextmanager
 
 from model_api.adapters.inference_adapter import InferenceAdapter
@@ -35,7 +36,7 @@ def __init__(self, wrapper_name, message):
         super().__init__(f"{wrapper_name}: {message}")
 
 
-class Model:
+class Model(ABC):
     """An abstract model wrapper
 
     The abstract model wrapper is free from any executor dependencies.
@@ -61,7 +62,7 @@ class Model:
         model_loaded (bool): a flag whether the model is loaded to device
     """
 
-    __model__ = None  # Abstract wrapper has no name
+    __model__ :str
 
     def __init__(self, inference_adapter, configuration=dict(), preload=False):
         """Model constructor
@@ -101,19 +102,11 @@ def __init__(self, inference_adapter, configuration=dict(), preload=False):
         self.callback_fn = lambda _: None
 
     def get_model(self):
-        """Returns the ov.Model object stored in the InferenceAdapter.
-
-        Note: valid only for local inference
-
-        Returns:
-            ov.Model object
-        Raises:
-            RuntimeError: in case of remote inference (serving)
-        """
-        if isinstance(self.inference_adapter, OpenvinoAdapter):
-            return self.inference_adapter.get_model()
-
-        raise RuntimeError("get_model() is not supported for remote inference")
+        model = self.inference_adapter.get_model()
+        model.set_rt_info(self.__model__, ["model_info", "model_type"])
+        for name in self.parameters():
+            model.set_rt_info(getattr(self, name), ["model_info", name])
+        return model
 
     @classmethod
     def get_model_class(cls, name):
@@ -281,8 +274,8 @@ def _load_config(self, config):
                 errors = parameters[name].validate(value)
                 if errors:
                     self.logger.error(f'Error with "{name}" parameter:')
-                    for error in errors:
-                        self.logger.error(f"\t{error}")
+                    for _error in errors:
+                        self.logger.error(f"\t{_error}")
                     self.raise_error("Incorrect user configuration")
                 value = parameters[name].get_value(value)
                 self.__setattr__(name, value)
@@ -359,7 +352,7 @@ def _check_io_number(self, number_of_inputs, number_of_outputs):
                     )
                 )
         else:
-            if not len(self.inputs) in number_of_inputs:
+            if len(self.inputs) not in number_of_inputs:
                 self.raise_error(
                     "Expected {} or {} input blobs, but {} found: {}".format(
                         ", ".join(str(n) for n in number_of_inputs[:-1]),
@@ -380,7 +373,7 @@ def _check_io_number(self, number_of_inputs, number_of_outputs):
                     )
                 )
         else:
-            if not len(self.outputs) in number_of_outputs:
+            if len(self.outputs) not in number_of_outputs:
                 self.raise_error(
                     "Expected {} or {} output blobs, but {} found: {}".format(
                         ", ".join(str(n) for n in number_of_outputs[:-1]),
@@ -523,12 +516,6 @@ def log_layers_info(self):
                 )
             )
 
-    def get_model(self):
-        model = self.inference_adapter.get_model()
-        model.set_rt_info(self.__model__, ["model_info", "model_type"])
-        for name in self.parameters():
-            model.set_rt_info(getattr(self, name), ["model_info", name])
-        return model
 
     def save(self, xml_path, bin_path="", version="UNSPECIFIED"):
         import openvino