fight-and-violence-detection

Fight and Violence Detection

---

Overview

Fight and Violence Detection is a Metro Analytics use case that flags physically aggressive activity -- such as fighting, brawling, and sparring -- in images and video streams and raises an on-screen alert whenever violence is present. It is built on Qwen2-VL-2B-Instruct, a compact, state-of-the-art vision-language model (VLM) from Alibaba released under the Apache-2.0 license, exported to OpenVINO and weight-compressed to INT4 (default) or INT8 so it runs efficiently and fully locally on an Intel Core Ultra processor.

Rather than relying on a narrow, single-purpose detector trained on an unverified dataset, the VLM is prompted in natural language -- for example "Is there a physical fight between people in this image?" -- and its yes/no answer drives the alert. Both the OpenVINO and DLStreamer samples overlay a VIOLENCE DETECTED banner across the top of each frame when violence is present, so operators get an immediate, unambiguous alert.

Because the model is a general vision-language model, the same use case can be re-targeted to related behaviours (for example vandalism or aggressive crowding) simply by editing the prompt -- no retraining is required.

Typical Metro deployments include:

• Platform and Concourse Safety -- flag altercations on platforms, stairs, and concourses for rapid operator response.
• Ticket Hall and Gateline Monitoring -- detect fights and physical confrontations around fare gates and queues.
• Depot and Facility Security -- monitor restricted areas and back-of-house spaces for violent incidents.
• Automated Incident Escalation -- trigger alerts and video capture the moment violence is confirmed.

---

Prerequisites

• Python 3.11+
• Install OpenVINO (latest version)
• Install OpenVINO GenAI (latest version)
• Install Intel DLStreamer (latest version, with the OpenVINO GenAI option enabled for gvagenai)
• FFmpeg (used to transcode the sample video for the DLStreamer pipeline)

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 the Qwen2-VL-2B-Instruct model and export it to OpenVINO with weight compression:

chmod +x export_and_quantize.sh
./export_and_quantize.sh

This exports the model in INT4 precision (smallest, fastest on Core Ultra).

Optional: Select a Different Precision

./export_and_quantize.sh INT8   # higher accuracy, larger footprint
./export_and_quantize.sh FP16   # full-precision weights

The script performs the following steps:

1. Installs dependencies (optimum[openvino], openvino, openvino-genai, nncf, transformers, qwen-vl-utils).
2. Exports the Qwen2-VL-2B-Instruct weights to OpenVINO with the selected weight format via optimum-cli export openvino.
3. Downloads a Pexels-licensed sample sparring video, transcoding it to test_video.mp4.

Output files:

• qwen2_vl_2b_ov/ -- OpenVINO model directory (language model, vision encoder, tokenizer, and preprocessor config) ready for OpenVINO GenAI.
• test_video.mp4 -- transcoded sample clip.

Precision / Device Compatibility

OpenVINO Sample

The sample below runs the Qwen2-VL-2B-Instruct VLM on the sample video with OpenVINO GenAI. To keep inference responsive, one frame is sampled every FRAME_STRIDE frames and sent to the VLM with a short violence prompt; the yes/no answer is held between samples and overlaid as an alert banner across the top of each frame. The annotated result is written 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
import openvino_genai

# Change DEVICE to "GPU" or "NPU" to run on integrated GPU or NPU.
DEVICE = "CPU"
MODEL_DIR = "qwen2_vl_2b_ov"
PROMPT = (
    "Does this image show people boxing, punching, or physically fighting "
    "each other? Answer with a single word: yes or no."
)
# Run the VLM every FRAME_STRIDE frames; the alert is held between inferences.
FRAME_STRIDE = 15

properties = {}
if DEVICE == "GPU":
    properties["CACHE_DIR"] = "vlm_cache"
pipe = openvino_genai.VLMPipeline(MODEL_DIR, DEVICE, **properties)

config = openvino_genai.GenerationConfig()
config.max_new_tokens = 8

