Intel/crowd-analysis

Crowd Analysis

Property	Value
Category	Object Detection (Crowd Density + Movement)
Base Model	YOLO26 (Ultralytics)
Source Framework	PyTorch (Ultralytics)
Supported Precisions	FP32, FP16, INT8 (mixed-precision)
Inference Engine	OpenVINO
Hardware	CPU, GPU, NPU
Detected Class	person (COCO class 0)

---

Overview

Crowd Analysis is a Metro Analytics use case that estimates crowd density and movement patterns in video streams. It detects people frame by frame, reports a per-frame count with a simple density level (LOW / MEDIUM / HIGH), and tracks each person across frames to estimate the dominant flow direction of the crowd. It is built on YOLO26, a state-of-the-art real-time object detector trained on the COCO dataset, exported to OpenVINO IR and filtered at runtime to the person class.

Typical Metro deployments include:

• Platform & Concourse Density -- gauge how crowded station platforms and concourses are and flag build-up before it becomes unsafe.
• Pedestrian Flow Analysis -- estimate the dominant direction people move through corridors, gates, and crossings.
• Public-Venue Occupancy -- monitor crowd density at stadiums, transit hubs, and event entrances.
• Situational Awareness -- combine density level and flow to support operator decisions in public venues and transportation hubs.

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

Density levels are defined by two count thresholds (defaults: LOW for fewer than 10 people, MEDIUM for 10-25, HIGH for more than 25). Tune these to the field of view and expected occupancy of your deployment site.

---

Prerequisites

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

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 a sample test image (test.jpg) and a sample test video (test_video.mp4).
3. Downloads the PyTorch weights and exports to OpenVINO IR.
4. (INT8 only) Quantizes the model using NNCF post-training quantization.

The sample video is a free-to-use pedestrians-crossing-the-street clip from Pexels.

Output files:

• yolo26n_openvino_model/ -- FP32 or FP16 OpenVINO IR model directory.
• yolo26n_crowdanalysis_int8.xml / yolo26n_crowdanalysis_int8.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

Note: The INT8 calibration uses the bundled sample image. For production accuracy, replace it with a representative set of frames from the target deployment site.

OpenVINO Sample

The sample below runs YOLO26 inference on the sample video, filters to the person class, reports the crowd count and density level per frame, tracks each person with a lightweight IoU tracker to estimate the dominant flow direction, and writes the annotated result to output_openvino.mp4. Change the device string to run on CPU, GPU, or NPU.

import cv2
import numpy as np
import openvino as ov

PERSON_CLASS_ID = 0
CONF_THRESHOLD = 0.4
INPUT_SIZE = 640

# Crowd-density thresholds (person count per frame).
DENSITY_LOW_MAX = 10      # fewer than 10 -> LOW
DENSITY_MEDIUM_MAX = 25   # 10-25 -> MEDIUM, more than 25 -> HIGH

# Movement tracking.
IOU_MATCH_THRESHOLD = 0.3
MAX_MISSED_FRAMES = 15


def density_level(count):
    if count < DENSITY_LOW_MAX:
        return "LOW", (0, 200, 0)
    if count <= DENSITY_MEDIUM_MAX:
        return "MEDIUM", (0, 200, 255)
    return "HIGH", (0, 0, 255)


def iou(box_a, box_b):
    ax1, ay1, ax2, ay2 = box_a
    bx1, by1, bx2, by2 = box_b
    ix1, iy1 = max(ax1, bx1), max(ay1, by1)
    ix2, iy2 = min(ax2, bx2), min(ay2, by2)
    inter = max(0, ix2 - ix1) * max(0, iy2 - iy1)
    if inter == 0:
        return 0.0
    area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1)
    area_b = max(0, bx2 - bx1) * max(0, by2 - by1)
    return inter / float(area_a + area_b - inter)


