vehicle-stopped-too-long-detection

Vehicle Stopped-Too-Long Detection

---

Overview

Vehicle Stopped-Too-Long Detection is a Metro Analytics use case that flags vehicles that remain inside a configurable no-stop or safety zone for longer than an allowed dwell-time threshold. It is built on YOLO26 for vehicle detection, paired with a multi-object tracker that assigns persistent IDs across frames. DLStreamer's 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.

Typical Metro deployments include:

• No-Stopping Zone Enforcement -- alert on vehicles parked in clearways, bus lanes, or red routes.
• Fire Lane and Emergency Access -- keep fire lanes and ambulance bays clear of parked vehicles.
• Loading Bay Overstay -- detect delivery vehicles that exceed the permitted dwell time.
• Station Drop-Off Management -- flag vehicles idling too long at kiss-and-ride zones or taxi ranks.

Available variants: yolo26n, yolo26s, yolo26m, yolo26l, yolo26x. Smaller variants (yolo26n, yolo26s) are recommended for high-FPS edge deployment; larger variants improve recall for distant vehicles.

---

Prerequisites

• Python 3.11+
• Install OpenVINO (latest version)
• Install Intel DLStreamer (latest version)

Create and activate a Python virtual environment before running the scripts:

python3 -m venv .venv --system-site-packages
source .venv/bin/activate

Note: The --system-site-packages flag is required so the virtual environment can access the system-installed OpenVINO and DLStreamer Python packages.

---

Getting Started

Download and Quantize Model

Run the provided script to download, export to OpenVINO IR, and optionally quantize:

chmod +x export_and_quantize.sh
./export_and_quantize.sh

This exports the default yolo26n model in FP16 precision.

Optional: Select a Different Variant or Precision

./export_and_quantize.sh yolo26n FP32   # full-precision
./export_and_quantize.sh yolo26n INT8   # quantized
./export_and_quantize.sh yolo26s        # larger variant, default FP16

Replace yolo26n with any variant (yolo26s, yolo26m, yolo26l, yolo26x). The second argument selects the precision (FP32, FP16, INT8); the default is FP16.

The script performs the following steps:

1. Installs dependencies (openvino, ultralytics; adds nncf for INT8).
2. Downloads the sample parking video (ParkingVideo.mp4) from the Intel Edge AI Resources project into the current directory.
3. Downloads the PyTorch weights and exports to OpenVINO IR.
4. (INT8 only) Quantizes the model using NNCF post-training quantization.

Output files:

• 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 / Device Compatibility

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.

Defining the No-Stop Zone

The zone is a polygon defined in JSON and passed to DLStreamer's 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:

[
  {
    "id": "no_stop_zone",
    "type": "polygon",
    "points": [
      {"x": 250, "y": 350},
      {"x": 1700, "y": 350},
      {"x": 1700, "y": 900},
      {"x": 250, "y": 900}
    ]
  }
]

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.

The 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). Use draw-zones=true (the default) so that gvawatermark renders the zone boundary on the output video. Match the polygon coordinates to your video's resolution.

OpenVINO Sample

The sample below runs YOLO26 inference on the parking video, keeps only vehicle classes (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.

import cv2
import numpy as np
import openvino as ov

VEHICLE_CLASS_IDS = {2: "car", 3: "motorcycle", 5: "bus", 7: "truck"}
CONF_THRESHOLD = 0.4
INPUT_SIZE = 640
MAX_DIST = 80          # max centroid movement (px) to link a track frame-to-frame
MAX_MISSED = 15        # keep a track alive this many frames through detection gaps
STOPPED_SECONDS = 5.0

# No-stop zone polygon (pixel coordinates; match your video resolution).
ZONE_POLY = np.array(
    [[250, 350], [1700, 350], [1700, 900], [250, 900]], dtype=np.int32)

core = ov.Core()
model = core.read_model("yolo26n_openvino_model/yolo26n.xml")

# Change device to "GPU" or "NPU" to run on integrated GPU or NPU.
compiled = core.compile_model(model, "CPU")

cap = cv2.VideoCapture("ParkingVideo.mp4")
fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))


def fmt_time(seconds: float) -> str:
    """Format elapsed video time as MM:SS.mmm."""
    minutes, secs = divmod(seconds, 60)
    return f"{int(minutes):02d}:{secs:06.3f}"


writer = cv2.VideoWriter(
    "output_openvino.mp4", cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height))

# Each track: {"centroid": (x, y), "box": (x1, y1, x2, y2, label), "missed": int}
tracks: dict[int, dict] = {}
dwell: dict[int, float] = {}
flagged: set[int] = set()
next_id = 0
frame_idx = 0