cap = cv2.VideoCapture("test_video.mp4")
fps = cap.get(cv2.CAP_PROP_FPS) or 25.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)
)


def ask_vlm(frame_bgr: np.ndarray) -> bool:
    """Return True when the VLM reports a fight or violence in the frame."""
    rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
    image = ov.Tensor(np.ascontiguousarray(rgb))
    result = pipe.generate(PROMPT, images=[image], generation_config=config)
    answer = str(result).strip().lower()
    return answer.startswith("yes") or "violence" in answer or "fight" in answer


frame_idx = 0
violence_frames = 0
detected = False
while True:
    ok, frame = cap.read()
    if not ok:
        break
    frame_idx += 1

    if (frame_idx - 1) % FRAME_STRIDE == 0:
        detected = ask_vlm(frame)

    if detected:
        violence_frames += 1
        cv2.rectangle(frame, (0, 0), (width, 40), (0, 0, 200), -1)
        cv2.putText(frame, "VIOLENCE DETECTED", (12, 28),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2)

    if frame_idx % 30 == 0:
        print(f"frame {frame_idx}: {'VIOLENCE DETECTED' if detected else 'normal'}",
              flush=True)

    writer.write(frame)

cap.release()
writer.release()
if violence_frames:
    print(f"VIOLENCE DETECTED in {violence_frames}/{frame_idx} frames")
print("Saved: output_openvino.mp4")

Device targets:

• "CPU" -- default, works on all Intel platforms.
• "GPU" -- Intel integrated or discrete GPU.
• "NPU" -- Intel NPU. The language model runs on the NPU while the vision encoder runs on CPU; validate the model loads with a short clip first.

Try It on a Sample Video

The export_and_quantize.sh script downloads and transcodes test_video.mp4 automatically. Re-run the OpenVINO sample above. The script reads test_video.mp4, prints a periodic status to the console, and writes the annotated video to output_openvino.mp4.

Expected console output (representative):

frame 30: VIOLENCE DETECTED
frame 60: VIOLENCE DETECTED
frame 90: normal

Expected Output

DLStreamer Sample

The pipeline below runs the same Qwen2-VL-2B-Instruct VLM on the sample video via the DLStreamer 2026 gvagenai element, which performs vision-language inference through OpenVINO GenAI. Frames are decoded, grouped into short chunks, and summarized by the VLM using a violence prompt; gvagenai attaches the answer to the buffer as JSON metadata. Frames are pulled through an appsink; for each frame a callback reads the latest VLM answer and overlays a VIOLENCE DETECTED banner across the top of the frame when violence is present. The annotated result is written to output_dlstreamer.mp4.

Notes on running this sample:

• Use the OpenVINO model directory produced by export_and_quantize.sh (qwen2_vl_2b_ov); gvagenai reads it via its model-path property.

• gvagenai requires an RGB input, so the decode chain converts to RGB before inference; the appsink then converts back to BGR and the banner is drawn with OpenCV, so no additional GStreamer overlay plugin is required.

• frame-rate controls how many frames per second are sampled for the VLM and chunk-size how many sampled frames form one inference call; keep both small to stay responsive on Core Ultra.

• The VLM answer is attached to the buffer as a GstGVAJSONMeta message and read in Python with gstgva.VideoFrame(buffer).messages().

• Export PYTHONPATH so the DLStreamer Python modules (gi, gstgva) are 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 json

import gi

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

Gst.init([])

# Import cv2 and gstgva after Gst.init to avoid a re-initialization conflict.
import cv2
import numpy as np
from gstgva import VideoFrame

MODEL_DIR = "qwen2_vl_2b_ov"
INPUT_VIDEO = "test_video.mp4"
PROMPT = (
    "Does this image show people boxing, punching, or physically fighting "
    "each other? Answer with a single word: yes or no."
)
ALERT_KEYWORDS = ("yes", "boxing", "punch", "fight", "violence")

