ML-PIPES-SUPERVISION

Track Objects

Track detected objects across video frames to assign persistent IDs and analyze motion. This guide ports the Supervision object detection and segmentation workflow to ml-pipes; keypoint tracking is not yet covered by this package.

Download the source video used throughout the tutorial:

from supervision.assets import VideoAssets, download_assets

download_assets(VideoAssets.PEOPLE_WALKING)

Object Detection & Segmentation

Run Inference

Run detection on each video frame, then draw boxes on the resulting scene.

import supervision as sv

from ml_pipes.core import Pipeline
from ml_pipes.standard import Recall, Select, Store
from ml_pipes.supervision import BoxAnnotator, Detections
from ml_pipes.supervision.inference import RoboflowInference

pipeline = Pipeline(
    [
        Store("source_frame"),
        RoboflowInference(model_id="rfdetr-small"),
        Select(0),
        Detections.FromInference(),
        Recall("source_frame", prepend=True),
        BoxAnnotator(),
    ]
)

def callback(frame, _: int):
    annotated_frame, _ = pipeline(frame)
    return annotated_frame

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

import numpy as np
import supervision as sv
from inference.models.utils import get_roboflow_model

model = get_roboflow_model(model_id="rfdetr-small", api_key="<ROBOFLOW_API_KEY>")
box_annotator = sv.BoxAnnotator()

def callback(frame: np.ndarray, _: int) -> np.ndarray:
    results = model.infer(frame)[0]
    detections = sv.Detections.from_inference(results)
    return box_annotator.annotate(frame.copy(), detections=detections)

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

Tracking

After inference, update a stateful tracker with each frame's detections. The ml-pipes ByteTrack wrapper and the direct Supervision examples both use ByteTrackTracker from the external trackers package.

ByteTrack uses low-confidence detections during association and consumes sv.Detections rather than model-specific results; see the upstream trackers package for algorithm and tuning details.

ByteTrack tracks boxes, not mask geometry. For segmentation results, MaskAnnotator can color each object's mask by tracker ID.

import supervision as sv

from ml_pipes.core import Pipeline
from ml_pipes.standard import Recall, Select, Store
from ml_pipes.supervision import BoxAnnotator, Detections
from ml_pipes.supervision.inference import RoboflowInference
from ml_pipes.supervision.trackers import ByteTrack

pipeline = Pipeline(
    [
        Store("source_frame"),
        RoboflowInference(model_id="rfdetr-small"),
        Select(0),
        Detections.FromInference(),
        ByteTrack(track_activation_threshold=0.25, minimum_consecutive_frames=1),
        Recall("source_frame", prepend=True),
        BoxAnnotator(),
    ]
)

def callback(frame, _: int):
    annotated_frame, _ = pipeline(frame)
    return annotated_frame

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

import numpy as np
import supervision as sv
import trackers
from inference.models.utils import get_roboflow_model

model = get_roboflow_model(model_id="rfdetr-small", api_key="<ROBOFLOW_API_KEY>")
tracker = trackers.ByteTrackTracker(track_activation_threshold=0.25, minimum_consecutive_frames=1)
box_annotator = sv.BoxAnnotator()

def callback(frame: np.ndarray, _: int) -> np.ndarray:
    results = model.infer(frame)[0]
    detections = sv.Detections.from_inference(results)
    detections = tracker.update(detections)
    return box_annotator.annotate(frame.copy(), detections=detections)

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

Annotate Video with Tracking IDs

Add persistent IDs and class names with LabelAnnotator after the tracker updates the detections.

The tracker returns -1 for detections without a confirmed track. Filter those out after tracking, before drawing ID labels or traces, as in the upstream tutorial source.

import supervision as sv

from ml_pipes.core import Pipeline
from ml_pipes.standard import Recall, Select, Store
from ml_pipes.supervision import BoxAnnotator, Detections, LabelAnnotator
from ml_pipes.supervision.inference import RoboflowInference
from ml_pipes.supervision.trackers import ByteTrack

pipeline = Pipeline(
    [
        Store("source_frame"),
        RoboflowInference(model_id="rfdetr-small"),
        Select(0),
        Detections.FromInference(),
        ByteTrack(track_activation_threshold=0.25, minimum_consecutive_frames=1),
        Detections.Filter(lambda detections: detections.tracker_id != -1),
        Recall("source_frame", prepend=True),
        BoxAnnotator(),
        LabelAnnotator(show_tracker_id=True, show_class=True),
    ]
)

def callback(frame, _: int):
    annotated_frame, _ = pipeline(frame)
    return annotated_frame

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

import numpy as np
import supervision as sv
import trackers
from inference.models.utils import get_roboflow_model

model = get_roboflow_model(model_id="rfdetr-small", api_key="<ROBOFLOW_API_KEY>")
tracker = trackers.ByteTrackTracker(track_activation_threshold=0.25, minimum_consecutive_frames=1)
box_annotator = sv.BoxAnnotator()
label_annotator = sv.LabelAnnotator()

def callback(frame: np.ndarray, _: int) -> np.ndarray:
    results = model.infer(frame)[0]
    detections = sv.Detections.from_inference(results)
    detections = tracker.update(detections)
    detections = detections[detections.tracker_id != -1]

    labels = [
        f"#{tracker_id} {class_name}"
        for class_name, tracker_id
        in zip(detections.data["class_name"], detections.tracker_id)
    ]

    annotated_frame = box_annotator.annotate(
        frame.copy(), detections=detections)
    return label_annotator.annotate(
        annotated_frame, detections=detections, labels=labels)

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

