move parsers back to ssd

Author: eugene123tw

model_api/python/model_api/models/result_types/__init__.py

@@ -5,12 +5,7 @@
 
 from .anomaly import AnomalyResult
 from .classification import ClassificationResult, Label
-from .detection import (
-    BoxesLabelsParser,
-    DetectionResult,
-    MultipleOutputParser,
-    SingleOutputParser,
-)
+from .detection import DetectionResult
 from .keypoint import DetectedKeypoints
 from .segmentation import (
     Contour,
@@ -22,13 +17,10 @@
 
 __all__ = [
     "AnomalyResult",
-    "BoxesLabelsParser",
     "ClassificationResult",
     "Contour",
     "DetectionResult",
     "DetectedKeypoints",
-    "MultipleOutputParser",
-    "SingleOutputParser",
     "Label",
     "ImageResultWithSoftPrediction",
     "InstanceSegmentationResult",

model_api/python/model_api/models/result_types/detection.py

@@ -10,19 +10,6 @@
 from .utils import array_shape_to_str
 
 
-def find_layer_by_name(name, layers):
-    suitable_layers = [layer_name for layer_name in layers if name in layer_name]
-    if not suitable_layers:
-        msg = f'Suitable layer for "{name}" output is not found'
-        raise ValueError(msg)
-
-    if len(suitable_layers) > 1:
-        msg = f'More than 1 layer matched to "{name}" output'
-        raise ValueError(msg)
-
-    return suitable_layers[0]
-
-
 class DetectionResult:
     """Result for detection model.
 
@@ -142,123 +129,3 @@ def feature_vector(self, value):
             msg = "Feature vector must be numpy array."
             raise ValueError(msg)
         self._feature_vector = value
-
-
-class SingleOutputParser:
-    def __init__(self, all_outputs):
-        if len(all_outputs) != 1:
-            msg = "Network must have only one output."
-            raise ValueError(msg)
-        self.output_name, output_data = next(iter(all_outputs.items()))
-        last_dim = output_data.shape[-1]
-        if last_dim != 7:
-            msg = f"The last dimension of the output blob must be equal to 7, got {last_dim} instead."
-            raise ValueError(msg)
-
-    def __call__(self, outputs) -> DetectionResult:
-        """Parse model outputs.
-
-        Args:
-            outputs (dict): Model outputs wrapped in dict.
-
-        Returns:
-            DetectionResult: Parsed model outputs.
-        """
-        bboxes = []
-        scores = []
-        labels = []
-        for _, label, score, xmin, ymin, xmax, ymax in outputs[self.output_name][0][0]:
-            bboxes.append((xmin, ymin, xmax, ymax))
-            scores.append(score)
-            labels.append(label)
-        bboxes = np.array(bboxes)
-        scores = np.array(scores)
-        labels = np.array(labels).astype(np.int32)
-        return DetectionResult(
-            bboxes=bboxes,
-            labels=labels,
-            scores=scores,
-        )
-
-
-class MultipleOutputParser:
-    def __init__(
-        self,
-        layers,
-        bboxes_layer="bboxes",
-        scores_layer="scores",
-        labels_layer="labels",
-    ):
-        self.labels_layer = find_layer_by_name(labels_layer, layers)
-        self.scores_layer = find_layer_by_name(scores_layer, layers)
-        self.bboxes_layer = find_layer_by_name(bboxes_layer, layers)
-
-    def __call__(self, outputs) -> DetectionResult:
-        """Parse model outputs.
-
-        Args:
-            outputs (dict): Model outputs wrapped in dict.
-
-        Returns:
-            DetectionResult: Parsed model outputs.
-        """
-        bboxes = np.array(outputs[self.bboxes_layer][0])
-        scores = np.array(outputs[self.scores_layer][0])
-        labels = np.array(outputs[self.labels_layer][0])
-        return DetectionResult(bboxes, scores, labels)
-
-
-class BoxesLabelsParser:
-    def __init__(self, layers, input_size, labels_layer="labels", default_label=0):
-        try:
-            self.labels_layer = find_layer_by_name(labels_layer, layers)
-        except ValueError:
-            self.labels_layer = None
-            self.default_label = default_label
-
-        self.bboxes_layer = self.find_layer_bboxes_output(layers)
-        self.input_size = input_size
-
-    @staticmethod
-    def find_layer_bboxes_output(layers):
-        filter_outputs = [
-            name
-            for name, data in layers.items()
-            if (len(data.shape) == 2 or len(data.shape) == 3) and data.shape[-1] == 5
-        ]
-        if not filter_outputs:
-            msg = "Suitable output with bounding boxes is not found"
-            raise ValueError(msg)
-        if len(filter_outputs) > 1:
-            msg = "More than 1 candidate for output with bounding boxes."
-            raise ValueError(msg)
-        return filter_outputs[0]
-
-    def __call__(self, outputs) -> DetectionResult:
-        """Parse model outputs.
-
-        Note: Bounding boxes layer from outputs are expected to be in format [xmin, ymin, xmax, ymax, score].
-
-        Args:
-            outputs (dict): Model outputs wrapped in dict.
-
-        Returns:
-            DetectionResult: Parsed model outputs.
-        """
-        bboxes = outputs[self.bboxes_layer]
-        bboxes = bboxes.squeeze(0)
-        scores = bboxes[:, 4]
-        bboxes = bboxes[:, :4]
-        bboxes[:, 0::2] /= self.input_size[0]
-        bboxes[:, 1::2] /= self.input_size[1]
-        if self.labels_layer:
-            labels = outputs[self.labels_layer]
-        else:
-            labels = np.full(len(bboxes), self.default_label, dtype=bboxes.dtype)
-        labels = labels.squeeze(0)
-
-        return DetectionResult(
-            bboxes=bboxes,
-            labels=labels,
-            scores=scores,
-        )

model_api/python/model_api/models/ssd.py

@@ -6,13 +6,146 @@
 import numpy as np
 
 from .detection_model import DetectionModel
-from .result_types import BoxesLabelsParser, DetectionResult, MultipleOutputParser, SingleOutputParser
+from .result_types import DetectionResult
 
 BBOX_AREA_THRESHOLD = 1.0
 SALIENCY_MAP_NAME = "saliency_map"
 FEATURE_VECTOR_NAME = "feature_vector"
 
 
+def find_layer_by_name(name, layers):
+    suitable_layers = [layer_name for layer_name in layers if name in layer_name]
+    if not suitable_layers:
+        msg = f'Suitable layer for "{name}" output is not found'
+        raise ValueError(msg)
+
+    if len(suitable_layers) > 1:
+        msg = f'More than 1 layer matched to "{name}" output'
+        raise ValueError(msg)
+
+    return suitable_layers[0]
+
+
+class SingleOutputParser:
+    def __init__(self, all_outputs):
+        if len(all_outputs) != 1:
+            msg = "Network must have only one output."
+            raise ValueError(msg)
+        self.output_name, output_data = next(iter(all_outputs.items()))
+        last_dim = output_data.shape[-1]
+        if last_dim != 7:
+            msg = f"The last dimension of the output blob must be equal to 7, got {last_dim} instead."
+            raise ValueError(msg)
+
+    def __call__(self, outputs) -> DetectionResult:
+        """Parse model outputs.
+
+        Args:
+            outputs (dict): Model outputs wrapped in dict.
+
+        Returns:
+            DetectionResult: Parsed model outputs.
+        """
+        bboxes = []
+        scores = []
+        labels = []
+        for _, label, score, xmin, ymin, xmax, ymax in outputs[self.output_name][0][0]:
+            bboxes.append((xmin, ymin, xmax, ymax))
+            scores.append(score)
+            labels.append(label)
+        bboxes = np.array(bboxes)
+        scores = np.array(scores)
+        labels = np.array(labels).astype(np.int32)
+        return DetectionResult(
+            bboxes=bboxes,
+            labels=labels,
+            scores=scores,
+        )
+
+
+class MultipleOutputParser:
+    def __init__(
+        self,
+        layers,
+        bboxes_layer="bboxes",
+        scores_layer="scores",
+        labels_layer="labels",
+    ):
+        self.labels_layer = find_layer_by_name(labels_layer, layers)
+        self.scores_layer = find_layer_by_name(scores_layer, layers)
+        self.bboxes_layer = find_layer_by_name(bboxes_layer, layers)
+
+    def __call__(self, outputs) -> DetectionResult:
+        """Parse model outputs.
+
+        Args:
+            outputs (dict): Model outputs wrapped in dict.
+
+        Returns:
+            DetectionResult: Parsed model outputs.
+        """
+        bboxes = np.array(outputs[self.bboxes_layer][0])
+        scores = np.array(outputs[self.scores_layer][0])
+        labels = np.array(outputs[self.labels_layer][0])
+        return DetectionResult(bboxes, scores, labels)
+
+
+class BoxesLabelsParser:
+    def __init__(self, layers, input_size, labels_layer="labels", default_label=0):
+        try:
+            self.labels_layer = find_layer_by_name(labels_layer, layers)
+        except ValueError:
+            self.labels_layer = None
+            self.default_label = default_label
+
+        self.bboxes_layer = self.find_layer_bboxes_output(layers)
+        self.input_size = input_size
+
+    @staticmethod
+    def find_layer_bboxes_output(layers):
+        filter_outputs = [
+            name
+            for name, data in layers.items()
+            if (len(data.shape) == 2 or len(data.shape) == 3) and data.shape[-1] == 5
+        ]
+        if not filter_outputs:
+            msg = "Suitable output with bounding boxes is not found"
+            raise ValueError(msg)
+        if len(filter_outputs) > 1:
+            msg = "More than 1 candidate for output with bounding boxes."
+            raise ValueError(msg)
+        return filter_outputs[0]
+
+    def __call__(self, outputs) -> DetectionResult:
+        """Parse model outputs.
+
+        Note: Bounding boxes layer from outputs are expected to be in format [xmin, ymin, xmax, ymax, score].
+
+        Args:
+            outputs (dict): Model outputs wrapped in dict.
+
+        Returns:
+            DetectionResult: Parsed model outputs.
+        """
+        bboxes = outputs[self.bboxes_layer]
+        bboxes = bboxes.squeeze(0)
+        scores = bboxes[:, 4]
+        bboxes = bboxes[:, :4]
+        bboxes[:, 0::2] /= self.input_size[0]
+        bboxes[:, 1::2] /= self.input_size[1]
+        if self.labels_layer:
+            labels = outputs[self.labels_layer]
+        else:
+            labels = np.full(len(bboxes), self.default_label, dtype=bboxes.dtype)
+        labels = labels.squeeze(0)
+
+        return DetectionResult(
+            bboxes=bboxes,
+            labels=labels,
+            scores=scores,
+        )
+
+
 class SSD(DetectionModel):
     __model__ = "SSD"