Skip to content

Commit

Permalink
push prediction to annotation converter to postprocessing module
Browse files Browse the repository at this point in the history 
Signed-off-by: Igor Davidyuk <igor.davidyuk@intel.com>
  • Loading branch information
@igor-davidyuk
igor-davidyuk committed Mar 6, 2024
1 parent 5f5602b commit c8a55e4
Show file tree
Hide file tree
Showing 2 changed files with 162 additions and 114 deletions.
131 changes: 18 additions & 113 deletions geti_sdk/deployment/deployed_model.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,17 +33,16 @@
from otx.api.entities.label_schema import LabelGroup, LabelGroupType, LabelSchemaEntity

from geti_sdk.data_models import OptimizedModel, Project, TaskConfiguration
from geti_sdk.data_models.annotations import Annotation
from geti_sdk.data_models.label import ScoredLabel

# from geti_sdk.data_models.annotations import Annotation
# from geti_sdk.data_models.label import ScoredLabel
from geti_sdk.data_models.predictions import Prediction, ResultMedium
from geti_sdk.data_models.shapes import Polygon, Rectangle, RotatedRectangle
from geti_sdk.data_models.task import Task, TaskType
from geti_sdk.deployment.legacy_converters import (
AnomalyClassificationToAnnotationConverter,
)
from geti_sdk.deployment.predictions_postprocessing.postprocessing import (
detection2array,
)
from geti_sdk.data_models.task import Task

# from geti_sdk.deployment.legacy_converters import (
# AnomalyClassificationToAnnotationConverter,
# )
from geti_sdk.deployment.predictions_postprocessing.postprocessing import Postprocessor
from geti_sdk.http_session import GetiSession
from geti_sdk.rest_converters import ConfigurationRESTConverter, ModelRESTConverter

Expand Down Expand Up @@ -101,6 +100,8 @@ def __attrs_post_init__(self):

self.openvino_model_parameters: Optional[Dict[str, Any]] = None

self._postprocessor: Optional[Postprocessor] = None

