Add a path deviation analytics block to workflows

development/stream_interface/path_deviation_demo.py

@@ -0,0 +1,75 @@
+import os
+from typing import Any, Dict, Union
+
+import cv2 as cv
+
+from inference.core.interfaces.camera.entities import VideoFrame
+from inference.core.interfaces.stream.inference_pipeline import InferencePipeline
+from inference.core.managers.base import ModelManager
+from inference.core.registries.roboflow import (
+    RoboflowModelRegistry,
+)
+from inference.core.workflows.core_steps.transformations.byte_tracker.v1 import OUTPUT_KEY as BYTE_TRACKER_OUTPUT_KEY
+from inference.core.workflows.core_steps.analytics.path_deviation.v1 import OUTPUT_KEY as PATH_DEVIATION_OUTPUT
+from inference.models.utils import ROBOFLOW_MODEL_TYPES
+import supervision as sv
+
+
+model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
+model_manager = ModelManager(model_registry=model_registry)
+
+
+WORKFLOW = {
+    "version": "1.0",
+    "inputs": [
+        {"type": "WorkflowImage", "name": "image", "video_metadata_input_name": "test"},
+        {"type": "WorkflowVideoMetadata", "name": "video_metadata"},
+        {"type": "InferenceParameter", "name": "reference_path"},
+    ],
+    "steps": [
+        {
+            "type": "ObjectDetectionModel",
+            "name": "people_detector",
+            "image": "$inputs.image",
+            "model_id": "yolov8n-640",
+        },
+        {
+            "type": "roboflow_core/byte_tracker@v1",
+            "name": "byte_tracker",
+            "detections": "$steps.people_detector.predictions",
+            "metadata": "$inputs.video_metadata"
+        },
+        {
+            "type": "roboflow_core/path_deviation_analytics@v1",
+            "name": "path_deviation",
+            "detections": f"$steps.byte_tracker.{BYTE_TRACKER_OUTPUT_KEY}",
+            "metadata": "$inputs.video_metadata",
+            "reference_path": "$inputs.reference_path",
+        }
+    ],
+    "outputs": [
+        {
+            "type": "JsonField",
+            "name": "predictions",
+            "selector": f"$steps.path_deviation.{PATH_DEVIATION_OUTPUT}",
+        }
+    ],
+}
+
+
+def custom_sink(prediction: Dict[str, Union[Any, sv.Detections]], video_frame: VideoFrame) -> None:
+    cv.imshow("", video_frame)
+    cv.waitKey(1)
+
+
+pipeline = InferencePipeline.init_with_workflow(
+    video_reference="/Users/grzegorzklimaszewski/Downloads/ball-track.mp4",
+    workflow_specification=WORKFLOW,
+    on_prediction=custom_sink,
+    workflows_parameters={
+        "reference_path": [[1, 2], [3, 4], [5, 6]]
+    },
+    video_metadata_input_name="video_metadata",
+)
+pipeline.start()
+pipeline.join()

inference/core/workflows/core_steps/analytics/path_deviation/v1.py