# For CPU: change device=GPU to device=CPU.
# NPU is not supported by gvagenai (OpenVINO does not yet run VLMs on NPU);
# use the OpenVINO GenAI sample above to target the NPU.
pipeline_str = (
    f"filesrc location={INPUT_VIDEO} ! decodebin3 ! "
    f"videoconvert ! video/x-raw,format=RGB ! "
    f"gvagenai name=genai model-path={MODEL_DIR} device=GPU "
    f'prompt="{PROMPT}" generation-config="max_new_tokens=8" '
    f"frame-rate=2 chunk-size=2 ! queue ! "
    f"videoconvert ! video/x-raw,format=BGR ! "
    f"appsink name=sink emit-signals=true sync=false max-buffers=4 drop=false"
)
pipeline = Gst.parse_launch(pipeline_str)
appsink = pipeline.get_by_name("sink")

state = {"writer": None, "frame": 0, "violence": 0, "detected": False}


def on_sample(sink):
    sample = sink.emit("pull-sample")
    if sample is None:
        return Gst.FlowReturn.OK

    buf = sample.get_buffer()
    caps = sample.get_caps().get_structure(0)
    width = caps.get_value("width")
    height = caps.get_value("height")

    # Read the latest VLM answer from the gvagenai JSON metadata (if present).
    for message in VideoFrame(buf).messages():
        try:
            answer = str(json.loads(message).get("result", "")).strip().lower()
        except (ValueError, TypeError):
            continue
        if answer:
            state["detected"] = any(k in answer for k in ALERT_KEYWORDS)

    ok, mapinfo = buf.map(Gst.MapFlags.READ)
    if not ok:
        return Gst.FlowReturn.OK
    frame = np.frombuffer(mapinfo.data, np.uint8).reshape(height, width, 3).copy()
    buf.unmap(mapinfo)

    detected = state["detected"]
    if detected:
        state["violence"] += 1
        cv2.rectangle(frame, (0, 0), (width, 40), (0, 0, 200), -1)
        cv2.putText(frame, "VIOLENCE DETECTED", (12, 28),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2)

    if state["writer"] is None:
        ok_fps, fn, fd = caps.get_fraction("framerate")
        fps = fn / fd if ok_fps and fd > 0 else 25.0
        state["writer"] = cv2.VideoWriter(
            "output_dlstreamer.mp4",
            cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height),
        )
    state["writer"].write(frame)

    state["frame"] += 1
    if state["frame"] % 30 == 0:
        print(f"frame {state['frame']}: "
              f"{'VIOLENCE DETECTED' if detected else 'normal'}", flush=True)
    return Gst.FlowReturn.OK


appsink.connect("new-sample", on_sample)

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)

if state["writer"] is not None:
    state["writer"].release()
if state["violence"]:
    print(f"VIOLENCE DETECTED in {state['violence']}/{state['frame']} frames")
print("Saved: output_dlstreamer.mp4")

Try It on a Sample Video

The export_and_quantize.sh script downloads and transcodes test_video.mp4 automatically. Run the DLStreamer sample above. The callback prints a periodic status and writes the annotated video.

Expected console output (representative):

frame 30: VIOLENCE DETECTED
frame 60: VIOLENCE DETECTED
frame 90: normal

The annotated video is saved to output_dlstreamer.mp4 with the alert banner drawn by OpenCV.

Expected Output

Device targets:

• device=GPU -- default in the sample code.
• device=CPU -- change device=GPU to device=CPU.
• device=NPU -- not supported by gvagenai; OpenVINO does not yet run vision-language models on the NPU. Target the NPU with the OpenVINO GenAI sample above instead.

---

License

Licensed under the MIT License. See LICENSE for details.

References

• Qwen2-VL-2B-Instruct Model
• Qwen2-VL Technical Report
• Sample video: "Men doing sparring" by cottonbro studio (Pexels License), via Pexels
• OpenVINO Documentation
• OpenVINO GenAI
• Intel DLStreamer
• DLStreamer gvagenai element