@property
def model_data_path(self) -> str:
"""
Expand Down Expand Up @@ -593,110 +594,14 @@ def infer(self, image: np.ndarray, task: Task, explain: bool = False) -> Predict
)
postprocessing_results = self._postprocess(inference_results, metadata=metadata)

# Create a converter
try:
import otx
from otx.api.usecases.exportable_code.prediction_to_annotation_converter import (
IPredictionToAnnotationConverter,
create_converter,
)
except ImportError as error:
raise ValueError(
f"Unable to load inference model for {self}. Relevant OpenVINO "
f"packages were not found. Please make sure that all packages from the "
f"file `requirements-deployment.txt` have been installed. "
) from error
if otx.__version__ > "1.2.0":
configuration = self.openvino_model_parameters
if "use_ellipse_shapes" not in configuration.keys():
configuration.update({"use_ellipse_shapes": False})
converter_args = {
"labels": self.ote_label_schema,
"configuration": configuration,
}
else:
converter_args = {"labels": self.ote_label_schema}

converter: IPredictionToAnnotationConverter = create_converter(
converter_type=task.type.to_ote_domain(), **converter_args
)

# Proceed with postprocessing
width: int = image.shape[1]
height: int = image.shape[0]

# Handle empty annotations
if isinstance(postprocessing_results, (np.ndarray, list)):
try:
n_outputs = len(postprocessing_results)
except TypeError:
n_outputs = 1
else:
# Handle the new modelAPI output formats for detection and instance
# segmentation models
if (
hasattr(postprocessing_results, "objects")
and task.type == TaskType.DETECTION
):
n_outputs = len(postprocessing_results.objects)
postprocessing_results = detection2array(postprocessing_results.objects)
elif hasattr(postprocessing_results, "segmentedObjects") and task.type in [
TaskType.INSTANCE_SEGMENTATION,
TaskType.ROTATED_DETECTION,
]:
n_outputs = len(postprocessing_results.segmentedObjects)
postprocessing_results = postprocessing_results.segmentedObjects
elif isinstance(postprocessing_results, tuple):
try:
n_outputs = len(postprocessing_results)
except TypeError:
n_outputs = 1
else:
raise ValueError(
f"Unknown postprocessing output of type "
f"`{type(postprocessing_results)}` for task `{task.title}`."
)

if n_outputs != 0:
try:
annotation_scene_entity = converter.convert_to_annotation(
predictions=postprocessing_results, metadata=metadata
)
except AttributeError:
# Add backwards compatibility for anomaly models created in Geti v1.8 and below
if task.type.is_anomaly:
legacy_converter = AnomalyClassificationToAnnotationConverter(
label_schema=self.ote_label_schema
)
annotation_scene_entity = legacy_converter.convert_to_annotation(
predictions=postprocessing_results, metadata=metadata
)
converter = legacy_converter

prediction = Prediction.from_ote(
annotation_scene_entity, image_width=width, image_height=height
# Create a postprocessor
if self._postprocessor is None:
self._postprocessor = Postprocessor(
labels=self.ote_label_schema,
configuration=self.openvino_model_parameters,
task=task,
)
else:
prediction = Prediction(annotations=[])

# Empty label is not generated by OTE correctly, append it here if there are
# no other predictions
if len(prediction.annotations) == 0:
empty_label = next((label for label in task.labels if label.is_empty), None)
if empty_label is not None:
prediction.append(
Annotation(
shape=Rectangle(x=0, y=0, width=width, height=height),
labels=[ScoredLabel.from_label(empty_label, probability=1)],
)
)

# Rotated detection models produce Polygons, convert them here to
# RotatedRectangles
if task.type == TaskType.ROTATED_DETECTION:
for annotation in prediction.annotations:
if isinstance(annotation.shape, Polygon):
annotation.shape = RotatedRectangle.from_polygon(annotation.shape)
prediction = self._postprocessor(postprocessing_results, image, metadata)

# Add optional explainability outputs
if explain:
Expand Down
145 changes: 144 additions & 1 deletion geti_sdk/deployment/predictions_postprocessing/postprocessing.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,26 @@
# See the License for the specific language governing permissions
# and limitations under the License.

from typing import List
"""Module implements the Postprocessor class."""

from typing import Dict, List, Tuple

import numpy as np
import otx
from otx.api.usecases.exportable_code.prediction_to_annotation_converter import (
IPredictionToAnnotationConverter,
create_converter,
)

from geti_sdk.data_models.annotations import Annotation
from geti_sdk.data_models.enums.task_type import TaskType
from geti_sdk.data_models.label import ScoredLabel
from geti_sdk.data_models.predictions import Prediction
from geti_sdk.data_models.shapes import Polygon, Rectangle, RotatedRectangle
from geti_sdk.data_models.task import Task
from geti_sdk.deployment.legacy_converters.legacy_anomaly_converter import (
AnomalyClassificationToAnnotationConverter,
)


def detection2array(detections: List) -> np.ndarray:
Expand All @@ -40,3 +57,129 @@ def detection2array(detections: List) -> np.ndarray:
)
detections = np.concatenate((labels, scores, boxes), -1)
return detections


class Postprocessor:
"""
Postprocessor class responsible for converting the output of the model to a Prediction object.
:param labels: Label schema to be used for the conversion.
:param configuration: Configuration to be used for the conversion.
:param task: Task object containing the task metadata.
"""

def __init__(self, labels, configuration, task: Task) -> None:
self.task = task
self.ote_label_schema = labels

# Create OTX converter
converter_args = {"labels": labels}
if otx.__version__ > "1.2.0":
if "use_ellipse_shapes" not in configuration.keys():
configuration.update({"use_ellipse_shapes": False})
converter_args["configuration"] = configuration

self.converter: IPredictionToAnnotationConverter = create_converter(
converter_type=self.task.type.to_ote_domain(), **converter_args
)

def __call__(
self,
postprocessing_results: List,
image: np.ndarray,
metadata: Dict[str, Tuple[int, int, int]],
) -> Prediction:
"""
Convert the postprocessing results to a Prediction object.
"""
# Handle empty annotations
if isinstance(postprocessing_results, (np.ndarray, list)):
try:
n_outputs = len(postprocessing_results)
except TypeError:
n_outputs = 1
else:
# Handle the new modelAPI output formats for detection and instance
# segmentation models
if (
hasattr(postprocessing_results, "objects")
and self.task.type == TaskType.DETECTION
):
n_outputs = len(postprocessing_results.objects)
postprocessing_results = detection2array(postprocessing_results.objects)
elif hasattr(
postprocessing_results, "segmentedObjects"
) and self.task.type in [
TaskType.INSTANCE_SEGMENTATION,
TaskType.ROTATED_DETECTION,
]:
n_outputs = len(postprocessing_results.segmentedObjects)
postprocessing_results = postprocessing_results.segmentedObjects
elif isinstance(postprocessing_results, tuple):
try:
n_outputs = len(postprocessing_results)
except TypeError:
n_outputs = 1
else:
raise ValueError(
f"Unknown postprocessing output of type "
f"`{type(postprocessing_results)}` for task `{self.task.title}`."
)

# Proceed with postprocessing
width: int = image.shape[1]
height: int = image.shape[0]

if n_outputs != 0:
try:
annotation_scene_entity = self.converter.convert_to_annotation(
predictions=postprocessing_results, metadata=metadata
)
except AttributeError:
# Add backwards compatibility for anomaly models created in Geti v1.8 and below
if self.task.type.is_anomaly:
legacy_converter = AnomalyClassificationToAnnotationConverter(
label_schema=self.ote_label_schema
)
annotation_scene_entity = legacy_converter.convert_to_annotation(
predictions=postprocessing_results, metadata=metadata
)
self.converter = legacy_converter

prediction = Prediction.from_ote(
annotation_scene_entity, image_width=width, image_height=height
)
else:
prediction = Prediction(annotations=[])

print(
"pre-converter",
postprocessing_results,
"metadata",
metadata,
)
print("width", width, "height", height)
print("post-converter", prediction)

# Empty label is not generated by OTE correctly, append it here if there are
# no other predictions
if len(prediction.annotations) == 0:
empty_label = next(
(label for label in self.task.labels if label.is_empty), None
)
if empty_label is not None:
prediction.append(
Annotation(
shape=Rectangle(x=0, y=0, width=width, height=height),
labels=[ScoredLabel.from_label(empty_label, probability=1)],
)
)

# Rotated detection models produce Polygons, convert them here to
# RotatedRectangles
if self.task.type == TaskType.ROTATED_DETECTION:
for annotation in prediction.annotations:
if isinstance(annotation.shape, Polygon):
annotation.shape = RotatedRectangle.from_polygon(annotation.shape)

return prediction

0 comments on commit c8a55e4

Please sign in to comment.