Partial annotation

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

Author: ashwinvaidya17

model_api/python/model_api/models/anomaly.py

@@ -15,10 +15,9 @@
 import numpy as np
 
 from model_api.adapters.inference_adapter import InferenceAdapter
-
-from .image_model import ImageModel
-from .result_types import AnomalyResult
-from .types import ListValue, NumericalValue, StringValue
+from model_api.models.image_model import ImageModel
+from model_api.models.result_types import AnomalyResult
+from model_api.models.types import ListValue, NumericalValue, StringValue
 
 
 class AnomalyDetection(ImageModel):

model_api/python/model_api/models/classification.py

@@ -9,23 +9,24 @@
 import json
 from collections import defaultdict
 from pathlib import Path
+from typing import List, Tuple
 
 import numpy as np
+from numpy import float32
 from openvino.preprocess import PrePostProcessor
 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 .result_types import ClassificationResult
-from .types import BooleanValue, ListValue, NumericalValue, StringValue
+from model_api.models.image_model import ImageModel
+from model_api.models.result_types import ClassificationResult
+from model_api.models.types import BooleanValue, ListValue, NumericalValue, StringValue
 
 
 class ClassificationModel(ImageModel):
     __model__ = "Classification"
 
-    def __init__(self, inference_adapter: InferenceAdapter, configuration: dict = {}, preload: bool = False):
+    def __init__(self, inference_adapter: InferenceAdapter, configuration: dict = {}, preload: bool = False) -> None:
         super().__init__(inference_adapter, configuration, preload=False)
         self.topk: int
         self.labels: list[str]
@@ -102,7 +103,7 @@ def __init__(self, inference_adapter: InferenceAdapter, configuration: dict = {}
         if preload:
             self.load()
 
-    def _load_labels(self, labels_file):
+    def _load_labels(self, labels_file: str) -> list:
         with Path(labels_file).open() as f:
             labels = []
             for s in f:
@@ -113,7 +114,7 @@ def _load_labels(self, labels_file):
                 labels.append(s[(begin_idx + 1) : end_idx])
         return labels
 
-    def _verify_single_output(self):
+    def _verify_single_output(self) -> None:
         layer_name = next(iter(self.outputs))
         layer_shape = self.outputs[layer_name].shape
 
@@ -137,7 +138,7 @@ def _verify_single_output(self):
                 )
 
     @classmethod