class CentroidTracker:
    """Minimal IoU tracker that records each track's last centroid so we can
    estimate per-frame movement (flow) vectors."""

    def __init__(self):
        self._next_id = 1
        self._tracks = {}  # id -> {"box", "centroid", "missed"}

    def update(self, boxes):
        unmatched = set(self._tracks)
        assignments, moves = [], []
        for box in boxes:
            cx = (box[0] + box[2]) / 2.0
            cy = (box[1] + box[3]) / 2.0
            best_id, best_iou = None, IOU_MATCH_THRESHOLD
            for tid in unmatched:
                score = iou(box, self._tracks[tid]["box"])
                if score > best_iou:
                    best_id, best_iou = tid, score
            if best_id is not None:
                tid = best_id
                unmatched.discard(tid)
                pcx, pcy = self._tracks[tid]["centroid"]
                moves.append((cx - pcx, cy - pcy))
            else:
                tid = self._next_id
                self._next_id += 1
            self._tracks[tid] = {"box": box, "centroid": (cx, cy), "missed": 0}
            assignments.append((box, tid))
        for tid in unmatched:
            self._tracks[tid]["missed"] += 1
            if self._tracks[tid]["missed"] > MAX_MISSED_FRAMES:
                del self._tracks[tid]
        return assignments, moves


core = ov.Core()
model = core.read_model("yolo26n_openvino_model/yolo26n.xml")
compiled = core.compile_model(model, "CPU")  # or "GPU", "NPU"

cap = cv2.VideoCapture("test_video.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))
writer = cv2.VideoWriter(
    "output_openvino.mp4", cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height))

tracker = CentroidTracker()

while True:
    ok, frame = cap.read()
    if not ok:
        break
    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, ...]  # NCHW

    # YOLO26 end-to-end output: [1, 300, 6] = [x1, y1, x2, y2, confidence, class_id]
    output = compiled([blob])[compiled.output(0)][0]
    mask = (output[:, 4] >= CONF_THRESHOLD) & (output[:, 5].astype(int) == PERSON_CLASS_ID)
    dets = output[mask]

    boxes = [(d[0] * sx, d[1] * sy, d[2] * sx, d[3] * sy) for d in dets]
    assignments, moves = tracker.update(boxes)

    for box, _tid in assignments:
        x1, y1, x2, y2 = (int(v) for v in box)
        cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)

    count = len(boxes)
    level, color = density_level(count)
    cv2.putText(frame, f"Crowd: {count} ({level})", (10, 40),
                cv2.FONT_HERSHEY_SIMPLEX, 1.0, color, 2)

    # Movement: mean of all per-track displacements -> dominant flow arrow.
    if moves:
        mdx = float(np.mean([m[0] for m in moves]))
        mdy = float(np.mean([m[1] for m in moves]))
        ox, oy = width // 2, height - 40
        cv2.arrowedLine(frame, (ox, oy),
                        (int(ox + mdx * 10), int(oy + mdy * 10)),
                        (255, 0, 0), 3, tipLength=0.3)
        cv2.putText(frame, f"Flow dx={mdx:+.1f} dy={mdy:+.1f}", (10, 75),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 0, 0), 2)

    writer.write(frame)

cap.release()
writer.release()
print("Saved: output_openvino.mp4")

Try It on the Sample Video

The export_and_quantize.sh script downloads test_video.mp4 automatically. Run the OpenVINO sample above. It reads test_video.mp4, prints the crowd count and density level per frame, and writes the annotated video to output_openvino.mp4 with a green box around each detected person, the Crowd: N (LEVEL) overlay, and a blue arrow showing the dominant crowd flow.

Tip: For production testing, replace the bundled test_video.mp4 with footage from your target deployment site and re-tune the density thresholds.

Expected Output

DLStreamer Sample

The pipeline below runs the FP16 YOLO26 detector on the sample video via gvadetect, assigns a stable track ID to each person with gvatrack, filters detections to the person class in a buffer probe using the GStreamer Analytics metadata API (GstAnalytics), overlays bounding boxes, and saves the annotated result to output_dlstreamer.mp4. The probe prints the crowd count, density level, and dominant flow direction per frame.