Annotate Video with Traces

Draw each track's historical path with TraceAnnotator after the box and label layers.

import supervision as sv

from ml_pipes.core import Pipeline
from ml_pipes.standard import Recall, Select, Store
from ml_pipes.supervision import BoxAnnotator, Detections, LabelAnnotator, TraceAnnotator
from ml_pipes.supervision.inference import RoboflowInference
from ml_pipes.supervision.trackers import ByteTrack

pipeline = Pipeline(
    [
        Store("source_frame"),
        RoboflowInference(model_id="rfdetr-small"),
        Select(0),
        Detections.FromInference(),
        ByteTrack(track_activation_threshold=0.25, minimum_consecutive_frames=1),
        Detections.Filter(lambda detections: detections.tracker_id != -1),
        Recall("source_frame", prepend=True),
        BoxAnnotator(),
        LabelAnnotator(show_tracker_id=True, show_class=True),
        TraceAnnotator(),
    ]
)

def callback(frame, _: int):
    annotated_frame, _ = pipeline(frame)
    return annotated_frame

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

import numpy as np
import supervision as sv
import trackers
from inference.models.utils import get_roboflow_model

model = get_roboflow_model(model_id="rfdetr-small", api_key="<ROBOFLOW_API_KEY>")
tracker = trackers.ByteTrackTracker(track_activation_threshold=0.25, minimum_consecutive_frames=1)
box_annotator = sv.BoxAnnotator()
label_annotator = sv.LabelAnnotator()
trace_annotator = sv.TraceAnnotator()

def callback(frame: np.ndarray, _: int) -> np.ndarray:
    results = model.infer(frame)[0]
    detections = sv.Detections.from_inference(results)
    detections = tracker.update(detections)
    detections = detections[detections.tracker_id != -1]

    labels = [
        f"#{tracker_id} {class_name}"
        for class_name, tracker_id
        in zip(detections.data["class_name"], detections.tracker_id)
    ]

    annotated_frame = box_annotator.annotate(
        frame.copy(), detections=detections)
    annotated_frame = label_annotator.annotate(
        annotated_frame, detections=detections, labels=labels)
    return trace_annotator.annotate(
        annotated_frame, detections=detections)

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

Smooth Tracked Detections

For detection-only video, optionally place DetectionsSmoother after tracking and the confirmed-track filter, before annotation. It requires tracker IDs and does not support segmentation; length is the smoothing window in frames. See the upstream smoother reference.

Extend the previous trace-annotation flow with the highlighted smoothing stage. In ml-pipes, insert it before Recall, while the payload is still detections. Reuse the same pipeline or stateful objects for the whole video.

import supervision as sv

from ml_pipes.core import Pipeline
from ml_pipes.standard import Recall, Select, Store
from ml_pipes.supervision import BoxAnnotator, Detections, DetectionsSmoother, LabelAnnotator, TraceAnnotator
from ml_pipes.supervision.inference import RoboflowInference
from ml_pipes.supervision.trackers import ByteTrack

pipeline = Pipeline(
    [
        Store("source_frame"),
        RoboflowInference(model_id="rfdetr-small"),
        Select(0),
        Detections.FromInference(),
        ByteTrack(track_activation_threshold=0.25, minimum_consecutive_frames=1),
        Detections.Filter(lambda detections: detections.tracker_id != -1),
        DetectionsSmoother(length=5),
        Recall("source_frame", prepend=True),
        BoxAnnotator(),
        LabelAnnotator(show_tracker_id=True, show_class=True),
        TraceAnnotator(),
    ]
)

def callback(frame, _: int):
    annotated_frame, _ = pipeline(frame)
    return annotated_frame

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

import numpy as np
import supervision as sv
import trackers
from inference.models.utils import get_roboflow_model

model = get_roboflow_model(model_id="rfdetr-small", api_key="<ROBOFLOW_API_KEY>")
tracker = trackers.ByteTrackTracker(track_activation_threshold=0.25, minimum_consecutive_frames=1)
smoother = sv.DetectionsSmoother(length=5)
box_annotator = sv.BoxAnnotator()
label_annotator = sv.LabelAnnotator()
trace_annotator = sv.TraceAnnotator()

def callback(frame: np.ndarray, _: int) -> np.ndarray:
    results = model.infer(frame)[0]
    detections = sv.Detections.from_inference(results)
    detections = tracker.update(detections)
    detections = detections[detections.tracker_id != -1]
    detections = smoother.update_with_detections(detections)

    labels = [
        f"#{tracker_id} {class_name}"
        for class_name, tracker_id
        in zip(detections.data["class_name"], detections.tracker_id)
    ]

    annotated_frame = box_annotator.annotate(
        frame.copy(), detections=detections)
    annotated_frame = label_annotator.annotate(
        annotated_frame, detections=detections, labels=labels)
    return trace_annotator.annotate(
        annotated_frame, detections=detections)

sv.process_video(
    source_path="people-walking.mp4",
    target_path="result.mp4",
    callback=callback,
)

Authors