File size: 2,554 Bytes
85d519e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | # SPDX-License-Identifier: MIT
# Copyright (C) Intel Corporation
#
# Motion Detection: DLStreamer pipeline sample.
# See README.md for prerequisites and inputs. Run: python3 dlstreamer_sample.py
import gi
gi.require_version("Gst", "1.0")
from gi.repository import Gst
import numpy as np
Gst.init([])
# Import cv2 after Gst.init to avoid GStreamer re-initialization conflicts.
import cv2
INPUT_VIDEO = "test_video.mp4"
MIN_AREA = 500
# Decode with the DLStreamer/GStreamer stack and hand BGR frames to OpenCV.
pipeline_str = (
f"filesrc location={INPUT_VIDEO} ! decodebin3 ! videoconvert ! "
"video/x-raw,format=BGR ! "
"appsink name=sink emit-signals=false sync=false"
)
pipeline = Gst.parse_launch(pipeline_str)
sink = pipeline.get_by_name("sink")
pipeline.set_state(Gst.State.PLAYING)
bg = cv2.createBackgroundSubtractorMOG2(
history=200, varThreshold=25, detectShadows=True)
kernel = np.ones((3, 3), np.uint8)
writer = None
frame_idx = 0
motion_frames = 0
while True:
sample = sink.emit("pull-sample")
if sample is None:
break
buf = sample.get_buffer()
caps = sample.get_caps().get_structure(0)
width = caps.get_value("width")
height = caps.get_value("height")
ok, mapinfo = buf.map(Gst.MapFlags.READ)
if not ok:
continue
frame = np.ndarray((height, width, 3), dtype=np.uint8,
buffer=mapinfo.data).copy()
buf.unmap(mapinfo)
frame_idx += 1
fg = bg.apply(frame)
fg = cv2.threshold(fg, 200, 255, cv2.THRESH_BINARY)[1]
fg = cv2.morphologyEx(fg, cv2.MORPH_OPEN, kernel)
contours, _ = cv2.findContours(
fg, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
regions = [c for c in contours if cv2.contourArea(c) >= MIN_AREA]
if regions:
motion_frames += 1
for c in regions:
x, y, w, h = cv2.boundingRect(c)
cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 0, 255), 2)
status_text = "Motion Detected" if regions else "No Motion"
status_color = (0, 0, 255) if regions else (0, 255, 0)
cv2.putText(frame, status_text, (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.9, status_color, 2)
if writer is None:
writer = cv2.VideoWriter(
"output_dlstreamer.mp4", cv2.VideoWriter_fourcc(*"mp4v"),
30.0, (width, height))
writer.write(frame)
print(f"Frame {frame_idx}: motion regions={len(regions)}", flush=True)
pipeline.set_state(Gst.State.NULL)
if writer:
writer.release()
print(f"Motion detected in {motion_frames} frames", flush=True)
|