@@ -0,0 +1,178 @@
+from typing import Dict, List, Optional, Tuple, Union
+
+import numpy as np
+import supervision as sv
+from pydantic import ConfigDict, Field
+from typing_extensions import Literal, Type
+
+from inference.core.workflows.execution_engine.constants import (
+    PATH_DEVIATION_KEY_IN_SV_DETECTIONS,
+)
+from inference.core.workflows.execution_engine.entities.base import (
+    OutputDefinition,
+    VideoMetadata,
+)
+from inference.core.workflows.execution_engine.entities.types import (
+    INSTANCE_SEGMENTATION_PREDICTION_KIND,
+    LIST_OF_VALUES_KIND,
+    OBJECT_DETECTION_PREDICTION_KIND,
+    STRING_KIND,
+    StepOutputSelector,
+    WorkflowParameterSelector,
+    WorkflowVideoMetadataSelector,
+)
+from inference.core.workflows.prototypes.block import (
+    BlockResult,
+    WorkflowBlock,
+    WorkflowBlockManifest,
+)
+
+OUTPUT_KEY: str = "path_deviation_detections"
+SHORT_DESCRIPTION = "Calculate Fréchet distance of object from reference path"
+LONG_DESCRIPTION = """
+The `PathDeviationAnalyticsBlock` is an analytics block designed to measure the Frechet distance
+of tracked objects from a user-defined reference path. The block requires detections to be tracked
+(i.e. each object must have a unique tracker_id assigned, which persists between frames).
+"""
+
+
+class PathDeviationManifest(WorkflowBlockManifest):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "name": "Path deviation",
+            "version": "v1",
+            "short_description": SHORT_DESCRIPTION,
+            "long_description": LONG_DESCRIPTION,
+            "license": "Apache-2.0",
+            "block_type": "analytics",
+        }
+    )
+    type: Literal["roboflow_core/path_deviation_analytics@v1"]
+    metadata: WorkflowVideoMetadataSelector
+    detections: StepOutputSelector(
+        kind=[
+            OBJECT_DETECTION_PREDICTION_KIND,
+            INSTANCE_SEGMENTATION_PREDICTION_KIND,
+        ]
+    ) = Field(  # type: ignore
+        description="Predictions",
+        examples=["$steps.object_detection_model.predictions"],
+    )
+    triggering_anchor: Union[str, WorkflowParameterSelector(kind=[STRING_KIND]), Literal[tuple(sv.Position.list())]] = Field(  # type: ignore
+        description=f"Triggering anchor. Allowed values: {', '.join(sv.Position.list())}",
+        default="CENTER",
+        examples=["CENTER"],
+    )
+    reference_path: Union[list, StepOutputSelector(kind=[LIST_OF_VALUES_KIND]), WorkflowParameterSelector(kind=[LIST_OF_VALUES_KIND])] = Field(  # type: ignore
+        description="Reference path in a format [(x1, y1), (x2, y2), (x3, y3), ...]",
+        examples=["$inputs.expected_path"],
+    )
+
+    @classmethod
+    def describe_outputs(cls) -> List[OutputDefinition]:
+        return [
+            OutputDefinition(
+                name=OUTPUT_KEY,
+                kind=[
+                    OBJECT_DETECTION_PREDICTION_KIND,
+                    INSTANCE_SEGMENTATION_PREDICTION_KIND,
+                ],
+            ),
+        ]
+
+    @classmethod
+    def get_execution_engine_compatibility(cls) -> Optional[str]:
+        return ">=1.0.0,<2.0.0"
+
+
+class PathDeviationAnalyticsBlockV1(WorkflowBlock):
+    def __init__(self):
+        self._object_paths: Dict[
+            str, Dict[Union[int, str], List[Tuple[float, float]]]
+        ] = {}
+
+    @classmethod
+    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+        return PathDeviationManifest
+
+    def run(
+        self,
+        detections: sv.Detections,
+        metadata: VideoMetadata,
+        triggering_anchor: str,
+        reference_path: List[Tuple[int, int]],
+    ) -> BlockResult:
+        if detections.tracker_id is None:
+            raise ValueError(
+                f"tracker_id not initialized, {self.__class__.__name__} requires detections to be tracked"
+            )
+
+        video_id = metadata.video_identifier
+        if video_id not in self._object_paths:
+            self._object_paths[video_id] = {}
+
+        anchor_points = detections.get_anchors_coordinates(anchor=triggering_anchor)
+        result_detections = []
+        for i, tracker_id in enumerate(detections.tracker_id):
+            detection = detections[i]
+            anchor_point = anchor_points[i]
+            if tracker_id not in self._object_paths[video_id]:
+                self._object_paths[video_id][tracker_id] = []
+            self._object_paths[video_id][tracker_id].append(anchor_point)
+
+            object_path = np.array(self._object_paths[video_id][tracker_id])
+            ref_path = np.array(reference_path)
+
+            frechet_distance = self._calculate_frechet_distance(object_path, ref_path)
+            detection[PATH_DEVIATION_KEY_IN_SV_DETECTIONS] = np.array(
+                [frechet_distance], dtype=np.float64
+            )
+            result_detections.append(detection)
+
+        return {OUTPUT_KEY: sv.Detections.merge(result_detections)}
+
+    def _calculate_frechet_distance(
+        self, path1: np.ndarray, path2: np.ndarray
+    ) -> float:
+        dist_matrix = np.ones((len(path1), len(path2))) * -1
+        return self._compute_distance(
+            dist_matrix, len(path1) - 1, len(path2) - 1, path1, path2
+        )
+
+    def _compute_distance(
+        self,
+        dist_matrix: np.ndarray,
+        i: int,
+        j: int,
+        path1: np.ndarray,
+        path2: np.ndarray,
+    ) -> float:
+        if dist_matrix[i, j] > -1:
+            return dist_matrix[i, j]
+        elif i == 0 and j == 0:
+            dist_matrix[i, j] = self._euclidean_distance(path1[0], path2[0])
+        elif i > 0 and j == 0:
+            dist_matrix[i, j] = max(
+                self._compute_distance(dist_matrix, i - 1, 0, path1, path2),
+                self._euclidean_distance(path1[i], path2[0]),
+            )
+        elif i == 0 and j > 0:
+            dist_matrix[i, j] = max(
+                self._compute_distance(dist_matrix, 0, j - 1, path1, path2),
+                self._euclidean_distance(path1[0], path2[j]),
+            )
+        elif i > 0 and j > 0:
+            dist_matrix[i, j] = max(
+                min(
+                    self._compute_distance(dist_matrix, i - 1, j, path1, path2),
+                    self._compute_distance(dist_matrix, i - 1, j - 1, path1, path2),
+                    self._compute_distance(dist_matrix, i, j - 1, path1, path2),
+                ),
+                self._euclidean_distance(path1[i], path2[j]),
+            )
+        else:
+            dist_matrix[i, j] = float("inf")
+        return dist_matrix[i, j]
+
+    def _euclidean_distance(self, point1: np.ndarray, point2: np.ndarray) -> float:
+        return np.sqrt(np.sum((point1 - point2) ** 2))

