| Property | Value |
|---|---|
| Category | Object Detection + Speed Heatmap Aggregation |
| Base Model | YOLO26 (Ultralytics) |
| Source Framework | PyTorch (Ultralytics) |
| Supported Precisions | FP32, FP16, INT8 (mixed-precision) |
| Inference Engine | OpenVINO |
| Hardware | CPU, GPU, NPU |
| Detected Class(es) | All 80 COCO classes (heatmap colored by object speed) |
yolo26n, yolo26s, yolo26m, yolo26l, yolo26x.
Smaller variants (yolo26n, yolo26s) are recommended for high-FPS edge
deployment; larger variants improve recall for small or distant objects.1python3 -m venv .venv --system-site-packages
2source .venv/bin/activateNote: The--system-site-packagesflag is required so the virtual environment can access the system-installed OpenVINO and DLStreamer Python packages.
1chmod +x export_and_quantize.sh
2./export_and_quantize.sh1./export_and_quantize.sh yolo26n FP32 # full-precision
2./export_and_quantize.sh yolo26n INT8 # quantized
3./export_and_quantize.sh yolo26s # larger variant, default FP16openvino, ultralytics; adds nncf for INT8).test.jpg) and a sample test video (test_video.mp4).yolo26n_openvino_model/ -- FP32 or FP16 OpenVINO IR model directory.yolo26n_heatmap_int8.xml / .bin -- INT8 quantized model (only when INT8 is selected).| Precision | CPU | GPU | NPU |
|---|---|---|---|
| FP32 | Yes | Yes | No |
| FP16 | Yes | Yes | Yes |
| INT8 | Yes | Yes | Yes |
output_openvino.mp4.
It also saves the final speed heatmap as heatmap.jpg.
Change the device string to run on CPU, GPU, or NPU.1import cv2
2import numpy as np
3import openvino as ov
4
5CONF_THRESHOLD = 0.4
6INPUT_SIZE = 640
7HEATMAP_ALPHA = 0.55
8MATCH_DIST = 80.0 # max px between frames to treat detections as the same object
9MAX_SPEED = 20.0 # px/frame that maps to full red
10
11
12def render_speed_heatmap(frame, speed_sum, count, alpha):
13 """Color traffic by average speed: blue = slow, yellow = medium,
14 red = fast. Only regions where vehicles were seen are tinted, so
15 empty background keeps its original color."""
16 avg = np.zeros_like(speed_sum)
17 seen = count > 0
18 avg[seen] = speed_sum[seen] / count[seen]
19 avg = cv2.GaussianBlur(avg, (0, 0), sigmaX=15)
20 presence = cv2.GaussianBlur(seen.astype(np.float32), (0, 0), sigmaX=15)
21 norm = np.clip(avg / MAX_SPEED, 0, 1) # 0 = slow (blue), 1 = fast (red)
22 color = cv2.applyColorMap((norm * 255).astype(np.uint8), cv2.COLORMAP_JET)
23 weight = (np.clip(presence, 0, 1) * alpha)[..., np.newaxis]
24 overlay = frame.astype(np.float32) * (1 - weight) + color.astype(np.float32) * weight
25 return overlay.astype(np.uint8), color
26
27
28core = ov.Core()
29model = core.read_model("yolo26n_openvino_model/yolo26n.xml")
30
31# Change device to "GPU" or "NPU" to run on integrated GPU or NPU.
32compiled = core.compile_model(model, "CPU")
33
34cap = cv2.VideoCapture("test_video.mp4")
35fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
36width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
37height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
38writer = cv2.VideoWriter(
39 "output_openvino.mp4", cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height))
40
41speed_sum = np.zeros((height, width), dtype=np.float32)
42count = np.zeros((height, width), dtype=np.float32)
43prev_centroids = []
44heatmap_color = None
45frame_idx = 0
46total_dets = 0
47
48while True:
49 ok, frame = cap.read()
50 if not ok:
51 break
52 frame_idx += 1
53 h0, w0 = frame.shape[:2]
54 sx, sy = w0 / INPUT_SIZE, h0 / INPUT_SIZE
55
56 blob = cv2.resize(frame, (INPUT_SIZE, INPUT_SIZE))
57 blob = cv2.cvtColor(blob, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
58 blob = blob.transpose(2, 0, 1)[np.newaxis, ...]
59
60 output = compiled([blob])[compiled.output(0)][0]
61 dets = output[output[:, 4] >= CONF_THRESHOLD]
62 total_dets += len(dets)
63
64 cur_centroids = []
65 for det in dets:
66 x1, y1 = int(det[0] * sx), int(det[1] * sy)
67 x2, y2 = int(det[2] * sx), int(det[3] * sy)
68 x1, x2 = max(0, x1), min(width, x2)
69 y1, y2 = max(0, y1), min(height, y2)
70 cx, cy = (x1 + x2) / 2, (y1 + y2) / 2
71 cur_centroids.append((cx, cy))
72
73 # Speed = displacement from the nearest detection in the previous frame.
74 speed = 0.0
75 if prev_centroids:
76 d = min(np.hypot(cx - px, cy - py) for px, py in prev_centroids)
77 if d <= MATCH_DIST:
78 speed = d
79 speed_sum[y1:y2, x1:x2] += speed
80 count[y1:y2, x1:x2] += 1.0
81 prev_centroids = cur_centroids
82
83 overlay, heatmap_color = render_speed_heatmap(frame, speed_sum, count, HEATMAP_ALPHA)
84 cv2.putText(overlay, f"Detections: {len(dets)}", (10, 30),
85 cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2)
86 writer.write(overlay)
87
88cap.release()
89writer.release()
90
91if heatmap_color is not None:
92 cv2.imwrite("heatmap.jpg", heatmap_color)
93 print("Saved: heatmap.jpg")
94
95print(f"Processed {frame_idx} frames, {total_dets} total detections", flush=True)"CPU" -- default, works on all Intel platforms."GPU" -- Intel integrated or discrete GPU."NPU" -- Intel NPU (validate with benchmark_app -d NPU).
gvadetect.
A buffer probe estimates each object's speed from its per-frame displacement
and overlays a speed-colored heatmap (red = fast, blue = slow) on each frame
before encoding to output_dlstreamer.mp4.Notes on running this sample:
Use the FP16 IR (yolo26n_openvino_model/yolo26n.xml). Class names are read automatically from the model's embeddedmetadata.yamlby DLStreamer 2026.0+ -- no externallabels-fileis required. ExportPYTHONPATHso the DLStreamer Python module is importable:bash1source /opt/intel/openvino_2026/setupvars.sh 2source /opt/intel/dlstreamer/scripts/setup_dls_env.sh 3export PYTHONPATH=/opt/intel/dlstreamer/python:\ 4/opt/intel/dlstreamer/gstreamer/lib/python3/dist-packages:${PYTHONPATH:-}
1import gi
2
3gi.require_version("Gst", "1.0")
4gi.require_version("GstAnalytics", "1.0")
5from gi.repository import Gst, GLib, GstAnalytics
6
7import numpy as np
8
9Gst.init([])
10
11# Import cv2 after Gst.init to avoid GStreamer re-initialization conflicts.
12import cv2
13
14INPUT_VIDEO = "test_video.mp4"
15HEATMAP_ALPHA = 0.55
16MATCH_DIST = 80.0 # max px between frames to treat detections as the same object
17MAX_SPEED = 20.0 # px/frame that maps to full red
18
19
20def render_speed_heatmap(frame, speed_sum, count, alpha):
21 """Color traffic by average speed: blue = slow, yellow = medium,
22 red = fast. Only regions where vehicles were seen are tinted, so
23 empty background keeps its original color."""
24 avg = np.zeros_like(speed_sum)
25 seen = count > 0
26 avg[seen] = speed_sum[seen] / count[seen]
27 avg = cv2.GaussianBlur(avg, (0, 0), sigmaX=15)
28 presence = cv2.GaussianBlur(seen.astype(np.float32), (0, 0), sigmaX=15)
29 norm = np.clip(avg / MAX_SPEED, 0, 1) # 0 = slow (blue), 1 = fast (red)
30 color = cv2.applyColorMap((norm * 255).astype(np.uint8), cv2.COLORMAP_JET)
31 weight = (np.clip(presence, 0, 1) * alpha)[..., np.newaxis]
32 overlay = frame.astype(np.float32) * (1 - weight) + color.astype(np.float32) * weight
33 return overlay.astype(np.uint8)
34
35
36# For CPU: change device=GPU to device=CPU.
37# For NPU: change device=GPU to device=NPU (batch-size=1, nireq=4 recommended).
38pipeline_str = (
39 f"filesrc location={INPUT_VIDEO} ! decodebin3 ! "
40 "videoconvert ! video/x-raw,format=BGR ! "
41 "gvadetect model=yolo26n_openvino_model/yolo26n.xml "
42 "device=GPU "
43 "threshold=0.4 ! queue ! "
44 "appsink name=sink emit-signals=false sync=false"
45)
46pipeline = Gst.parse_launch(pipeline_str)
47sink = pipeline.get_by_name("sink")
48pipeline.set_state(Gst.State.PLAYING)
49
50speed_sum = None
51count = None
52prev_centroids = []
53writer = None
54frame_idx = 0
55total_dets = 0
56
57while True:
58 sample = sink.emit("pull-sample")
59 if sample is None:
60 break
61 buf = sample.get_buffer()
62 caps = sample.get_caps().get_structure(0)
63 width = caps.get_value("width")
64 height = caps.get_value("height")
65
66 if speed_sum is None:
67 speed_sum = np.zeros((height, width), dtype=np.float32)
68 count = np.zeros((height, width), dtype=np.float32)
69
70 ok, mapinfo = buf.map(Gst.MapFlags.READ)
71 if not ok:
72 continue
73 frame = np.ndarray((height, width, 3), dtype=np.uint8,
74 buffer=mapinfo.data).copy()
75 buf.unmap(mapinfo)
76 frame_idx += 1
77
78 rmeta = GstAnalytics.buffer_get_analytics_relation_meta(buf)
79 cur_centroids = []
80 det_count = 0
81 if rmeta is not None:
82 idx = 1
83 while True:
84 ok_od, od = rmeta.get_od_mtd(idx)
85 if not ok_od:
86 break
87 _, x, y, w, h, _ = od.get_location()
88 x1, y1 = max(0, int(x)), max(0, int(y))
89 x2, y2 = min(width, int(x + w)), min(height, int(y + h))
90 cx, cy = (x1 + x2) / 2, (y1 + y2) / 2
91 cur_centroids.append((cx, cy))
92
93 # Speed = displacement from the nearest detection last frame.
94 speed = 0.0
95 if prev_centroids:
96 d = min(np.hypot(cx - px, cy - py) for px, py in prev_centroids)
97 if d <= MATCH_DIST:
98 speed = d
99 speed_sum[y1:y2, x1:x2] += speed
100 count[y1:y2, x1:x2] += 1.0
101 det_count += 1
102 idx += 1
103 prev_centroids = cur_centroids
104 total_dets += det_count
105
106 overlay = render_speed_heatmap(frame, speed_sum, count, HEATMAP_ALPHA)
107
108 if writer is None:
109 writer = cv2.VideoWriter(
110 "output_dlstreamer.mp4", cv2.VideoWriter_fourcc(*"mp4v"),
111 30.0, (width, height))
112 writer.write(overlay)
113 print(f"Frame {frame_idx}: detections={det_count}", flush=True)
114
115pipeline.set_state(Gst.State.NULL)
116if writer:
117 writer.release()
118print(f"Processed {frame_idx} frames, {total_dets} total detections", flush=True)device=GPU -- default in the sample code.device=CPU -- change device=GPU to device=CPU.device=NPU -- change device=GPU to device=NPU; use batch-size=1 and nireq=4 for best NPU utilization.