| |
| |
| |
| |
| |
|
|
| import gi |
|
|
| gi.require_version("Gst", "1.0") |
| from gi.repository import Gst |
|
|
| import numpy as np |
|
|
| Gst.init([]) |
|
|
| |
| import cv2 |
|
|
| INPUT_VIDEO = "test_video.mp4" |
| MIN_AREA = 500 |
|
|
| |
| 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) |
|
|
