Object tracker on video¶
This example shows how to run MobileNetv2SSD on video input frame, and perform object tracking on persons.
Similar samples:
Demo¶
Setup¶
Please run the install script to download all required dependencies. Please note that this script must be ran from git context, so you have to download the depthai-python repository first and then run the script
git clone https://github.com/luxonis/depthai-python.git
cd depthai-python/examples
python3 install_requirements.py
For additional information, please follow installation guide
This example script requires external file(s) to run. If you are using:
depthai-python, run
python3 examples/install_requirements.py
to download required file(s)dephtai-core, required file(s) will get downloaded automatically when building the example
Source code¶
Also available on GitHub
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | #!/usr/bin/env python3 from pathlib import Path import cv2 import depthai as dai import numpy as np import time import argparse labelMap = ["person", ""] nnPathDefault = str((Path(__file__).parent / Path('../models/person-detection-retail-0013_openvino_2021.4_7shave.blob')).resolve().absolute()) videoPathDefault = str((Path(__file__).parent / Path('../models/construction_vest.mp4')).resolve().absolute()) parser = argparse.ArgumentParser() parser.add_argument('-nnPath', help="Path to mobilenet detection network blob", default=nnPathDefault) parser.add_argument('-v', '--videoPath', help="Path to video frame", default=videoPathDefault) args = parser.parse_args() # Create pipeline pipeline = dai.Pipeline() # Define sources and outputs manip = pipeline.create(dai.node.ImageManip) objectTracker = pipeline.create(dai.node.ObjectTracker) detectionNetwork = pipeline.create(dai.node.MobileNetDetectionNetwork) manipOut = pipeline.create(dai.node.XLinkOut) xinFrame = pipeline.create(dai.node.XLinkIn) trackerOut = pipeline.create(dai.node.XLinkOut) xlinkOut = pipeline.create(dai.node.XLinkOut) nnOut = pipeline.create(dai.node.XLinkOut) manipOut.setStreamName("manip") xinFrame.setStreamName("inFrame") xlinkOut.setStreamName("trackerFrame") trackerOut.setStreamName("tracklets") nnOut.setStreamName("nn") # Properties xinFrame.setMaxDataSize(1920*1080*3) manip.initialConfig.setResizeThumbnail(544, 320) # manip.initialConfig.setResize(384, 384) # manip.initialConfig.setKeepAspectRatio(False) #squash the image to not lose FOV # The NN model expects BGR input. By default ImageManip output type would be same as input (gray in this case) manip.initialConfig.setFrameType(dai.ImgFrame.Type.BGR888p) manip.inputImage.setBlocking(True) # setting node configs detectionNetwork.setBlobPath(args.nnPath) detectionNetwork.setConfidenceThreshold(0.5) detectionNetwork.input.setBlocking(True) objectTracker.inputTrackerFrame.setBlocking(True) objectTracker.inputDetectionFrame.setBlocking(True) objectTracker.inputDetections.setBlocking(True) objectTracker.setDetectionLabelsToTrack([1]) # track only person # possible tracking types: ZERO_TERM_COLOR_HISTOGRAM, ZERO_TERM_IMAGELESS, SHORT_TERM_IMAGELESS, SHORT_TERM_KCF objectTracker.setTrackerType(dai.TrackerType.ZERO_TERM_COLOR_HISTOGRAM) # take the smallest ID when new object is tracked, possible options: SMALLEST_ID, UNIQUE_ID objectTracker.setTrackerIdAssignmentPolicy(dai.TrackerIdAssignmentPolicy.SMALLEST_ID) # Linking manip.out.link(manipOut.input) manip.out.link(detectionNetwork.input) xinFrame.out.link(manip.inputImage) xinFrame.out.link(objectTracker.inputTrackerFrame) detectionNetwork.out.link(nnOut.input) detectionNetwork.out.link(objectTracker.inputDetections) detectionNetwork.passthrough.link(objectTracker.inputDetectionFrame) objectTracker.out.link(trackerOut.input) objectTracker.passthroughTrackerFrame.link(xlinkOut.input) # Connect and start the pipeline with dai.Device(pipeline) as device: qIn = device.getInputQueue(name="inFrame") trackerFrameQ = device.getOutputQueue(name="trackerFrame", maxSize=4) tracklets = device.getOutputQueue(name="tracklets", maxSize=4) qManip = device.getOutputQueue(name="manip", maxSize=4) qDet = device.getOutputQueue(name="nn", maxSize=4) startTime = time.monotonic() counter = 0 fps = 0 detections = [] frame = None def to_planar(arr: np.ndarray, shape: tuple) -> np.ndarray: return cv2.resize(arr, shape).transpose(2, 0, 1).flatten() # nn data, being the bounding box locations, are in <0..1> range - they need to be normalized with frame width/height def frameNorm(frame, bbox): normVals = np.full(len(bbox), frame.shape[0]) normVals[::2] = frame.shape[1] return (np.clip(np.array(bbox), 0, 1) * normVals).astype(int) def displayFrame(name, frame): for detection in detections: bbox = frameNorm(frame, (detection.xmin, detection.ymin, detection.xmax, detection.ymax)) cv2.rectangle(frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), (255, 0, 0), 2) cv2.putText(frame, labelMap[detection.label], (bbox[0] + 10, bbox[1] + 20), cv2.FONT_HERSHEY_TRIPLEX, 0.5, 255) cv2.putText(frame, f"{int(detection.confidence * 100)}%", (bbox[0] + 10, bbox[1] + 40), cv2.FONT_HERSHEY_TRIPLEX, 0.5, 255) cv2.imshow(name, frame) cap = cv2.VideoCapture(args.videoPath) baseTs = time.monotonic() simulatedFps = 30 inputFrameShape = (1920, 1080) while cap.isOpened(): read_correctly, frame = cap.read() if not read_correctly: break img = dai.ImgFrame() img.setType(dai.ImgFrame.Type.BGR888p) img.setData(to_planar(frame, inputFrameShape)) img.setTimestamp(baseTs) baseTs += 1/simulatedFps img.setWidth(inputFrameShape[0]) img.setHeight(inputFrameShape[1]) qIn.send(img) trackFrame = trackerFrameQ.tryGet() if trackFrame is None: continue track = tracklets.get() manip = qManip.get() inDet = qDet.get() counter+=1 current_time = time.monotonic() if (current_time - startTime) > 1 : fps = counter / (current_time - startTime) counter = 0 startTime = current_time detections = inDet.detections manipFrame = manip.getCvFrame() displayFrame("nn", manipFrame) color = (255, 0, 0) trackerFrame = trackFrame.getCvFrame() trackletsData = track.tracklets for t in trackletsData: roi = t.roi.denormalize(trackerFrame.shape[1], trackerFrame.shape[0]) x1 = int(roi.topLeft().x) y1 = int(roi.topLeft().y) x2 = int(roi.bottomRight().x) y2 = int(roi.bottomRight().y) try: label = labelMap[t.label] except: label = t.label cv2.putText(trackerFrame, str(label), (x1 + 10, y1 + 20), cv2.FONT_HERSHEY_TRIPLEX, 0.5, 255) cv2.putText(trackerFrame, f"ID: {[t.id]}", (x1 + 10, y1 + 35), cv2.FONT_HERSHEY_TRIPLEX, 0.5, 255) cv2.putText(trackerFrame, t.status.name, (x1 + 10, y1 + 50), cv2.FONT_HERSHEY_TRIPLEX, 0.5, 255) cv2.rectangle(trackerFrame, (x1, y1), (x2, y2), color, cv2.FONT_HERSHEY_SIMPLEX) cv2.putText(trackerFrame, "Fps: {:.2f}".format(fps), (2, trackerFrame.shape[0] - 4), cv2.FONT_HERSHEY_TRIPLEX, 0.4, color) cv2.imshow("tracker", trackerFrame) if cv2.waitKey(1) == ord('q'): break |
Also available on GitHub
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | #include <chrono> #include <iostream> #include "utility.hpp" // Includes common necessary includes for development using depthai library #include "depthai/depthai.hpp" static const std::vector<std::string> labelMap = {"", "person"}; static std::atomic<bool> fullFrameTracking{false}; int main(int argc, char** argv) { using namespace std; using namespace std::chrono; std::string nnPath(BLOB_PATH); std::string videoPath(VIDEO_PATH); // If path to blob specified, use that if(argc > 2) { nnPath = std::string(argv[1]); videoPath = std::string(argv[2]); } // Print which blob we are using printf("Using blob at path: %s\n", nnPath.c_str()); printf("Using video at path: %s\n", videoPath.c_str()); // Create pipeline dai::Pipeline pipeline; // Define sources and outputs auto manip = pipeline.create<dai::node::ImageManip>(); auto objectTracker = pipeline.create<dai::node::ObjectTracker>(); auto detectionNetwork = pipeline.create<dai::node::MobileNetDetectionNetwork>(); auto manipOut = pipeline.create<dai::node::XLinkOut>(); auto xinFrame = pipeline.create<dai::node::XLinkIn>(); auto trackerOut = pipeline.create<dai::node::XLinkOut>(); auto xlinkOut = pipeline.create<dai::node::XLinkOut>(); auto nnOut = pipeline.create<dai::node::XLinkOut>(); manipOut->setStreamName("manip"); xinFrame->setStreamName("inFrame"); xlinkOut->setStreamName("trackerFrame"); trackerOut->setStreamName("tracklets"); nnOut->setStreamName("nn"); // Properties xinFrame->setMaxDataSize(1920 * 1080 * 3); manip->initialConfig.setResizeThumbnail(544, 320); // manip->initialConfig.setResize(384, 384); // manip->initialConfig.setKeepAspectRatio(false); //squash the image to not lose FOV // The NN model expects BGR input. By default ImageManip output type would be same as input (gray in this case) manip->initialConfig.setFrameType(dai::ImgFrame::Type::BGR888p); manip->inputImage.setBlocking(true); // setting node configs detectionNetwork->setBlobPath(nnPath); detectionNetwork->setConfidenceThreshold(0.5); detectionNetwork->input.setBlocking(true); objectTracker->inputTrackerFrame.setBlocking(true); objectTracker->inputDetectionFrame.setBlocking(true); objectTracker->inputDetections.setBlocking(true); objectTracker->setDetectionLabelsToTrack({1}); // track only person // possible tracking types: ZERO_TERM_COLOR_HISTOGRAM, ZERO_TERM_IMAGELESS, SHORT_TERM_IMAGELESS, SHORT_TERM_KCF objectTracker->setTrackerType(dai::TrackerType::ZERO_TERM_COLOR_HISTOGRAM); // take the smallest ID when new object is tracked, possible options: SMALLEST_ID, UNIQUE_ID objectTracker->setTrackerIdAssignmentPolicy(dai::TrackerIdAssignmentPolicy::SMALLEST_ID); // Linking manip->out.link(manipOut->input); manip->out.link(detectionNetwork->input); xinFrame->out.link(manip->inputImage); xinFrame->out.link(objectTracker->inputTrackerFrame); detectionNetwork->out.link(nnOut->input); detectionNetwork->out.link(objectTracker->inputDetections); detectionNetwork->passthrough.link(objectTracker->inputDetectionFrame); objectTracker->out.link(trackerOut->input); objectTracker->passthroughTrackerFrame.link(xlinkOut->input); // Connect to device and start pipeline dai::Device device(pipeline); auto qIn = device.getInputQueue("inFrame", 4); auto trackerFrameQ = device.getOutputQueue("trackerFrame", 4); auto tracklets = device.getOutputQueue("tracklets", 4); auto qManip = device.getOutputQueue("manip", 4); auto qDet = device.getOutputQueue("nn", 4); auto startTime = steady_clock::now(); int counter = 0; float fps = 0; cv::Mat frame; cv::Mat manipFrame; std::vector<dai::ImgDetection> detections; // Add bounding boxes and text to the frame and show it to the user auto displayFrame = [](std::string name, cv::Mat frame, std::vector<dai::ImgDetection>& detections) { auto color = cv::Scalar(255, 0, 0); // nn data, being the bounding box locations, are in <0..1> range - they need to be normalized with frame width/height for(auto& detection : detections) { int x1 = detection.xmin * frame.cols; int y1 = detection.ymin * frame.rows; int x2 = detection.xmax * frame.cols; int y2 = detection.ymax * frame.rows; uint32_t labelIndex = detection.label; std::string labelStr = to_string(labelIndex); if(labelIndex < labelMap.size()) { labelStr = labelMap[labelIndex]; } cv::putText(frame, labelStr, cv::Point(x1 + 10, y1 + 20), cv::FONT_HERSHEY_TRIPLEX, 0.5, color); std::stringstream confStr; confStr << std::fixed << std::setprecision(2) << detection.confidence * 100; cv::putText(frame, confStr.str(), cv::Point(x1 + 10, y1 + 40), cv::FONT_HERSHEY_TRIPLEX, 0.5, color); cv::rectangle(frame, cv::Rect(cv::Point(x1, y1), cv::Point(x2, y2)), color, cv::FONT_HERSHEY_SIMPLEX); } // Show the frame cv::imshow(name, frame); }; cv::VideoCapture cap(videoPath); auto baseTs = steady_clock::now(); float simulatedFps = 30; while(cap.isOpened()) { // Read frame from video cap >> frame; if(frame.empty()) break; auto img = std::make_shared<dai::ImgFrame>(); frame = resizeKeepAspectRatio(frame, cv::Size(1920, 1080), cv::Scalar(0)); toPlanar(frame, img->getData()); img->setTimestamp(baseTs); baseTs += steady_clock::duration(static_cast<int64_t>((1000 * 1000 * 1000 / simulatedFps))); img->setWidth(1920); img->setHeight(1080); img->setType(dai::ImgFrame::Type::BGR888p); qIn->send(img); auto trackFrame = trackerFrameQ->tryGet<dai::ImgFrame>(); if(!trackFrame) { continue; } auto track = tracklets->get<dai::Tracklets>(); auto inManip = qManip->get<dai::ImgFrame>(); auto inDet = qDet->get<dai::ImgDetections>(); counter++; auto currentTime = steady_clock::now(); auto elapsed = duration_cast<duration<float>>(currentTime - startTime); if(elapsed > seconds(1)) { fps = counter / elapsed.count(); counter = 0; startTime = currentTime; } detections = inDet->detections; manipFrame = inManip->getCvFrame(); displayFrame("nn", manipFrame, detections); auto color = cv::Scalar(255, 0, 0); cv::Mat trackerFrame = trackFrame->getCvFrame(); auto trackletsData = track->tracklets; for(auto& t : trackletsData) { auto roi = t.roi.denormalize(trackerFrame.cols, trackerFrame.rows); int x1 = roi.topLeft().x; int y1 = roi.topLeft().y; int x2 = roi.bottomRight().x; int y2 = roi.bottomRight().y; uint32_t labelIndex = t.label; std::string labelStr = to_string(labelIndex); if(labelIndex < labelMap.size()) { labelStr = labelMap[labelIndex]; } cv::putText(trackerFrame, labelStr, cv::Point(x1 + 10, y1 + 20), cv::FONT_HERSHEY_TRIPLEX, 0.5, color); std::stringstream idStr; idStr << "ID: " << t.id; cv::putText(trackerFrame, idStr.str(), cv::Point(x1 + 10, y1 + 40), cv::FONT_HERSHEY_TRIPLEX, 0.5, color); std::stringstream statusStr; statusStr << "Status: " << t.status; cv::putText(trackerFrame, statusStr.str(), cv::Point(x1 + 10, y1 + 60), cv::FONT_HERSHEY_TRIPLEX, 0.5, color); cv::rectangle(trackerFrame, cv::Rect(cv::Point(x1, y1), cv::Point(x2, y2)), color, cv::FONT_HERSHEY_SIMPLEX); } std::stringstream fpsStr; fpsStr << "NN fps:" << std::fixed << std::setprecision(2) << fps; cv::putText(trackerFrame, fpsStr.str(), cv::Point(2, trackFrame->getHeight() - 4), cv::FONT_HERSHEY_TRIPLEX, 0.4, color); cv::imshow("tracker", trackerFrame); int key = cv::waitKey(1); if(key == 'q' || key == 'Q') { return 0; } } return 0; } |
Got questions?
We’re always happy to help with code or other questions you might have.