Roboflow Video Inference with Customized Annotators

Performing real-time video inference is essential for a lot of purposes like autonomous autos, safety methods, logistics, and extra. Nevertheless, organising a sturdy video inference pipeline might be time consuming. You want a educated laptop imaginative and prescient mannequin, inference code, and visualization instruments to annotate every body. It is a lot of overhead simply to run a mannequin on video! On this publish, I am going to present how one can run scalable video inference utilizing Roboflow Universe, accessing customizable pre-trained fashions with only a few strains of Python.

We’ll use a Roboflow Logistics Object Detection Mannequin from Roboflow Universe and add customized annotations like packing containers, labels, monitoring, and extra on high of the detected objects utilizing the Supervision library.

Here’s a fast overview of the steps we’ll cowl:

1. Getting began
2. Load customized mannequin from Roboflow Universe
3. Run inference on a video
4. Apply Customized Annotations 
5. Annotate frames
6. Extract the annotated video utilizing Supervision

Use Case: Logistics Monitoring

Let us take a look at a use case in logistics – monitoring stock in warehouses and distribution facilities. Protecting observe of stock and property precisely is vital however could be a guide course of. We are able to use cameras to document video feeds inside warehouses. Then object detection fashions can establish and find vital property like pallets, packing containers, carts, and forklifts. Lastly, we will add customized overlays like labels, field outlines, and monitoring to create a list monitoring system.

Getting Began

First, we’ll import Roboflow and different dependencies:

import os
from roboflow import Roboflow
import numpy as np
import supervision as sv
import cv2

Roboflow supplies simple APIs for working with fashions and runs inference routinely within the background whereas Supervision incorporates annotators that allow us simply customise the output video.

Load Customized Mannequin from Roboflow Universe

Subsequent, we configure our Roboflow venture. You’ll be able to decide any deployed mannequin from Roboflow Universe. 

PROJECT_NAME = "logistics-sz9jr"
VIDEO_FILE = "instance.mp4"
ANNOTATED_VIDEO = "output.mp4" rf = Roboflow(api_key="ROBOFLOW_API_KEY")
venture = rf.workspace().venture(PROJECT_NAME)
mannequin = venture.model(2).mannequin

Run Video Inference

Then, we name the predict_video methodology to run inference on the video and get the outcomes:

job_id, signed_url, expire_time = mannequin.predict_video( VIDEO_FILE, fps=5, prediction_type="batch-video",
)
outcomes = mannequin.poll_until_video_results(job_id)

Apply Customized Annotations

To boost the standard of our annotations, we will make the most of customized annotators from the ‘supervision’ package deal. On this instance, we use ‘BoxMaskAnnotator’ and ‘LabelAnnotator’ to enhance our annotations:

box_mask_annotator = sv.BoxMaskAnnotator()
label_annotator = sv.LabelAnnotator()
tracker = sv.ByteTrack()
# box_annotator = sv.BoundingBoxAnnotator()
# halo_annotator = sv.HaloAnnotator()
# corner_annotator = sv.BoxCornerAnnotator()
# circle_annotator = sv.CircleAnnotator()
# blur_annotator = sv.BlurAnnotator()
# heat_map_annotator = sv.HeatMapAnnotator()

Annotate Frames

To annotate frames within the video, we’ll outline a perform referred to as `annotate_frame` that makes use of the obtained outcomes from video inference. 

Then for every body, we:

1. Get the detection information from Roboflow’s outcomes
2. Create annotations with Supervision for that body
3. Overlay annotations on the body

cap = cv2.VideoCapture(VIDEO_FILE)
frame_rate = cap.get(cv2.CAP_PROP_FPS)
cap.launch() def annotate_frame(body: np.ndarray, frame_number: int) -> np.ndarray: attempt: time_offset = frame_number / frame_rate closest_time_offset = min(outcomes['time_offset'], key=lambda t: abs(t - time_offset)) index = outcomes['time_offset'].index(closest_time_offset) detection_data = outcomes[PROJECT_NAME][index] roboflow_format = { "predictions": detection_data['predictions'], "picture": {"width": body.form[1], "top": body.form[0]} } detections = sv.Detections.from_roboflow(roboflow_format) detections = tracker.update_with_detections(detections) labels = [pred['class'] for pred in detection_data['predictions']] besides (IndexError, KeyError, ValueError) as e: print(f"Exception in processing body {frame_number}: {e}") detections = sv.Detections(xyxy=np.empty((0, 4)), confidence=np.empty(0), class_id=np.empty(0, dtype=int)) labels = [] annotated_frame = box_mask_annotator.annotate(body.copy(), detections=detections) annotated_frame = label_annotator.annotate(annotated_frame, detections=detections, labels=labels) return annotated_frame

Write Output Video

Lastly, we annotate the total video:

sv.process_video( source_path=VIDEO_FILE, target_path=ANNOTATED_VIDEO, callback=annotate_frame
)

Conclusion

Roboflow mixed with Supervision supplies an end-to-end system to simply add visible intelligence to video feeds. This helps construct production-ready options for stock administration, warehouse monitoring and extra.

The complete code is accessible on Colab. Blissful inferring!