Merge pull request #559 from roboflow/allow-static-zone-in-perspectiv…

…e-correction-block

Allow zone defined as input parameter to be passed to perspective_correction

inference/core/version.py

@@ -1,4 +1,4 @@
-__version__ = "0.15.2"
+__version__ = "0.15.3"
 
 
 if __name__ == "__main__":

inference/core/workflows/core_steps/transformations/perspective_correction.py

@@ -1,10 +1,10 @@
 import math
-from typing import List, Literal, Optional, Type, Union
 
 import cv2 as cv
 import numpy as np
 import supervision as sv
 from pydantic import AliasChoices, ConfigDict, Field
+from typing_extensions import List, Literal, Optional, Type, Union
 
 from inference.core.logger import logger
 from inference.core.workflows.constants import KEYPOINTS_XY_KEY_IN_SV_DETECTIONS
@@ -57,13 +57,16 @@ class PerspectiveCorrectionManifest(WorkflowBlockManifest):
         }
     )
     type: Literal[f"{TYPE}"]
-    predictions: StepOutputSelector(
-        kind=[
-            BATCH_OF_OBJECT_DETECTION_PREDICTION_KIND,
-            BATCH_OF_INSTANCE_SEGMENTATION_PREDICTION_KIND,
-        ]
-    ) = Field(  # type: ignore
+    predictions: Optional[
+        StepOutputSelector(
+            kind=[
+                BATCH_OF_OBJECT_DETECTION_PREDICTION_KIND,
+                BATCH_OF_INSTANCE_SEGMENTATION_PREDICTION_KIND,
+            ]
+        )
+    ] = Field(  # type: ignore
         description="Predictions",
+        default=None,
         examples=["$steps.object_detection_model.predictions"],
     )
     images: Union[WorkflowImageSelector, StepOutputImageSelector] = Field(
@@ -72,7 +75,7 @@ class PerspectiveCorrectionManifest(WorkflowBlockManifest):
         examples=["$inputs.image", "$steps.cropping.crops"],
         validation_alias=AliasChoices("images", "image"),
     )
-    perspective_polygons: Union[list, StepOutputSelector(kind=[LIST_OF_VALUES_KIND])] = Field(  # type: ignore
+    perspective_polygons: Union[list, StepOutputSelector(kind=[LIST_OF_VALUES_KIND]), WorkflowParameterSelector(kind=[LIST_OF_VALUES_KIND])] = Field(  # type: ignore
         description="Perspective polygons (for each batch at least one must be consisting of 4 vertices)",
     )
     transformed_rect_width: Union[int, WorkflowParameterSelector(kind=[INTEGER_KIND])] = Field(  # type: ignore
@@ -127,6 +130,11 @@ def pick_largest_perspective_polygons(
         raise ValueError("Unexpected type of input")
     if not perspective_polygons_batch:
         raise ValueError("Unexpected empty batch")
+    if len(perspective_polygons_batch) == 4 and all(
+        isinstance(p, list) and len(p) == 2 for p in perspective_polygons_batch
+    ):
+        perspective_polygons_batch = [perspective_polygons_batch]
+
     largest_perspective_polygons: List[np.ndarray] = []
     for polygons in perspective_polygons_batch:
         if polygons is None:
@@ -223,9 +231,9 @@ def extend_perspective_polygon(
 
 def generate_transformation_matrix(
     src_polygon: np.ndarray,
-    detections: sv.Detections,
     transformed_rect_width: int,
     transformed_rect_height: int,
+    detections: Optional[sv.Detections] = None,
     detections_anchor: Optional[sv.Position] = None,
 ) -> np.ndarray:
     polygon_with_vertices_clockwise = sort_polygon_vertices_clockwise(
@@ -234,7 +242,7 @@ def generate_transformation_matrix(
     src_polygon = roll_polygon_vertices_to_start_from_leftmost_bottom(
         polygon=polygon_with_vertices_clockwise
     )
-    if detections_anchor:
+    if detections and detections_anchor:
         src_polygon = extend_perspective_polygon(
             polygon=src_polygon,
             detections=detections,
@@ -291,7 +299,7 @@ def correct_detections(
                 ]
             )
         else:
-            xmin, ymin, xmax, ymax = np.around(detection[i].xyxy[0]).tolist()
+            xmin, ymin, xmax, ymax = np.around(detection.xyxy[0]).tolist()
             polygon = np.array(
                 [[[xmin, ymin], [xmax, ymin], [xmax, ymax], [xmin, ymax]]],
                 dtype=np.float32,
@@ -333,22 +341,42 @@ def get_manifest(cls) -> Type[WorkflowBlockManifest]:
     async def run(
         self,
         images: Batch[WorkflowImageData],
-        predictions: Batch[sv.Detections],
-        perspective_polygons: Batch[List[np.ndarray]],
+        predictions: Optional[Batch[sv.Detections]],
+        perspective_polygons: Union[
+            List[np.ndarray],
+            List[List[np.ndarray]],
+            List[List[List[int]]],
+            List[List[List[List[int]]]],
+        ],
         transformed_rect_width: int,
         transformed_rect_height: int,
         extend_perspective_polygon_by_detections_anchor: Optional[str],
         warp_image: Optional[bool],
     ) -> BlockResult:
+        if not predictions and not images:
+            raise ValueError(
+                "Either predictions or images are required to apply perspective correction."
+            )
+        if warp_image and not images:
+            raise ValueError(
+                "images are required to warp image into requested perspective."
+            )
+        if not predictions:
+            predictions = [None] * len(images)
+
         if not self.perspective_transformers:
-            try:
-                largest_perspective_polygons = pick_largest_perspective_polygons(
-                    perspective_polygons
-                )
-            except ValueError as exc:
-                logger.error(exc)
-                largest_perspective_polygons = [None for _ in perspective_polygons]
+            largest_perspective_polygons = pick_largest_perspective_polygons(
+                perspective_polygons
+            )
 
+            batch_size = len(predictions) if predictions else len(images)
+            if len(largest_perspective_polygons) == 1 and batch_size > 1:
+                largest_perspective_polygons = largest_perspective_polygons * batch_size
+
+            if len(largest_perspective_polygons) != batch_size:
+                raise ValueError(
+                    f"Predictions batch size ({batch_size}) does not match number of perspective polygons ({largest_perspective_polygons})"
+                )
             for polygon, detections in zip(largest_perspective_polygons, predictions):
                 if polygon is None:
                     self.perspective_transformers.append(None)
@@ -376,7 +404,7 @@ async def run(
                     dsize=(transformed_rect_width, transformed_rect_height),
                 )
 
-            if detections is None or perspective_transformer is None:
+            if detections is None:
                 result.append(
                     {OUTPUT_DETECTIONS_KEY: None, OUTPUT_IMAGE_KEY: result_image}
                 )