Cover model

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

model_api/python/model_api/adapters/onnx_adapter.py

@@ -122,6 +122,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

@@ -4,6 +4,7 @@
 #
 
 import logging as log
+from os import PathLike
 from pathlib import Path
 from typing import Any
 
@@ -86,7 +87,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 = {}
 
@@ -117,14 +118,14 @@ def __init__(
         self,
         core: Core,
         model: str,
-        weights_path: 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: None = None,
-        cache_dir: None = None,
+        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

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,

model_api/python/model_api/models/model.py

@@ -6,6 +6,10 @@
 import logging as log
 import re
 from contextlib import contextmanager
+from os import PathLike
+from typing import Any, NoReturn, Type
+
+from numpy import ndarray
 
 from model_api.adapters.inference_adapter import InferenceAdapter
 from model_api.adapters.onnx_adapter import ONNXRuntimeAdapter
@@ -20,7 +24,7 @@
 class WrapperError(Exception):
     """The class for errors occurred in Model API wrappers"""
 
-    def __init__(self, wrapper_name, message):
+    def __init__(self, wrapper_name, message) -> None:
         super().__init__(f"{wrapper_name}: {message}")
 
 
@@ -52,7 +56,7 @@ class Model:
 
     __model__: str = "Model"
 
-    def __init__(self, inference_adapter, configuration: dict = {}, preload=False):
+    def __init__(self, inference_adapter: InferenceAdapter, configuration: dict = {}, preload: bool = False) -> None:
         """Model constructor
 
         Args:
@@ -98,7 +102,7 @@ def get_model(self):
         return model
 
     @classmethod
-    def get_model_class(cls, name):
+    def get_model_class(cls, name: str) -> Type:
         subclasses = [subclass for subclass in cls.get_subclasses() if subclass.__model__]
         if cls.__model__:
             subclasses.append(cls)
@@ -113,21 +117,21 @@ def get_model_class(cls, name):
     @classmethod
     def create_model(
         cls,
-        model,
-        model_type=None,
-        configuration={},
-        preload=True,
-        core=None,
-        weights_path="",
-        adaptor_parameters={},
-        device="AUTO",
-        nstreams="1",
-        nthreads=None,
-        max_num_requests=0,
-        precision="FP16",
-        download_dir=None,
-        cache_dir=None,
-    ):
+        model: str,
+        model_type: Any | None = None,
+        configuration: dict[str, Any] = {},
+        preload: bool = True,
+        core: Any | None = None,
+        weights_path: PathLike | None = None,
+        adaptor_parameters: dict[str, Any] = {},
+        device: str = "AUTO",
+        nstreams: str = "1",
+        nthreads: int | None = None,
+        max_num_requests: int = 0,
+        precision: str = "FP16",
+        download_dir: PathLike | None = None,
+        cache_dir: PathLike | None = None,
+    ) -> Any:
         """Create an instance of the Model API model
 
         Args:
@@ -152,9 +156,8 @@ def create_model(
         Returns:
             Model object
         """
-        if isinstance(model, InferenceAdapter):
-            inference_adapter = model
-        elif isinstance(model, str) and re.compile(
+        inference_adapter: InferenceAdapter
+        if isinstance(model, str) and re.compile(
             r"(\w+\.*\-*)*\w+:\d+\/models\/[a-zA-Z0-9._-]+(\:\d+)*",
         ).fullmatch(model):
             inference_adapter = OVMSAdapter(model)
@@ -182,7 +185,7 @@ def create_model(
         return Model(inference_adapter, configuration, preload)
 
     @classmethod
-    def get_subclasses(cls):
+    def get_subclasses(cls) -> list[Any]:
         all_subclasses = []
         for subclass in cls.__subclasses__():
             all_subclasses.append(subclass)
@@ -196,7 +199,7 @@ def available_wrappers(cls):
         return [subclass.__model__ for subclass in available_classes if subclass.__model__]
 
     @classmethod
-    def parameters(cls):
+    def parameters(cls) -> dict[str, Any]:
         """Defines the description and type of configurable data parameters for the wrapper.
 
         See `types.py` to find available types of the data parameter. For each parameter
@@ -214,7 +217,7 @@ def parameters(cls):
         """
         return {}
 
-    def _load_config(self, config):
+    def _load_config(self, config: dict[str, Any]) -> None:
         """Reads the configuration and creates data attributes
            by setting the wrapper parameters with values from configuration.
 
@@ -265,7 +268,7 @@ def _load_config(self, config):
                 )
 
     @classmethod
-    def raise_error(cls, message):
+    def raise_error(cls, message) -> NoReturn:
         """Raises the WrapperError.
 
         Args:
@@ -292,7 +295,7 @@ def preprocess(self, inputs):
         """
         raise NotImplementedError
 
-    def postprocess(self, outputs, meta):
+    def postprocess(self, outputs: dict[str, Any], meta: dict[str, Any]):
         """Interface for postprocess method.
 
         Args:
@@ -309,7 +312,11 @@ def postprocess(self, outputs, meta):
         """
         raise NotImplementedError
 
-    def _check_io_number(self, number_of_inputs, number_of_outputs):
+    def _check_io_number(
+        self,
+        number_of_inputs: int | tuple[int, ...],
+        number_of_outputs: int | tuple[int, ...],
+    ) -> None:
         """Checks whether the number of model inputs/outputs is supported.
 
         Args:
@@ -321,47 +328,32 @@ def _check_io_number(self, number_of_inputs, number_of_outputs):
         Raises:
             WrapperError: if the model has unsupported number of inputs/outputs
         """
-        if not isinstance(number_of_inputs, tuple):
+        if isinstance(number_of_inputs, int):
             if len(self.inputs) != number_of_inputs and number_of_inputs != -1:
                 self.raise_error(
-                    "Expected {} input blob{}, but {} found: {}".format(
-                        number_of_inputs,
-                        "s" if number_of_inputs != 1 else "",
-                        len(self.inputs),
-                        ", ".join(self.inputs),
-                    ),
+                    f"Expected {number_of_inputs} input blob {'s' if number_of_inputs != 1 else ''}, "
+                    f"but {len(self.inputs)} found: {', '.join(self.inputs)}",
                 )
         elif 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]),
