Views
No views yet
| Property | Value |
|---|---|
| Category | Object Detection + Tracking + Zone Analytics (GstAnalytics) |
| 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) | car (2), motorcycle (3), bus (5), truck (7) |
gvaanalytics element defines the monitoring zone and
automatically attaches GstAnalyticsZoneMtd metadata to every tracked vehicle
whose center falls inside the polygon.
A Python probe reads this GstAnalytics metadata to accumulate per-vehicle dwell
time and raises a stopped-too-long event when the threshold is exceeded.yolo26n, yolo26s, yolo26m, yolo26l, yolo26x.
Smaller variants (yolo26n, yolo26s) are recommended for high-FPS edge
deployment; larger variants improve recall for distant vehicles.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 FP16yolo26n with any variant (yolo26s, yolo26m, yolo26l, yolo26x).
The second argument selects the precision (FP32, FP16, INT8); the default is FP16.openvino, ultralytics; adds nncf for INT8).ParkingVideo.mp4) from the Intel Edge
AI Resources project into the current directory.yolo26n_openvino_model/ -- FP32 or FP16 OpenVINO IR model directory.yolo26n_stopped_vehicle_int8.xml / yolo26n_stopped_vehicle_int8.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 |
Note: The INT8 calibration uses frames from the bundled sample video. For production accuracy, replace it with a representative set of frames from the target deployment site.
gvaanalytics element, which automatically detects when tracked vehicles are
inside the zone using GstAnalytics metadata -- no Python polygon math required.
A typical no-stop-zone configuration on the 1920x1080 sample video might be:1[
2 {
3 "id": "no_stop_zone",
4 "type": "polygon",
5 "points": [
6 {"x": 250, "y": 350},
7 {"x": 1700, "y": 350},
8 {"x": 1700, "y": 900},
9 {"x": 250, "y": 900}
10 ]
11 }
12]STOPPED_SECONDS = 5.0 # dwell threshold, in seconds (demo value)Note: The sample uses a 5-second threshold so that stopped-too-long events are triggered quickly on the short demo video. For production deployments, increase this to 30--300 seconds depending on the site's operational requirements.
gvaanalytics element attaches GstAnalyticsZoneMtd to each detection
whose center falls inside the polygon. The Python probe checks for this
metadata to accumulate per-vehicle dwell time.Note: The zone polygon supports arbitrary shapes (not just rectangles). Usedraw-zones=true(the default) so thatgvawatermarkrenders the zone boundary on the output video. Match the polygon coordinates to your video's resolution.
car, motorcycle, bus, truck), applies simple centroid tracking
with track IDs, and accumulates dwell time for every tracked vehicle whose
centroid falls inside the no-stop zone polygon. A STOPPED_TOO_LONG event is
logged -- with the wall-clock timestamp inside the video -- when a vehicle's
dwell time crosses the threshold. The saved output video draws the zone polygon
and the vehicle bounding boxes with their per-track dwell time.
Change the device string to run on CPU, GPU, or NPU.1import cv2
2import numpy as np
3import openvino as ov
4
5VEHICLE_CLASS_IDS = {2: "car", 3: "motorcycle", 5: "bus", 7: "truck"}
6CONF_THRESHOLD = 0.4
7INPUT_SIZE = 640
8MAX_DIST = 80 # max centroid movement (px) to link a track frame-to-frame
9MAX_MISSED = 15 # keep a track alive this many frames through detection gaps
10STOPPED_SECONDS = 5.0
11
12# No-stop zone polygon (pixel coordinates; match your video resolution).
13ZONE_POLY = np.array(
14 [[250, 350], [1700, 350], [1700, 900], [250, 900]], dtype=np.int32)
15
16core = ov.Core()
17model = core.read_model("yolo26n_openvino_model/yolo26n.xml")
18
19# Change device to "GPU" or "NPU" to run on integrated GPU or NPU.
20compiled = core.compile_model(model, "CPU")
21
22cap = cv2.VideoCapture("ParkingVideo.mp4")
23fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
24width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
25height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
26
27
28def fmt_time(seconds: float) -> str:
29 """Format elapsed video time as MM:SS.mmm."""
30 minutes, secs = divmod(seconds, 60)
31 return f"{int(minutes):02d}:{secs:06.3f}"
32
33
34writer = cv2.VideoWriter(
35 "output_openvino.mp4", cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height))
36
37# Each track: {"centroid": (x, y), "box": (x1, y1, x2, y2, label), "missed": int}
38tracks: dict[int, dict] = {}
39dwell: dict[int, float] = {}
40flagged: set[int] = set()
41next_id = 0
42frame_idx = 0
43
44while True:
45 ok, frame = cap.read()
46 if not ok:
47 break
48 frame_idx += 1
49 now = frame_idx / fps
50 h0, w0 = frame.shape[:2]
51 sx, sy = w0 / INPUT_SIZE, h0 / INPUT_SIZE
52
53 blob = cv2.resize(frame, (INPUT_SIZE, INPUT_SIZE))
54 blob = cv2.cvtColor(blob, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
55 blob = blob.transpose(2, 0, 1)[np.newaxis, ...]
56
57 output = compiled([blob])[compiled.output(0)][0]
58 mask = (output[:, 4] >= CONF_THRESHOLD) & np.isin(
59 output[:, 5].astype(int), list(VEHICLE_CLASS_IDS.keys()))
60 dets = output[mask]
61
62 boxes = []
63 centroids = []
64 for det in dets:
65 x1 = int(det[0] * sx)
66 y1 = int(det[1] * sy)
67 x2 = int(det[2] * sx)
68 y2 = int(det[3] * sy)
69 boxes.append((x1, y1, x2, y2, VEHICLE_CLASS_IDS[int(det[5])]))
70 centroids.append(((x1 + x2) // 2, (y1 + y2) // 2))
71
72 # Greedy nearest-centroid association. Unmatched tracks are kept alive for
73 # up to MAX_MISSED frames so brief detection gaps do not reset dwell time.
74 used = set()
75 for tid, track in tracks.items():
76 px, py = track["centroid"]
77 best_d, best_j = MAX_DIST, -1
78 for j, (cx, cy) in enumerate(centroids):
79 if j in used:
80 continue
81 d = abs(cx - px) + abs(cy - py)
82 if d < best_d:
83 best_d, best_j = d, j
84 if best_j >= 0:
85 used.add(best_j)
86 track["centroid"] = centroids[best_j]
87 track["box"] = boxes[best_j]
88 track["missed"] = 0
89 else:
90 track["missed"] += 1
91 for tid in [t for t, tr in tracks.items() if tr["missed"] > MAX_MISSED]:
92 del tracks[tid]
93 for j, centroid in enumerate(centroids):
94 if j not in used:
95 tracks[next_id] = {"centroid": centroid, "box": boxes[j], "missed": 0}
96 next_id += 1
97
98 # Accumulate dwell time for vehicles whose centroid is inside the zone.
99 for tid, track in tracks.items():
100 cx, cy = track["centroid"]
101 inside = cv2.pointPolygonTest(ZONE_POLY, (cx, cy), False) >= 0
102 if inside:
103 dwell[tid] = dwell.get(tid, 0.0) + 1.0 / fps
104 if dwell[tid] >= STOPPED_SECONDS and tid not in flagged:
105 flagged.add(tid)
106 print(
107 f"STOPPED_TOO_LONG track={tid:<3} "
108 f"dwell={dwell[tid]:.1f}s time={fmt_time(now)} "
109 f"pos=({cx},{cy})", flush=True)
110 else:
111 dwell[tid] = 0.0
112
113 cv2.polylines(frame, [ZONE_POLY], True, (0, 0, 255), 2)
114 for tid, track in tracks.items():
115 x1, y1, x2, y2, label = track["box"]
116 color = (0, 0, 255) if tid in flagged else (0, 255, 0)
117 cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
118 cv2.putText(frame, f"{label} {dwell.get(tid, 0.0):.1f}s",
119 (x1, max(y1 - 6, 0)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
120 writer.write(frame)
121
122cap.release()
123writer.release()
124print(f"Total stopped-too-long vehicles: {len(flagged)}", flush=True)"CPU" -- default, works on all Intel platforms."GPU" -- Intel integrated or discrete GPU."NPU" -- Intel NPU (validate with benchmark_app -d NPU).MM:SS.mmm), and the vehicle position:1STOPPED_TOO_LONG track=0 dwell=5.0s time=00:05.033 pos=(984,489)
2STOPPED_TOO_LONG track=4 dwell=5.0s time=00:22.867 pos=(660,442)
3Total stopped-too-long vehicles: 2
1source /opt/intel/openvino_2026/setupvars.sh
2source /opt/intel/dlstreamer/scripts/setup_dls_env.sh
3export PYTHONPATH=/opt/intel/dlstreamer/python:/opt/intel/dlstreamer/gstreamer/lib/python3/dist-packages:${PYTHONPATH:-}1from collections import defaultdict
2import json
3import sys
4import gi
5gi.require_version("Gst", "1.0")
6gi.require_version("GstAnalytics", "1.0")
7gi.require_version("DLStreamerMeta", "1.0")
8gi.require_version("DLStreamerWatermarkMeta", "1.0")
9from gi.repository import Gst, GLib, GstAnalytics, DLStreamerMeta, DLStreamerWatermarkMeta
10
11Gst.init([])
12
13# Register DLStreamerMeta types so GstAnalytics iteration can handle them
14_ov = sys.modules["gi.overrides.GstAnalytics"]
15_ov.__mtd_types__[DLStreamerMeta.ZoneMtd.get_mtd_type()] = DLStreamerMeta.relation_meta_get_zone_mtd
16_ov.__mtd_types__[DLStreamerMeta.TripwireMtd.get_mtd_type()] = DLStreamerMeta.relation_meta_get_tripwire_mtd
17
18MODEL = "yolo26n_openvino_model/yolo26n.xml"
19VIDEO = "ParkingVideo.mp4"
20VEHICLE_LABELS = {"car", "motorcycle", "bus", "truck"}
21ZONE_JSON = json.dumps([{
22 "id": "no_stop_zone",
23 "type": "polygon",
24 "points": [{"x": 250, "y": 350}, {"x": 1700, "y": 350},
25 {"x": 1700, "y": 900}, {"x": 250, "y": 900}]
26}])
27STOPPED_SECONDS = 5.0
28
29pipeline = Gst.parse_launch(
30 f"filesrc location={VIDEO} ! decodebin3 ! videoconvert ! "
31 f"gvadetect model={MODEL} device=GPU threshold=0.4 ! queue ! "
32 f"gvatrack tracking-type=short-term-imageless ! queue ! "
33 f"gvaanalytics name=analytics draw-zones=true ! "
34 f"gvafpscounter ! identity name=probe ! gvawatermark name=watermark ! "
35 f"videoconvert ! video/x-raw,format=I420 ! "
36 f"openh264enc ! h264parse ! mp4mux ! filesink location=output_dlstreamer.mp4"
37)
38
39pipeline.get_by_name("analytics").set_property("zones", ZONE_JSON)
40
41dwell = defaultdict(float)
42last_seen = {}
43flagged = set()
44
45def on_buffer(pad, info):
46 buf = info.get_buffer()
47 now = buf.pts / Gst.SECOND if buf.pts != Gst.CLOCK_TIME_NONE else 0.0
48 rmeta = GstAnalytics.buffer_get_analytics_relation_meta(buf)
49 if not rmeta:
50 return Gst.PadProbeReturn.OK
51
52 # Iterate only over object-detection entries
53 for od in rmeta.iter_on_type(GstAnalytics.ODMtd):
54 label = GLib.quark_to_string(od.get_obj_type())
55 if label not in VEHICLE_LABELS:
56 continue
57
58 # Find tracking ID via direct relation
59 track_id = None
60 for trk in od.iter_direct_related(GstAnalytics.RelTypes.RELATE_TO, GstAnalytics.TrackingMtd):
61 success, tracking_id, *_ = trk.get_info()
62 if success:
63 track_id = tracking_id
64 break
65 if track_id is None:
66 continue
67
68 # Check if gvaanalytics placed this detection inside the zone
69 in_zone = False
70 for zone in od.iter_direct_related(GstAnalytics.RelTypes.RELATE_TO, DLStreamerMeta.ZoneMtd):
71 in_zone = True
72 break
73
74 if not in_zone:
75 # Vehicle left the zone; reset its dwell accumulator.
76 dwell[track_id] = 0.0
77 last_seen.pop(track_id, None)
78 continue
79
80 # Accumulate dwell time for vehicles inside the zone
81 dwell[track_id] += now - last_seen.get(track_id, now)
82 last_seen[track_id] = now
83
84 if dwell[track_id] >= STOPPED_SECONDS and track_id not in flagged:
85 flagged.add(track_id)
86 _, x, y, w, h, _ = od.get_location()
87 print(f"STOPPED_TOO_LONG id={track_id} {label} "
88 f"dwell={dwell[track_id]:.1f}s pos=({int(x + w/2)},{int(y + h)})")
89
90 return Gst.PadProbeReturn.OK
91
92pipeline.get_by_name("probe").get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, on_buffer)
93pipeline.set_state(Gst.State.PLAYING)
94pipeline.get_bus().timed_pop_filtered(Gst.CLOCK_TIME_NONE, Gst.MessageType.EOS | Gst.MessageType.ERROR)
95pipeline.set_state(Gst.State.NULL)1STOPPED_TOO_LONG id=1 car dwell=5.0s pos=(988,672)
2STOPPED_TOO_LONG id=2 car dwell=5.0s pos=(665,583)
3...output_dlstreamer.mp4.
The gvaanalytics element also draws the zone polygon on each frame via gvawatermark.
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.