vagheshpatel's picture
Sync vehicle-entry-exit-logging from metro-analytics-catalog
d9ece88 verified
|
Raw
History Blame Contribute Delete
13.8 kB
---
license: mit
license_link: LICENSE
library_name: openvino
pipeline_tag: object-detection
tags:
- openvino
- intel
- yolo
- yolo26
- vehicle-entry-exit
- tracking
- short-term-imageless
- line-crossing
- edge-ai
- metro
- dlstreamer
language:
- en
---
# Vehicle Entry/Exit Logging
| Property | Value |
|---|---|
| **Category** | Object Detection + Tracking + Line Crossing |
| **Base Model** | [YOLO26](https://docs.ultralytics.com/models/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) |
---
## Overview
Vehicle Entry/Exit Logging is a Metro Analytics use case that detects vehicles,
tracks them across frames with BoT-SORT, and logs directional entry and exit
events when a tracked vehicle crosses a configurable virtual line.
It is built on [YOLO26](https://docs.ultralytics.com/models/yolo26/), a
state-of-the-art real-time object detector, quantized to INT8 for efficient
inference on Intel hardware.
The tracking and line-crossing logic runs as a thin post-processing layer on
top of the strongest vehicle detector available, keeping the solution accurate
and extensible to other zone shapes.
Typical Metro deployments include:
- **Parking Garage Management** -- count vehicles entering and leaving a lot.
- **Toll Gate Analytics** -- log each vehicle that passes through a toll point.
- **Depot and Fleet Monitoring** -- track bus/truck entry and exit from depots.
- **Traffic Flow Analysis** -- measure directional flow at intersections.
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](https://docs.openvino.ai/2026/get-started/install-openvino.html) (latest version)
- [Install Intel DLStreamer](https://docs.openedgeplatform.intel.com/2026.0/edge-ai-libraries/dlstreamer/get_started/install/install_guide_ubuntu.html) (latest version)
Create and activate a Python virtual environment before running the scripts:
```bash
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:
```bash
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
```bash
./export_and_quantize.sh yolo26n FP32 # full-precision
./export_and_quantize.sh yolo26n INT8 # quantized
./export_and_quantize.sh yolo26s # larger variant, default FP16
```
The script performs the following steps:
1. Installs dependencies (`openvino`, `ultralytics`; adds `nncf` for INT8).
2. Downloads a sample test image (`test.jpg`) and the smart-parking sample video (`smart_parking_720p_30fps.mp4`).
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_vehicle_entry_exit_int8.xml` / `.bin` -- INT8 quantized model *(only when `INT8` is selected)*.
#### Precision / Device Compatibility
| Precision | CPU | GPU | NPU |
|---|---|---|---|
| FP32 | Yes | Yes | No |
| FP16 | Yes | Yes | Yes |
| INT8 | Yes | Yes | Yes |
### OpenVINO Sample
The sample below runs YOLO26 inference on a video, keeps only the `car` class,
applies simple centroid tracking with track IDs, and logs an entry or exit
event -- with the wall-clock timestamp inside the video -- when a tracked car's
centroid crosses a horizontal virtual line placed at 60% of the frame height.
The saved output video shows only the car detection bounding boxes (no counter
overlay or line).
Change the `device` string to run on CPU, GPU, or NPU.
```python
import cv2
import numpy as np
import openvino as ov
VEHICLE_CLASS_IDS = {2: "car"}
CONF_THRESHOLD = 0.4
INPUT_SIZE = 640
LINE_RATIO = 0.6
MAX_DIST = 80
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("smart_parking_720p_30fps.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))
line_y = int(height * LINE_RATIO)
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))
tracks: dict[int, tuple[int, int]] = {}
entry_time: dict[int, float] = {}
next_id = 0
entered = 0
exited = 0
frame_idx = 0
while True:
ok, frame = cap.read()
if not ok:
break
frame_idx += 1
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]
centroids = []
for det in dets:
cx = int(((det[0] + det[2]) / 2) * sx)
cy = int(((det[1] + det[3]) / 2) * sy)
centroids.append((cx, cy))
new_tracks: dict[int, tuple[int, int]] = {}
used = set()
for tid, (px, py) in tracks.items():
best_d = MAX_DIST
best_j = -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 = d
best_j = j
if best_j >= 0:
cx, cy = centroids[best_j]
used.add(best_j)
t = frame_idx / fps
if py < line_y <= cy:
exited += 1
enter_t = entry_time.pop(tid, None)
if enter_t is not None:
print(
f"EXIT track={tid:<3} entry={fmt_time(enter_t)} "
f"exit={fmt_time(t)}", flush=True)
else:
print(f"EXIT track={tid:<3} exit={fmt_time(t)}", flush=True)
elif py >= line_y > cy:
entered += 1
entry_time[tid] = t
print(f"ENTRY track={tid:<3} entry={fmt_time(t)}", flush=True)
new_tracks[tid] = (cx, cy)
for j, (cx, cy) in enumerate(centroids):
if j not in used:
new_tracks[next_id] = (cx, cy)
next_id += 1
tracks = new_tracks
for det in dets:
x1 = int(det[0] * sx)
y1 = int(det[1] * sy)
x2 = int(det[2] * sx)
y2 = int(det[3] * sy)
cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
cv2.putText(frame, "car", (x1, max(y1 - 6, 0)),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
writer.write(frame)
cap.release()
writer.release()
print(f"Total: entered={entered} exited={exited}", 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 with the entry timestamp, and on exit the paired
entry and exit timestamps (`MM:SS.mmm` within the video):
```text
ENTRY track=3 entry=00:02.400
ENTRY track=7 entry=00:05.133
EXIT track=3 entry=00:02.400 exit=00:09.867
ENTRY track=12 entry=00:11.267
EXIT track=7 entry=00:05.133 exit=00:14.700
EXIT track=12 entry=00:11.267 exit=00:18.933
Total: entered=3 exited=3
```
![OpenVINO expected output](expected_output_openvino.gif)
### DLStreamer Sample
The pipeline below runs the FP16 YOLO26 detector with `gvatrack`
(BoT-SORT) for stable vehicle IDs, keeping only the `car` class.
A buffer probe reads the tracking metadata and fires entry/exit events
-- logging the entry and exit timestamps taken from each buffer's
presentation time -- when a tracked car crosses the virtual line.
The annotated result is saved 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 embedded `metadata.yaml` by
> DLStreamer 2026.0+ -- no external `labels-file` is required.
> - Detections are read with the `gstgva` `VideoFrame` API
> (`region.object_id()` carries the `gvatrack` ID).
> - Export `PYTHONPATH` so the DLStreamer Python module is importable:
>
> ```bash
> 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:-}
> ```
```python
import gi
gi.require_version("Gst", "1.0")
from gi.repository import Gst
from gstgva import VideoFrame
Gst.init([])
INPUT_VIDEO = "smart_parking_720p_30fps.mp4"
VEHICLE_LABELS = {"car"}
LINE_RATIO = 0.6
# For CPU: change device=GPU to device=CPU.
# For NPU: change device=GPU to device=NPU (batch-size=1, nireq=4 recommended).
pipeline_str = (
f"filesrc location={INPUT_VIDEO} ! decodebin3 ! "
"videoconvert ! "
"gvadetect model=yolo26n_openvino_model/yolo26n.xml "
"device=GPU "
"threshold=0.4 ! queue ! "
"gvatrack tracking-type=short-term-imageless ! queue ! "
"identity name=probe ! "
"gvawatermark displ-cfg=show-roi=car ! "
"videoconvert ! video/x-raw,format=I420 ! "
"openh264enc ! h264parse ! "
"mp4mux ! filesink location=output_dlstreamer.mp4"
)
pipeline = Gst.parse_launch(pipeline_str)
prev_positions: dict[int, int] = {}
entry_time: dict[int, float] = {}
entered = 0
exited = 0
frame_height = 0
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}"
def on_buffer(pad, info):
global entered, exited, frame_height
buf = info.get_buffer()
caps = pad.get_current_caps()
if caps and frame_height == 0:
frame_height = caps.get_structure(0).get_value("height") or 720
line_y = int(frame_height * LINE_RATIO)
t = buf.pts / Gst.SECOND if buf.pts != Gst.CLOCK_TIME_NONE else 0.0
frame = VideoFrame(buf, caps=caps)
current: dict[int, int] = {}
for region in frame.regions():
if region.label() not in VEHICLE_LABELS:
continue
rect = region.rect()
cy = int(rect.y + rect.h / 2)
tid = region.object_id()
current[tid] = cy
if tid in prev_positions:
py = prev_positions[tid]
if py < line_y <= cy:
exited += 1
enter_t = entry_time.pop(tid, None)
if enter_t is not None:
print(
f"EXIT track={tid:<3} entry={fmt_time(enter_t)} "
f"exit={fmt_time(t)}", flush=True)
else:
print(f"EXIT track={tid:<3} exit={fmt_time(t)}", flush=True)
elif py >= line_y > cy:
entered += 1
entry_time[tid] = t
print(f"ENTRY track={tid:<3} entry={fmt_time(t)}", flush=True)
prev_positions.clear()
prev_positions.update(current)
return Gst.PadProbeReturn.OK
probe = pipeline.get_by_name("probe")
probe.get_static_pad("src").add_probe(Gst.PadProbeType.BUFFER, on_buffer)
pipeline.set_state(Gst.State.PLAYING)
bus = pipeline.get_bus()
bus.timed_pop_filtered(
Gst.CLOCK_TIME_NONE,
Gst.MessageType.EOS | Gst.MessageType.ERROR,
)
pipeline.set_state(Gst.State.NULL)
print(f"Total: entered={entered} exited={exited}", flush=True)
```
**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.
#### Expected Output
The terminal logs each vehicle's entry timestamp and, on exit, the paired
entry and exit timestamps (`MM:SS.mmm` within the video):
```text
ENTRY track=1 entry=00:01.900
ENTRY track=4 entry=00:04.633
EXIT track=1 entry=00:01.900 exit=00:08.767
ENTRY track=9 entry=00:10.500
EXIT track=4 entry=00:04.633 exit=00:13.400
EXIT track=9 entry=00:10.500 exit=00:17.833
Total: entered=3 exited=3
```
![DLStreamer expected output](expected_output_dlstreamer.gif)
---
## License
Licensed under the MIT License. See [LICENSE](LICENSE) for details.
## References
- [YOLO26 Documentation](https://docs.ultralytics.com/models/yolo26/)
- [Intel DLStreamer gvatrack](https://docs.openedgeplatform.intel.com/2026.0/edge-ai-libraries/dlstreamer/elements/gvatrack.html)
- [BoT-SORT: Robust Multi-Object Tracking](https://arxiv.org/abs/2206.14651)
- [OpenVINO Documentation](https://docs.openvino.ai/)
- [Intel DLStreamer](https://docs.openedgeplatform.intel.com/2026.0/edge-ai-libraries/dlstreamer/index.html)