while True:
    ok, frame = cap.read()
    if not ok:
        break
    frame_idx += 1
    now = frame_idx / fps
    h0, w0 = frame.shape[:2]
    sx, sy = w0 / INPUT_SIZE, h0 / INPUT_SIZE

    blob = cv2.resize(frame, (INPUT_SIZE, INPUT_SIZE))
    blob = cv2.cvtColor(blob, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
    blob = blob.transpose(2, 0, 1)[np.newaxis, ...]

    output = compiled([blob])[compiled.output(0)][0]
    mask = (output[:, 4] >= CONF_THRESHOLD) & np.isin(
        output[:, 5].astype(int), list(VEHICLE_CLASS_IDS.keys()))
    dets = output[mask]

    boxes = []
    centroids = []
    for det in dets:
        x1 = int(det[0] * sx)
        y1 = int(det[1] * sy)
        x2 = int(det[2] * sx)
        y2 = int(det[3] * sy)
        boxes.append((x1, y1, x2, y2, VEHICLE_CLASS_IDS[int(det[5])]))
        centroids.append(((x1 + x2) // 2, (y1 + y2) // 2))

    # Greedy nearest-centroid association. Unmatched tracks are kept alive for
    # up to MAX_MISSED frames so brief detection gaps do not reset dwell time.
    used = set()
    for tid, track in tracks.items():
        px, py = track["centroid"]
        best_d, best_j = MAX_DIST, -1
        for j, (cx, cy) in enumerate(centroids):
            if j in used:
                continue
            d = abs(cx - px) + abs(cy - py)
            if d < best_d:
                best_d, best_j = d, j
        if best_j >= 0:
            used.add(best_j)
            track["centroid"] = centroids[best_j]
            track["box"] = boxes[best_j]
            track["missed"] = 0
        else:
            track["missed"] += 1
    for tid in [t for t, tr in tracks.items() if tr["missed"] > MAX_MISSED]:
        del tracks[tid]
    for j, centroid in enumerate(centroids):
        if j not in used:
            tracks[next_id] = {"centroid": centroid, "box": boxes[j], "missed": 0}
            next_id += 1

    # Accumulate dwell time for vehicles whose centroid is inside the zone.
    for tid, track in tracks.items():
        cx, cy = track["centroid"]
        inside = cv2.pointPolygonTest(ZONE_POLY, (cx, cy), False) >= 0
        if inside:
            dwell[tid] = dwell.get(tid, 0.0) + 1.0 / fps
            if dwell[tid] >= STOPPED_SECONDS and tid not in flagged:
                flagged.add(tid)
                print(
                    f"STOPPED_TOO_LONG track={tid:<3} "
                    f"dwell={dwell[tid]:.1f}s time={fmt_time(now)} "
                    f"pos=({cx},{cy})", flush=True)
        else:
            dwell[tid] = 0.0

    cv2.polylines(frame, [ZONE_POLY], True, (0, 0, 255), 2)
    for tid, track in tracks.items():
        x1, y1, x2, y2, label = track["box"]
        color = (0, 0, 255) if tid in flagged else (0, 255, 0)
        cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
        cv2.putText(frame, f"{label} {dwell.get(tid, 0.0):.1f}s",
                    (x1, max(y1 - 6, 0)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
    writer.write(frame)

cap.release()
writer.release()
print(f"Total stopped-too-long vehicles: {len(flagged)}", flush=True)

Device targets:

• "CPU" -- default, works on all Intel platforms.
• "GPU" -- Intel integrated or discrete GPU.
• "NPU" -- Intel NPU (validate with benchmark_app -d NPU).

Expected Output

Each line prints the track ID, accumulated dwell time, the wall-clock time inside the video (MM:SS.mmm), and the vehicle position:

STOPPED_TOO_LONG track=0   dwell=5.0s time=00:05.033 pos=(984,489)
STOPPED_TOO_LONG track=4   dwell=5.0s time=00:22.867 pos=(660,442)
Total stopped-too-long vehicles: 2

DLStreamer Sample

Set up the environment:

source /opt/intel/openvino_2026/setupvars.sh
source /opt/intel/dlstreamer/scripts/setup_dls_env.sh
export PYTHONPATH=/opt/intel/dlstreamer/python:/opt/intel/dlstreamer/gstreamer/lib/python3/dist-packages:${PYTHONPATH:-}

Run stopped-too-long detection:

from collections import defaultdict
import json
import sys
import gi
gi.require_version("Gst", "1.0")
gi.require_version("GstAnalytics", "1.0")
gi.require_version("DLStreamerMeta", "1.0")
gi.require_version("DLStreamerWatermarkMeta", "1.0")
from gi.repository import Gst, GLib, GstAnalytics, DLStreamerMeta, DLStreamerWatermarkMeta

Gst.init([])

# Register DLStreamerMeta types so GstAnalytics iteration can handle them
_ov = sys.modules["gi.overrides.GstAnalytics"]
_ov.__mtd_types__[DLStreamerMeta.ZoneMtd.get_mtd_type()] = DLStreamerMeta.relation_meta_get_zone_mtd
_ov.__mtd_types__[DLStreamerMeta.TripwireMtd.get_mtd_type()] = DLStreamerMeta.relation_meta_get_tripwire_mtd

MODEL = "yolo26n_openvino_model/yolo26n.xml"
VIDEO = "ParkingVideo.mp4"
VEHICLE_LABELS = {"car", "motorcycle", "bus", "truck"}
ZONE_JSON = json.dumps([{
    "id": "no_stop_zone",
    "type": "polygon",
    "points": [{"x": 250, "y": 350}, {"x": 1700, "y": 350},
               {"x": 1700, "y": 900}, {"x": 250, "y": 900}]
}])
STOPPED_SECONDS = 5.0

pipeline = Gst.parse_launch(
    f"filesrc location={VIDEO} ! decodebin3 ! videoconvert ! "
    f"gvadetect model={MODEL} device=GPU threshold=0.4 ! queue ! "
    f"gvatrack tracking-type=short-term-imageless ! queue ! "
    f"gvaanalytics name=analytics draw-zones=true ! "
    f"gvafpscounter ! identity name=probe ! gvawatermark name=watermark ! "
    f"videoconvert ! video/x-raw,format=I420 ! "
    f"openh264enc ! h264parse ! mp4mux ! filesink location=output_dlstreamer.mp4"
)

pipeline.get_by_name("analytics").set_property("zones", ZONE_JSON)

dwell = defaultdict(float)
last_seen = {}
flagged = set()

def on_buffer(pad, info):
    buf = info.get_buffer()
    now = buf.pts / Gst.SECOND if buf.pts != Gst.CLOCK_TIME_NONE else 0.0
    rmeta = GstAnalytics.buffer_get_analytics_relation_meta(buf)
    if not rmeta:
        return Gst.PadProbeReturn.OK

    # Iterate only over object-detection entries
    for od in rmeta.iter_on_type(GstAnalytics.ODMtd):
        label = GLib.quark_to_string(od.get_obj_type())
        if label not in VEHICLE_LABELS:
            continue

        # Find tracking ID via direct relation
        track_id = None
        for trk in od.iter_direct_related(GstAnalytics.RelTypes.RELATE_TO, GstAnalytics.TrackingMtd):
            success, tracking_id, *_ = trk.get_info()
            if success:
                track_id = tracking_id
            break
        if track_id is None:
            continue

        # Check if gvaanalytics placed this detection inside the zone
        in_zone = False
        for zone in od.iter_direct_related(GstAnalytics.RelTypes.RELATE_TO, DLStreamerMeta.ZoneMtd):
            in_zone = True
            break

        if not in_zone:
            # Vehicle left the zone; reset its dwell accumulator.
            dwell[track_id] = 0.0
            last_seen.pop(track_id, None)
            continue

        # Accumulate dwell time for vehicles inside the zone
        dwell[track_id] += now - last_seen.get(track_id, now)
        last_seen[track_id] = now

        if dwell[track_id] >= STOPPED_SECONDS and track_id not in flagged:
            flagged.add(track_id)
            _, x, y, w, h, _ = od.get_location()
            print(f"STOPPED_TOO_LONG id={track_id} {label} "
                  f"dwell={dwell[track_id]:.1f}s pos=({int(x + w/2)},{int(y + h)})")

    return Gst.PadProbeReturn.OK

pipeline.get_by_name("probe").get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, on_buffer)
pipeline.set_state(Gst.State.PLAYING)
pipeline.get_bus().timed_pop_filtered(Gst.CLOCK_TIME_NONE, Gst.MessageType.EOS | Gst.MessageType.ERROR)
pipeline.set_state(Gst.State.NULL)

Expected output:

STOPPED_TOO_LONG id=1 car dwell=5.0s pos=(988,672)
STOPPED_TOO_LONG id=2 car dwell=5.0s pos=(665,583)
...

The annotated video is saved to output_dlstreamer.mp4. The gvaanalytics element also draws the zone polygon on each frame via gvawatermark.

Expected Output

Device targets:

• 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.

---

License

Licensed under the MIT License. See LICENSE for details.

References

• YOLO26 Documentation
• OpenVINO YOLO26 Notebook
• Intel DLStreamer Object Tracking
• Intel DLStreamer gvaanalytics
• OpenVINO Documentation
• NNCF Post-Training Quantization
• COCO Dataset