Add GuardTrackedDetectionsBlock

[grzegorz-roboflow]

Description

When processing videos ByteTracker (and other trackers) may loose track of object for just a few frames. This results in flickering

Type of change

• New feature (non-breaking change which adds functionality)

How has this change been tested, please provide a testcase or example of how you tested the change?

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.transformations.guard_tracked_detections.v1 import OUTPUT_KEY as GUARD_TRACKED_DETECTIONS_OUTPUT_KEY
from inference.core.workflows.core_steps.analytics.line_counter.v1 import OUTPUT_KEY_COUNT_IN, OUTPUT_KEY_COUNT_OUT
from inference.core.workflows.core_steps.visualizations.common.base import OUTPUT_IMAGE_KEY
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"},
        {"type": "WorkflowVideoMetadata", "name": "video_metadata"},
    ],
    "steps": [
        {
            "type": "ObjectDetectionModel",
            "name": "people_detector",
            "image": "$inputs.image",
            "model_id": "yolov8n-640",
            "confidence": 0.3,
        },
        {
            "type": "roboflow_core/byte_tracker@v1",
            "name": "byte_tracker",
            "detections": "$steps.people_detector.predictions",
            "metadata": "$inputs.video_metadata"
        },
        {
            "type": "roboflow_core/guard_tracked_detections@v1",
            "name": "guard_detections",
            "detections": f"$steps.byte_tracker.{BYTE_TRACKER_OUTPUT_KEY}",
            "metadata": "$inputs.video_metadata",
            "consider_detection_gone_timeout": 0.5
        },
        {
            "type": "roboflow_core/bounding_box_visualization@v1",
            "name": "bbox_visualisation",
            "predictions": f"$steps.guard_detections.{GUARD_TRACKED_DETECTIONS_OUTPUT_KEY}",
            "image": "$inputs.image",
            "copy_image": False,
        },
    ],
    "outputs": [
        {
            "type": "JsonField",
            "name": "annotated_frame",
            "selector": f"$steps.bbox_visualisation.{OUTPUT_IMAGE_KEY}",
        }
    ],
}


box_annotator = sv.BoundingBoxAnnotator()
label_annotator = sv.LabelAnnotator()


frame_reader = cv.VideoCapture()
frame_reader.open("/Users/grzegorzklimaszewski/Downloads/traffic.mp4", apiPreference=cv.CAP_FFMPEG)

fps = float(frame_reader.get(cv.CAP_PROP_FPS))
width = int(frame_reader.get(cv.CAP_PROP_FRAME_WIDTH))
height = int(frame_reader.get(cv.CAP_PROP_FRAME_HEIGHT))
#fourcc = cv.VideoWriter_fourcc(*"MP4V")
fourcc = cv.VideoWriter_fourcc(*"AVC1")

frame_writer = cv.VideoWriter()
frame_writer.open("/path/to/results.mp4", fps=fps, fourcc=fourcc, apiPreference=cv.CAP_FFMPEG, frameSize=(width, height))

frame_reader.release()

def custom_sink(prediction: Dict[str, Union[Any, sv.Detections]], video_frame: VideoFrame) -> None:
    frame_writer.write(prediction["annotated_frame"].numpy_image)
    cv.imshow("", prediction["annotated_frame"].numpy_image)
    cv.waitKey(1)


pipeline = InferencePipeline.init_with_workflow(
    video_reference=0,
    workflow_specification=WORKFLOW,
    on_prediction=custom_sink,
    workflows_parameters={
    },
)
pipeline.start()
pipeline.join()
frame_writer.release()

Any specific deployment considerations

N/A

Docs

N/A

[PawelPeczek-Roboflow]

Could you please sync with @EmilyGavrilenko regarding naming conventions and descriptions?