-                    int(number_of_inputs[-1]),
-                    len(self.inputs),
-                    ", ".join(self.inputs),
-                ),
+                f"Expected {', '.join(str(n) for n in number_of_inputs[:-1])} or "
+                f"{int(number_of_inputs[-1])} input blobs, but {len(self.inputs)} found: {', '.join(self.inputs)}",
             )
 
-        if not isinstance(number_of_outputs, tuple):
+        if isinstance(number_of_outputs, int):
             if len(self.outputs) != number_of_outputs and number_of_outputs != -1:
                 self.raise_error(
-                    "Expected {} output blob{}, but {} found: {}".format(
-                        number_of_outputs,
-                        "s" if number_of_outputs != 1 else "",
-                        len(self.outputs),
-                        ", ".join(self.outputs),
-                    ),
+                    f"Expected {number_of_outputs} output blob {'s' if number_of_outputs != 1 else ''}, "
+                    f"but {len(self.outputs)} found: {', '.join(self.outputs)}",
                 )
         elif 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]),
-                    int(number_of_outputs[-1]),
-                    len(self.outputs),
-                    ", ".join(self.outputs),
-                ),
+                f"Expected {', '.join(str(n) for n in number_of_outputs[:-1])} or "
+                f"{int(number_of_outputs[-1])} output blobs, "
+                f"but {len(self.outputs)} found: {', '.join(self.outputs)}",
             )
 
-    def __call__(self, inputs):
+    def __call__(self, inputs: ndarray):
         """Applies preprocessing, synchronous inference, postprocessing routines while one call.
 
         Args:
@@ -407,7 +399,7 @@ def batch_infer_callback(result, id):
 
         return [completed_results[i] for i in range(len(inputs))]
 
-    def load(self, force=False):
+    def load(self, force: bool = False) -> None:
         if not self.model_loaded or force:
             self.model_loaded = True
             self.inference_adapter.load_model()
@@ -423,7 +415,7 @@ def reshape(self, new_shape):
         self.inputs = self.inference_adapter.get_input_layers()
         self.outputs = self.inference_adapter.get_output_layers()
 
-    def infer_sync(self, dict_data):
+    def infer_sync(self, dict_data: dict[str, ndarray]) -> dict[str, ndarray]:
         if not self.model_loaded:
             self.raise_error(
                 "The model is not loaded to the device. Please, create the wrapper "

model_api/python/model_api/models/sam_models.py

@@ -192,8 +192,8 @@ def _get_preprocess_shape(
 
     def _check_io_number(
         self,
-        number_of_inputs: int | tuple[int],
-        number_of_outputs: int | tuple[int],
+        number_of_inputs: int | tuple[int, ...],
+        number_of_outputs: int | tuple[int, ...],
     ) -> None:
         pass