-    def parameters(cls):
+    def parameters(cls) -> dict:
         parameters = super().parameters()
         parameters.update(
             {
@@ -180,7 +181,8 @@ def parameters(cls):
         )
         return parameters
 
-    def postprocess(self, outputs, meta):
+    def postprocess(self, outputs: dict, meta: dict) -> ClassificationResult:
+        del meta  # unused
         if self.multilabel:
             result = self.get_multilabel_predictions(
                 outputs[self.out_layer_names[0]].squeeze(),
@@ -220,7 +222,7 @@ def get_saliency_maps(self, outputs: dict) -> np.ndarray:
                 reordered_saliency_maps[batch].append(saliency_maps[batch][idx])
         return np.array(reordered_saliency_maps)
 
-    def get_all_probs(self, logits: np.ndarray):
+    def get_all_probs(self, logits: np.ndarray) -> np.ndarray:
         if self.multilabel:
             probs = sigmoid_numpy(logits.reshape(-1))
         elif self.hierarchical:
@@ -266,10 +268,10 @@ def get_hierarchical_predictions(self, logits: np.ndarray):
                     predicted_labels.append(label_str)
                     predicted_scores.append(head_logits[i])
 
-        predictions = zip(predicted_labels, predicted_scores)
+        predictions = list(zip(predicted_labels, predicted_scores))
         return self.labels_resolver.resolve_labels(predictions)
 
-    def get_multilabel_predictions(self, logits: np.ndarray):
+    def get_multilabel_predictions(self, logits: np.ndarray) -> List[Tuple[int, str, float32]]:
         logits = sigmoid_numpy(logits)
         scores = []
         indices = []
@@ -281,7 +283,7 @@ def get_multilabel_predictions(self, logits: np.ndarray):
 
         return list(zip(indices, labels, scores))
 
-    def get_multiclass_predictions(self, outputs):
+    def get_multiclass_predictions(self, outputs: dict) -> list[tuple[int, str, float]]:
         if self.embedded_topk:
             indicesTensor = outputs[self.out_layer_names[0]][0]
             scoresTensor = outputs[self.out_layer_names[1]][0]
@@ -293,7 +295,7 @@ def get_multiclass_predictions(self, outputs):
         return list(zip(indicesTensor, labels, scoresTensor))
 
 
-def addOrFindSoftmaxAndTopkOutputs(inference_adapter: InferenceAdapter, topk: int, output_raw_scores: bool):
+def addOrFindSoftmaxAndTopkOutputs(inference_adapter: InferenceAdapter, topk: int, output_raw_scores: bool) -> None:
     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()
@@ -345,17 +347,17 @@ def addOrFindSoftmaxAndTopkOutputs(inference_adapter: InferenceAdapter, topk: in
     inference_adapter.model = ppp.build()
 
 
-def sigmoid_numpy(x: np.ndarray):
+def sigmoid_numpy(x: np.ndarray) -> np.ndarray:
     return 1.0 / (1.0 + np.exp(-x))
 
 
-def softmax_numpy(x: np.ndarray, eps: float = 1e-9):
+def softmax_numpy(x: np.ndarray, eps: float = 1e-9) -> np.ndarray:
     x = np.exp(x - np.max(x))
     return x / (np.sum(x) + eps)
 
 
 class GreedyLabelsResolver:
-    def __init__(self, hierarchical_config) -> None:
+    def __init__(self, hierarchical_config: dict) -> None:
         self.label_to_idx = hierarchical_config["cls_heads_info"]["label_to_idx"]
         self.label_relations = hierarchical_config["label_tree_edges"]
         self.label_groups = hierarchical_config["cls_heads_info"]["all_groups"]
@@ -364,7 +366,7 @@ def __init__(self, hierarchical_config) -> None:
         for child, parent in self.label_relations:
             self.label_tree.add_edge(parent, child)
 
-    def resolve_labels(self, predictions):
+    def resolve_labels(self, predictions: list[tuple]) -> list:
         """Resolves hierarchical labels and exclusivity based on a list of ScoredLabels (labels with probability).
         The following two steps are taken:
         - select the most likely label from each label group
@@ -374,7 +376,7 @@ def resolve_labels(self, predictions):
             predictions: a list of tuples (label name, score)
         """
 
-        def get_predecessors(lbl, candidates):
+        def get_predecessors(lbl: str, candidates: list[str]) -> list:
             """Return all predecessors.
 
             Returns all the predecessors of the input label or an empty list if one of the predecessors is not a
@@ -422,12 +424,12 @@ def get_predecessors(lbl, candidates):
 
 
 class ProbabilisticLabelsResolver(GreedyLabelsResolver):
-    def __init__(self, hierarchical_config, warmup_cache=True) -> None:
+    def __init__(self, hierarchical_config: dict, warmup_cache: bool = True) -> None:
         super().__init__(hierarchical_config)
         if warmup_cache:
             self.label_tree.get_labels_in_topological_order()
 
-    def resolve_labels(self, predictions):
+    def resolve_labels(self, predictions: list[tuple[str, float]]) -> list[tuple[int, str, float]]:
         """Resolves hierarchical labels and exclusivity based on a list of ScoredLabels (labels with probability).
 
         The following two steps are taken:
@@ -446,8 +448,8 @@ def resolve_labels(self, predictions):
 
     def __resolve_labels_probabilistic(
         self,
-        label_to_probability,
-    ):
+        label_to_probability: dict[str, float],
+    ) -> list[tuple[int, str, float]]:
         """Resolves hierarchical labels and exclusivity based on a probabilistic label output.
 
         - selects the most likely (max) label from an exclusive group
@@ -546,24 +548,24 @@ class SimpleLabelsGraph:
     like adding edges, getting children and parents.
     """
 
-    def __init__(self, vertices):
+    def __init__(self, vertices: list[str]) -> None:
         self._v = vertices
-        self._adj = defaultdict(list)
-        self._topological_order_cache = None
-        self._parents_map = {}
+        self._adj: dict[str, list] = {v: [] for v in vertices}
+        self._topological_order_cache: list | None = None
+        self._parents_map: dict[str, str] = {}
 
-    def add_edge(self, parent, child):
+    def add_edge(self, parent: str, child: str) -> None:
         self._adj[parent].append(child)
         self._parents_map[child] = parent
         self.clear_topological_cache()
 
-    def get_children(self, label):
+    def get_children(self, label: str) -> list:
         return self._adj[label]
 
-    def get_parent(self, label):
+    def get_parent(self, label: str) -> str | None:
         return self._parents_map.get(label, None)
 
-    def get_ancestors(self, label):
+    def get_ancestors(self, label: str) -> list[str]:
         """Returns all the ancestors of the input label, including self."""
         predecessors = [label]
         last_parent = self.get_parent(label)
@@ -576,14 +578,14 @@ def get_ancestors(self, label):
 
         return predecessors
 
-    def get_labels_in_topological_order(self):
+    def get_labels_in_topological_order(self) -> list:
         if self._topological_order_cache is None:
             self._topological_order_cache = self.topological_sort()
 
         return self._topological_order_cache
 
-    def topological_sort(self):
-        in_degree = defaultdict(int)
+    def topological_sort(self) -> list:
+        in_degree: dict[str, int] = dict.fromkeys(self._v, 0)
 
         for node_adj in self._adj.values():
             for j in node_adj:
@@ -610,7 +612,7 @@ def topological_sort(self):
 
         return ordered
 
-    def clear_topological_cache(self):
+    def clear_topological_cache(self) -> None:
         self._topological_order_cache = None
 
 
@@ -619,15 +621,15 @@ def clear_topological_cache(self):
 _raw_scores_name = "raw_scores"
 
 
-def _get_non_xai_names(output_names):
+def _get_non_xai_names(output_names: list[str]) -> list[str]:
     return [
         output_name
         for output_name in output_names
         if _saliency_map_name != output_name and _feature_vector_name != output_name
     ]
 
 
-def _append_xai_names(outputs, output_names):
+def _append_xai_names(outputs: dict, output_names: list[str]) -> None:
     if _saliency_map_name in outputs:
         output_names.append(_saliency_map_name)
     if _feature_vector_name in outputs:

model_api/python/model_api/models/image_model.py

@@ -3,10 +3,11 @@
 # SPDX-License-Identifier: Apache-2.0
 #
 
-from model_api.adapters.utils import RESIZE_TYPES, InputTransform
+import numpy as np
 
-from .model import Model
-from .types import BooleanValue, ListValue, NumericalValue, StringValue
+from model_api.adapters.utils import RESIZE_TYPES, InputTransform
+from model_api.models.model import Model
+from model_api.models.types import BooleanValue, ListValue, NumericalValue, StringValue
 
 
 class ImageModel(Model):
@@ -89,7 +90,7 @@ def __init__(self, inference_adapter, configuration: dict = {}, preload=False):
             self.orig_height, self.orig_width = self.h, self.w
 
     @classmethod
-    def parameters(cls):
+    def parameters(cls) -> dict:
         parameters = super().parameters()
         parameters.update(
             {
@@ -169,7 +170,7 @@ def _get_inputs(self):
             )
         return image_blob_names, image_info_blob_names
 
-    def preprocess(self, inputs):
+    def preprocess(self, inputs) -> list[dict]:
         """Data preprocess method
 
         It performs basic preprocessing of a single image:
@@ -194,10 +195,13 @@ def preprocess(self, inputs):
                 }
             - the input metadata, which might be used in `postprocess` method
         """
-        return {self.image_blob_name: inputs[None]}, {
-            "original_shape": inputs.shape,
-            "resized_shape": (self.w, self.h, self.c),
-        }
+        return [
+            {self.image_blob_name: inputs[None]},
+            {
+                "original_shape": inputs.shape,
+                "resized_shape": (self.w, self.h, self.c),
+            },
+        ]
 
     def _change_layout(self, image):
         """Changes the input image layout to fit the layout of the model input layer.