Notes on running this sample:

• Use the FP16 IR (yolo26n_openvino_model/yolo26n.xml). On DLStreamer 2026.0.0, gvadetect cannot auto-derive a YOLO post-processor from the INT8 model produced by the bundled script. To use the INT8 model, supply a matching model-proc JSON.

• Class names are read automatically from the model's embedded metadata.yaml by DLStreamer 2026.0+ -- no external labels-file is required.

• Filtering with object-class=person directly on gvadetect is rejected when inference-region is full-frame (the default), so the sample filters by detection label in the buffer probe instead.

• Export PYTHONPATH so the DLStreamer Python module is importable:

  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:-}
  

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstAnalytics", "1.0")
from gi.repository import Gst, GLib, GstAnalytics

Gst.init([])

INPUT_VIDEO = "test_video.mp4"

# Crowd-density thresholds (person count per frame).
DENSITY_LOW_MAX = 10      # fewer than 10 -> LOW
DENSITY_MEDIUM_MAX = 25   # 10-25 -> MEDIUM, more than 25 -> HIGH


def density_level(count):
    if count < DENSITY_LOW_MAX:
        return "LOW"
    if count <= DENSITY_MEDIUM_MAX:
        return "MEDIUM"
    return "HIGH"


# 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=zero-term-imageless ! queue ! "
    "gvawatermark displ-cfg=show-roi=person ! "
    "videoconvert ! video/x-raw,format=I420 ! "
    "openh264enc ! h264parse ! "
    "mp4mux ! filesink name=sink location=output_dlstreamer.mp4"
)
pipeline = Gst.parse_launch(pipeline_str)

sink = pipeline.get_by_name("sink")
sink_pad = sink.get_static_pad("sink")

prev_centroid = {}  # track_id -> (cx, cy)


def on_buffer(pad, info):
    buf = info.get_buffer()
    rmeta = GstAnalytics.buffer_get_analytics_relation_meta(buf)
    if rmeta is None:
        return Gst.PadProbeReturn.OK

    # OD and tracking metadata share one id space and can be interleaved
    # (id=1 -> ODMtd, id=2 -> TrackingMtd, ...), so scan every id and stop only
    # after several consecutive misses.
    ods, tracks = [], []
    idx, misses = 1, 0
    while misses < 20:
        ok_od, od = rmeta.get_od_mtd(idx)
        ok_trk, trk = rmeta.get_tracking_mtd(idx)
        if ok_od:
            ods.append(od)
            misses = 0
        elif ok_trk:
            tracks.append(trk)
            misses = 0
        else:
            misses += 1
        idx += 1

    count, moves = 0, []
    for od in ods:
        if GLib.quark_to_string(od.get_obj_type()) != "person":
            continue
        count += 1
        _, x, y, w, h, _ = od.get_location()
        cx, cy = x + w / 2.0, y + h / 2.0
        for trk in tracks:
            if rmeta.get_relation(od.id, trk.id) == GstAnalytics.RelTypes.NONE:
                continue
            ok_trk, track_id, _, _, _ = trk.get_info()
            if not ok_trk:
                continue
            if track_id in prev_centroid:
                pcx, pcy = prev_centroid[track_id]
                moves.append((cx - pcx, cy - pcy))
            prev_centroid[track_id] = (cx, cy)
            break

    if count:
        level = density_level(count)
        if moves:
            mdx = sum(m[0] for m in moves) / len(moves)
            mdy = sum(m[1] for m in moves) / len(moves)
            print(f"Crowd: {count} ({level})  flow dx={mdx:+.1f} dy={mdy:+.1f}",
                  flush=True)
        else:
            print(f"Crowd: {count} ({level})", flush=True)
    return Gst.PadProbeReturn.OK


sink_pad.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)

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
• COCO Dataset
• OpenVINO Documentation
• NNCF Post-Training Quantization
• Intel DLStreamer
• Sample video: Pedestrians crossing the street (Pexels)