Object tracking sample application

This recipe detects and tracks vehicles in a video and writes an annotated copy to output.avi. It uses the external videoflow-contrib package for the TensorFlow detector and the SORT tracker. See the full script in examples/object_tracking.py.

import numpy as np
import videoflow
from videoflow.core import Flow
from videoflow.core.constants import BATCH
from videoflow.consumers import VideofileWriter
from videoflow.producers import VideofileReader
from videoflow.processors.vision.annotators import TrackerAnnotator
from videoflow.utils.downloader import get_file

URL_VIDEO = "https://github.com/videoflow/videoflow/releases/download/examples/intersection.mp4"

class BoundingBoxesFilter(videoflow.core.node.ProcessorNode):
    def __init__(self, class_indexes_to_keep, **kwargs):
        self._class_indexes_to_keep = class_indexes_to_keep
        super().__init__(**kwargs)

    def process(self, dets):
        f = np.array([dets[:, 4] == a for a in self._class_indexes_to_keep])
        f = np.any(f, axis=0)
        return dets[f]

class FrameIndexSplitter(videoflow.core.node.ProcessorNode):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def process(self, data):
        index, frame = data
        return frame

def build_flow():
    from videoflow_contrib.detector_tf import TensorflowObjectDetector
    from videoflow_contrib.tracker_sort import KalmanFilterBoundingBoxTracker

    input_file = get_file("intersection.mp4", URL_VIDEO)
    reader    = VideofileReader(input_file, name='reader')
    frame     = FrameIndexSplitter(name='frame')(reader)
    detector  = TensorflowObjectDetector(name='detector')(frame)
    # keep only vehicle classes
    filtered  = BoundingBoxesFilter([1, 2, 3, 4, 6, 8, 10, 13], name='filter')(detector)
    tracker   = KalmanFilterBoundingBoxTracker(name='tracker')(filtered)
    annotator = TrackerAnnotator(name='annotator')(frame, tracker)
    writer    = VideofileWriter("output.avi", fps=30, name='writer')(annotator)
    return Flow([writer], flow_type=BATCH)

if __name__ == "__main__":
    from videoflow.engines.local import LocalProcessEngine
    flow = build_flow()
    flow.run(LocalProcessEngine())
    flow.join()

Walking through the graph:

• reader reads the video file frame by frame (as (index, frame) tuples), and frame strips the index off. Note the reader fans out: both the detector branch and the final annotator consume frame.

• detector runs a model and emits bounding boxes; filter (a custom node) keeps only vehicle classes; tracker assigns a stable id to each box across frames.

• annotator is a join: it receives both the original frame and the tracker boxes, and draws the boxes onto the frame. writer encodes the annotated frames into a new video.

Two things to note for a real deployment:

• The whole flow uses BATCH mode. A video file must never run in REALTIME mode — the reader emits frames far faster than the detector can process them, and realtime’s drop policy would discard most of the video. See Batch versus realtime mode.

• The tracker is stateful (it correlates boxes across frames), so it subclasses OneTaskProcessorNode and always runs as a single worker. The detector, being stateless, can be scaled with nb_tasks to keep up with the reader.

To run this on Kubernetes instead, build an image with your detector/tracker dependencies (FROM videoflow-base) and deploy the same factory, pointing every node at it:

videoflow deploy examples/object_tracking.py:build_flow \
    --nats nats://nats.videoflow.svc:4222 --namespace videoflow \
    --image ghcr.io/acme/tracking:v1

See Deploying to Kubernetes.