Type annotation (#229)

* Partial annotation (#226)

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

* Typing: core modules (#227)

* Typing for OV adapter

* Enable type checking option

* Fix typing issue in adapter

* Cover model

* Cover ImageModel

* Cover types

* Cover utils

* Add missing future imports

* Update type checking imports

---------

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

Author: sovrasov

model_api/python/model_api/adapters/inference_adapter.py

@@ -83,7 +83,7 @@ def reshape_model(self, new_shape):
         """
 
     @abstractmethod
-    def infer_sync(self, dict_data):
+    def infer_sync(self, dict_data) -> dict:
         """Performs the synchronous model inference. The infer is a blocking method.
 
         Args:
@@ -121,6 +121,22 @@ def infer_async(self, dict_data, callback_data):
             - callback_data: the data for callback, that will be taken after the model inference is ended
         """
 
+    @abstractmethod
+    def get_raw_result(self, infer_result) -> dict:
+        """Gets raw results from the internal inference framework representation as a dict.
+
+        Args:
+            - infer_result: framework-specific result of inference from the model
+
+        Returns:
+            - raw result (dict) - model raw output in the following format:
+                {
+                    'output_layer_name_1': raw_result_1,
+                    'output_layer_name_2': raw_result_2,
+                    ...
+                }
+        """
+
     @abstractmethod
     def is_ready(self):
         """In case of asynchronous execution checks if one can submit input data
@@ -153,7 +169,7 @@ def embed_preprocessing(
         layout,
         resize_mode: str,
         interpolation_mode,
-        target_shape: tuple[int],
+        target_shape: tuple[int, ...],
         pad_value,
         dtype: type = int,
         brg2rgb=False,

model_api/python/model_api/adapters/onnx_adapter.py

@@ -3,6 +3,8 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
+from __future__ import annotations  # TODO: remove when Python3.9 support is dropped
+
 import sys
 from functools import partial, reduce
 
@@ -122,6 +124,9 @@ def await_all(self):
     def await_any(self):
         pass
 
+    def get_raw_result(self, infer_result):
+        pass
+
     def embed_preprocessing(
         self,
         layout,

model_api/python/model_api/adapters/openvino_adapter.py

@@ -3,15 +3,24 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
+from __future__ import annotations  # TODO: remove when Python3.9 support is dropped
+
 import logging as log
 from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from os import PathLike
+
+    from numpy import ndarray
 
 try:
     import openvino.runtime as ov
     from openvino import (
         AsyncInferQueue,
         Core,
         Dimension,
+        OVAny,
         PartialShape,
         Type,
         get_version,
@@ -35,7 +44,7 @@
 )
 
 
-def create_core():
+def create_core() -> Core:
     if openvino_absent:
         msg = "The OpenVINO package is not installed"
         raise ImportError(msg)
@@ -45,7 +54,7 @@ def create_core():
     return Core()
 
 
-def parse_devices(device_string):
+def parse_devices(device_string: str) -> list[str]:
     colon_position = device_string.find(":")
     if colon_position != -1:
         device_type = device_string[:colon_position]
@@ -57,7 +66,7 @@ def parse_devices(device_string):
                 if parenthesis_position != -1:
                     device = device[:parenthesis_position]
             return devices
-    return (device_string,)
+    return [device_string]
 
 
 def parse_value_per_device(devices: set[str], values_string: str) -> dict[str, int]:
@@ -82,7 +91,7 @@ def parse_value_per_device(devices: set[str], values_string: str) -> dict[str, i
 def get_user_config(
     flags_d: str,
     flags_nstreams: str,
-    flags_nthreads: int,
+    flags_nthreads: int | None = None,
 ) -> dict[str, str]:
     config = {}
 
@@ -111,17 +120,17 @@ class OpenvinoAdapter(InferenceAdapter):
 
     def __init__(
         self,
-        core,
-        model,
-        weights_path="",
-        model_parameters={},
-        device="CPU",
-        plugin_config=None,
-        max_num_requests=0,
-        precision="FP16",
-        download_dir=None,
-        cache_dir=None,
-    ):
+        core: Core,
+        model: str,
+        weights_path: PathLike | None = None,
+        model_parameters: dict[str, Any] = {},
+        device: str = "CPU",
+        plugin_config: dict[str, Any] | None = None,
+        max_num_requests: int = 0,
+        precision: str = "FP16",
+        download_dir: PathLike | None = None,
+        cache_dir: PathLike | None = None,
+    ) -> None:
         """precision, download_dir and cache_dir are ignored if model is a path to a file"""
         self.core = core
         self.model_path = model
@@ -179,7 +188,7 @@ def __init__(
         msg = "Model must be bytes, a file or existing OMZ model name"
         raise RuntimeError(msg)
 
-    def load_model(self):
+    def load_model(self) -> None:
         self.compiled_model = self.core.compile_model(
             self.model,
             self.device,
@@ -201,7 +210,7 @@ def load_model(self):
         )
         self.log_runtime_settings()
 
-    def log_runtime_settings(self):
+    def log_runtime_settings(self) -> None:
         devices = set(parse_devices(self.device))
         if "AUTO" not in devices:
             for device in devices:
@@ -222,7 +231,7 @@ def log_runtime_settings(self):
                     pass
         log.info(f"\tNumber of model infer requests: {len(self.async_queue)}")
 
-    def get_input_layers(self):
+    def get_input_layers(self) -> dict[str, Metadata]:
         inputs = {}
         for input in self.model.inputs:
             input_shape = get_input_shape(input)
@@ -235,7 +244,11 @@ def get_input_layers(self):
             )
         return self._get_meta_from_ngraph(inputs)
 
-    def get_layout_for_input(self, input, shape=None) -> str:
+    def get_layout_for_input(
+        self,
+        input: ov.Output,
+        shape: list[int] | tuple[int, int, int, int] | None = None,
+    ) -> str:
         input_layout = ""
         if self.model_parameters["input_layouts"]:
             input_layout = Layout.from_user_layouts(
@@ -251,7 +264,7 @@ def get_layout_for_input(self, input, shape=None) -> str:
                 )
         return input_layout
 
-    def get_output_layers(self):
+    def get_output_layers(self) -> dict[str, Metadata]:
         outputs = {}
         for i, output in enumerate(self.model.outputs):
             output_shape = output.partial_shape.get_min_shape() if self.model.is_dynamic() else output.shape
@@ -273,13 +286,13 @@ def reshape_model(self, new_shape):
         }
         self.model.reshape(new_shape)
 
-    def get_raw_result(self, request):
+    def get_raw_result(self, request: ov.InferRequest) -> dict[str, ndarray]:
         return {key: request.get_tensor(key).data for key in self.get_output_layers()}
 
     def copy_raw_result(self, request):
         return {key: request.get_tensor(key).data.copy() for key in self.get_output_layers()}
 
-    def infer_sync(self, dict_data):
+    def infer_sync(self, dict_data: dict[str, ndarray]) -> dict[str, ndarray]:
         self.infer_request = self.async_queue[self.async_queue.get_idle_request_id()]
         self.infer_request.infer(dict_data)
         return self.get_raw_result(self.infer_request)
@@ -299,7 +312,7 @@ def await_all(self) -> None:
     def await_any(self) -> None:
         self.async_queue.get_idle_request_id()
 
-    def _get_meta_from_ngraph(self, layers_info):
+    def _get_meta_from_ngraph(self, layers_info: dict[str, Metadata]) -> dict[str, Metadata]:
         for node in self.model.get_ordered_ops():
             layer_name = node.get_friendly_name()
             if layer_name not in layers_info:
@@ -319,24 +332,24 @@ def operations_by_type(self, operation_type):
                 )
         return layers_info
 
-    def get_rt_info(self, path):
+    def get_rt_info(self, path: list[str]) -> OVAny:
         if self.is_onnx_file:
             return get_rt_info_from_dict(self.onnx_metadata, path)
         return self.model.get_rt_info(path)
 
     def embed_preprocessing(
         self,
-        layout,
+        layout: str,
         resize_mode: str,
-        interpolation_mode,
-        target_shape: tuple[int],
-        pad_value,
+        interpolation_mode: str,
+        target_shape: tuple[int, ...],
+        pad_value: int,
         dtype: type = int,
-        brg2rgb=False,
-        mean=None,
-        scale=None,
-        input_idx=0,
-    ):
+        brg2rgb: bool = False,
+        mean: list[Any] | None = None,
+        scale: list[Any] | None = None,
+        input_idx: int = 0,
+    ) -> None:
         ppp = PrePostProcessor(self.model)
 
         # Change the input type to the 8-bit image
@@ -371,7 +384,7 @@ def embed_preprocessing(
                 ppp.input(input_idx).tensor().set_shape(input_shape)
                 ppp.input(input_idx).preprocess().custom(
                     RESIZE_MODE_MAP[resize_mode](
-                        target_shape,
+                        (target_shape[0], target_shape[1]),
                         INTERPOLATION_MODE_MAP[interpolation_mode],
                         pad_value,
                     ),
@@ -407,7 +420,7 @@ def get_model(self):
         return self.model
 
 
-def get_input_shape(input_tensor):
+def get_input_shape(input_tensor: ov.Output) -> list[int]:
     def string_to_tuple(string, casting_type=int):
         processed = string.replace(" ", "").replace("(", "").replace(")", "").split(",")
         processed = filter(lambda x: x, processed)
@@ -428,4 +441,4 @@ def string_to_tuple(string, casting_type=int):
             else:
                 shape_list.append(int(dim))
         return shape_list
-    return string_to_tuple(preprocessed)
+    return list(string_to_tuple(preprocessed))

model_api/python/model_api/adapters/ovms_adapter.py

@@ -97,6 +97,9 @@ def await_all(self):
     def await_any(self):
         pass
 
+    def get_raw_result(self, infer_result):
+        pass
+
     def embed_preprocessing(
         self,
         layout,