inference/core/workflows/core_steps/analytics/time_in_zone/v1.py

@@ -5,6 +5,9 @@
 from pydantic import ConfigDict, Field
 from typing_extensions import Literal, Type
 
+from inference.core.workflows.execution_engine.constants import (
+    TIME_IN_ZONE_KEY_IN_SV_DETECTIONS,
+)
 from inference.core.workflows.execution_engine.entities.base import (
     OutputDefinition,
     VideoMetadata,
@@ -29,7 +32,6 @@
 )
 
 OUTPUT_KEY: str = "timed_detections"
-DETECTIONS_TIME_IN_ZONE_PARAM: str = "time_in_zone"
 SHORT_DESCRIPTION = "Track duration of time spent by objects in zone"
 LONG_DESCRIPTION = """
 The `TimeInZoneBlock` is an analytics block designed to measure time spent by objects in a zone.
@@ -175,10 +177,12 @@ def run(
             # copy
             detection = detections[i]
 
-            detection[DETECTIONS_TIME_IN_ZONE_PARAM] = np.array([0], dtype=np.float64)
+            detection[TIME_IN_ZONE_KEY_IN_SV_DETECTIONS] = np.array(
+                [0], dtype=np.float64
+            )
             if is_in_zone:
                 ts_start = tracked_ids_in_zone.setdefault(tracker_id, ts_end)
-                detection[DETECTIONS_TIME_IN_ZONE_PARAM] = np.array(
+                detection[TIME_IN_ZONE_KEY_IN_SV_DETECTIONS] = np.array(
                     [ts_end - ts_start], dtype=np.float64
                 )
             elif tracker_id in tracked_ids_in_zone:

inference/core/workflows/core_steps/common/serializers.py

@@ -4,6 +4,14 @@
 import supervision as sv
 
 from inference.core.workflows.execution_engine.constants import (
+    BOUNDING_RECT_ANGLE_KEY_IN_INFERENCE_RESPONSE,
+    BOUNDING_RECT_ANGLE_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_HEIGHT_KEY_IN_INFERENCE_RESPONSE,
+    BOUNDING_RECT_HEIGHT_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_RECT_KEY_IN_INFERENCE_RESPONSE,
+    BOUNDING_RECT_RECT_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_WIDTH_KEY_IN_INFERENCE_RESPONSE,
+    BOUNDING_RECT_WIDTH_KEY_IN_SV_DETECTIONS,
     CLASS_ID_KEY,
     CLASS_NAME_KEY,
     CONFIDENCE_KEY,
@@ -17,7 +25,11 @@
     KEYPOINTS_KEY_IN_INFERENCE_RESPONSE,
     KEYPOINTS_XY_KEY_IN_SV_DETECTIONS,
     PARENT_ID_KEY,
+    PATH_DEVIATION_KEY_IN_INFERENCE_RESPONSE,
+    PATH_DEVIATION_KEY_IN_SV_DETECTIONS,
     POLYGON_KEY,
+    TIME_IN_ZONE_KEY_IN_INFERENCE_RESPONSE,
+    TIME_IN_ZONE_KEY_IN_SV_DETECTIONS,
     TRACKER_ID_KEY,
     WIDTH_KEY,
     X_KEY,
@@ -56,6 +68,32 @@ def serialise_sv_detections(detections: sv.Detections) -> dict:
             detection_dict[TRACKER_ID_KEY] = int(tracker_id)
         detection_dict[CLASS_NAME_KEY] = str(data["class_name"])
         detection_dict[DETECTION_ID_KEY] = str(data[DETECTION_ID_KEY])
+        if PATH_DEVIATION_KEY_IN_SV_DETECTIONS in data:
+            detection_dict[PATH_DEVIATION_KEY_IN_INFERENCE_RESPONSE] = data[
+                PATH_DEVIATION_KEY_IN_SV_DETECTIONS
+            ]
+        if TIME_IN_ZONE_KEY_IN_SV_DETECTIONS in data:
+            detection_dict[TIME_IN_ZONE_KEY_IN_INFERENCE_RESPONSE] = data[
+                TIME_IN_ZONE_KEY_IN_SV_DETECTIONS
+            ]
+        if (
+            BOUNDING_RECT_ANGLE_KEY_IN_SV_DETECTIONS in data
+            and BOUNDING_RECT_RECT_KEY_IN_SV_DETECTIONS in data
+            and BOUNDING_RECT_HEIGHT_KEY_IN_SV_DETECTIONS in data
+            and BOUNDING_RECT_WIDTH_KEY_IN_SV_DETECTIONS in data
+        ):
+            detection_dict[BOUNDING_RECT_ANGLE_KEY_IN_INFERENCE_RESPONSE] = data[
+                BOUNDING_RECT_ANGLE_KEY_IN_SV_DETECTIONS
+            ]
+            detection_dict[BOUNDING_RECT_RECT_KEY_IN_INFERENCE_RESPONSE] = data[
+                BOUNDING_RECT_RECT_KEY_IN_SV_DETECTIONS
+            ]
+            detection_dict[BOUNDING_RECT_HEIGHT_KEY_IN_INFERENCE_RESPONSE] = data[
+                BOUNDING_RECT_HEIGHT_KEY_IN_SV_DETECTIONS
+            ]
+            detection_dict[BOUNDING_RECT_WIDTH_KEY_IN_INFERENCE_RESPONSE] = data[
+                BOUNDING_RECT_WIDTH_KEY_IN_SV_DETECTIONS
+            ]
         if PARENT_ID_KEY in data:
             detection_dict[PARENT_ID_KEY] = str(data[PARENT_ID_KEY])
         if (

inference/core/workflows/core_steps/transformations/bounding_rect/v1.py

@@ -5,6 +5,12 @@
 import supervision as sv
 from pydantic import ConfigDict, Field
 
+from inference.core.workflows.execution_engine.constants import (
+    BOUNDING_RECT_ANGLE_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_HEIGHT_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_RECT_KEY_IN_SV_DETECTIONS,
+    BOUNDING_RECT_WIDTH_KEY_IN_SV_DETECTIONS,
+)
 from inference.core.workflows.execution_engine.entities.base import OutputDefinition
 from inference.core.workflows.execution_engine.entities.types import (
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
@@ -17,10 +23,6 @@
 )
 
 OUTPUT_KEY: str = "detections_with_rect"
-DETECTIONS_RECT_PARAM: str = "rect"
-DETECTIONS_WIDTH_PARAM: str = "width"
-DETECTIONS_HEIGHT_PARAM: str = "height"
-DETECTIONS_ANGLE_PARAM: str = "angle"
 
 SHORT_DESCRIPTION = "Find minimal bounding rectangle surrounding detection contour"
 LONG_DESCRIPTION = """
@@ -108,10 +110,18 @@ def run(
                 det.mask[0]
             )
 
-            det[DETECTIONS_RECT_PARAM] = np.array([rect], dtype=np.float16)
-            det[DETECTIONS_WIDTH_PARAM] = np.array([width], dtype=np.float16)
-            det[DETECTIONS_HEIGHT_PARAM] = np.array([height], dtype=np.float16)
-            det[DETECTIONS_ANGLE_PARAM] = np.array([angle], dtype=np.float16)
+            det[BOUNDING_RECT_RECT_KEY_IN_SV_DETECTIONS] = np.array(
+                [rect], dtype=np.float16
+            )
+            det[BOUNDING_RECT_WIDTH_KEY_IN_SV_DETECTIONS] = np.array(
+                [width], dtype=np.float16
+            )
+            det[BOUNDING_RECT_HEIGHT_KEY_IN_SV_DETECTIONS] = np.array(
+                [height], dtype=np.float16
+            )
+            det[BOUNDING_RECT_ANGLE_KEY_IN_SV_DETECTIONS] = np.array(
+                [angle], dtype=np.float16
+            )
 
             to_be_merged.append(det)