diff --git a/.gitignore b/.gitignore index e964a677..58c8de58 100644 --- a/.gitignore +++ b/.gitignore @@ -6,7 +6,7 @@ __pycache__/ .Python # Outputs -outputs/ +output/ dataset/ data/classification_dataset data/output_images/ diff --git a/docs/access_raw_frames.md b/docs/access_raw_frames.md new file mode 100644 index 00000000..2c7b5210 --- /dev/null +++ b/docs/access_raw_frames.md @@ -0,0 +1,115 @@ +# Accessing Raw Frames in DeepStream Pipeline + +## Overview + +This document describes how to access and manipulate raw video frames within a DeepStream pipeline using pad probes. This mechanism allows custom frame processing outside of standard DeepStream plugins. + +## Architecture + +### Pipeline Structure + +``` +RTSP Source → Decode → VideoConvert → NvVideoConvert → CapsFilter → StreamMux → ... + ↑ + [Probe Point] +``` + +### Key Components + +1. **nvvideoconvert**: Converts frames to NVMM (NVIDIA Memory Management) format for GPU processing +2. **capsfilter**: Ensures RGBA format output (`video/x-raw(memory:NVMM), format=RGBA`) +3. **Pad Probe**: Attached to capsfilter's src pad to intercept buffers + +## Implementation Details + +### 1. Probe Attachment Point + +The probe is attached **after** the capsfilter element, ensuring frames are in RGBA format on GPU memory: + +```python +def attach_buffer_probe(elems, buffer_probe): + srcpad = elems["capsfilter"].get_static_pad("src") + srcpad.add_probe(Gst.PadProbeType.BUFFER, buffer_probe) +``` + +**Why this location?** +- Frames are in NVMM memory (GPU accessible) +- Format is standardized (RGBA) +- Before heavy processing elements (inference, OSD) + +### 2. Frame Extraction + +The probe callback extracts frames using PyDS (DeepStream Python bindings): + +```python +def buffer_probe(pad, info): + buf = info.get_buffer() + if not buf: + return Gst.PadProbeReturn.OK + + # Retrieve NvBufSurface from GPU memory + surface = pyds.get_nvds_buf_surface(hash(buf), 0) + + # Copy to CPU as NumPy array (RGBA format) + frame = np.array(surface, copy=True, order='C') +``` + +**Key Functions:** +- `pyds.get_nvds_buf_surface(hash(buf), 0)`: Gets GPU surface from buffer +- `np.array(surface, copy=True)`: Copies data from GPU to CPU for manipulation + +### 3. Frame Manipulation + +Once in NumPy format, frames can be manipulated using standard Python operations: + +```python +# Example 1: Draw red rectangle at top +h, w, _ = frame.shape +frame[0:20, 0:w] = [255, 0, 0, 255] # RGBA: red bar, 20px height + +# Example 2: Darken entire frame +frame = (frame * 0.7).astype(np.uint8) +``` + +### 4. Writing Back to Pipeline + +After manipulation, copy the modified frame back to GPU: + +```python +# Copy modified frame back to GPU surface +np.copyto(np.array(surface, copy=False, order='C'), frame) + +return Gst.PadProbeReturn.OK +``` + +**Important:** Use `copy=False` to get a writable view of the GPU memory. + +## Format Requirements + +### Capsfilter Configuration + +```python +capsfilter.set_property("caps", + Gst.Caps.from_string("video/x-raw(memory:NVMM), format=RGBA")) +``` + +- **Memory Type**: `NVMM` (required for `get_nvds_buf_surface`) +- **Format**: `RGBA` (4 channels, 8-bit per channel) +- **Frame Shape**: `(height, width, 4)` in NumPy + +## References + +- [NVIDIA DeepStream Python Apps](https://github.com/NVIDIA-AI-IOT/deepstream_python_apps) +- [GStreamer Pad Probes](https://gstreamer.freedesktop.org/documentation/additional/design/probes.html) +- [DeepStream Python Bindings (PyDS)](https://docs.nvidia.com/metropolis/deepstream/python-api/index.html) + +## Summary + +This mechanism provides a flexible way to access and manipulate raw frames in DeepStream pipelines: + +- **Probe Location**: After capsfilter (RGBA, NVMM format) +- **Extraction**: `pyds.get_nvds_buf_surface()` + NumPy array +- **Manipulation**: Standard NumPy/OpenCV operations +- **Write-back**: `np.copyto()` to GPU surface + +This approach enables custom frame processing while maintaining DeepStream pipeline efficiency. \ No newline at end of file diff --git a/env/arm/Dockerfile b/env/arm/Dockerfile index e6517570..207ede60 100644 --- a/env/arm/Dockerfile +++ b/env/arm/Dockerfile @@ -27,10 +27,14 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ gstreamer1.0-plugins-bad \ gstreamer1.0-plugins-ugly \ gstreamer1.0-libav \ + libcairo2-dev pkg-config \ && rm -rf /var/lib/apt/lists/* RUN pip3 install --no-cache-dir --upgrade pip wheel setuptools +COPY pyds-1.1.10-py3-none-linux_aarch64.whl /tmp/ +RUN pip3 install /tmp/pyds-1.1.10-py3-none-linux_aarch64.whl + RUN groupadd -g ${USER_ID} ${USER_NAME} || true \ && useradd -m -s /bin/bash -u ${USER_ID} -g ${USER_ID} ${USER_NAME} || true diff --git a/env/arm/pyds-1.1.10-py3-none-linux_aarch64.whl b/env/arm/pyds-1.1.10-py3-none-linux_aarch64.whl new file mode 100644 index 00000000..2e978d65 Binary files /dev/null and b/env/arm/pyds-1.1.10-py3-none-linux_aarch64.whl differ diff --git a/src/access_raw_frames_pipeline.py b/src/access_raw_frames_pipeline.py new file mode 100644 index 00000000..b1e8e7f6 --- /dev/null +++ b/src/access_raw_frames_pipeline.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +DeepStream Pipeline +RTSP → Decode → Convert → Save Images +""" + +import gi +gi.require_version("Gst", "1.0") +from gi.repository import Gst, GLib +import os +import sys +import numpy as np +import pyds + +RTSP_URL = "rtsp://127.0.0.1:8554/test" +OUTPUT_DIR = "/workspace/output/frames/" + +def buffer_probe(pad, info): + buf = info.get_buffer() + if not buf: + return Gst.PadProbeReturn.OK + + # Retrieve NvBufSurface from GPU memory + surface = pyds.get_nvds_buf_surface(hash(buf), 0) + + # Copy to CPU (RGBA format) + frame = np.array(surface, copy=True, order='C') + + # draw a red rectangle at the top of the image + h, w, _ = frame.shape + frame[0:20, 0:w] = [255, 0, 0, 255] # red bar, 20 pixels height + + # Copy back to GPU + np.copyto(np.array(surface, copy=False, order='C'), frame) + + return Gst.PadProbeReturn.OK + + +def on_message(bus, msg, loop): + if msg.type == Gst.MessageType.ERROR: + err, debug = msg.parse_error() + print(f"ERROR: {err}, debug: {debug}") + loop.quit() + elif msg.type == Gst.MessageType.EOS: + print("End of stream") + loop.quit() + + +def create_elements(): + rtspsrc = Gst.ElementFactory.make("rtspsrc", "source") + depay = Gst.ElementFactory.make("rtph264depay", "depay") + parse = Gst.ElementFactory.make("h264parse", "parse") + decode = Gst.ElementFactory.make("decodebin", "decode") + convert1 = Gst.ElementFactory.make("videoconvert", "convert1") + nvvideoconvert1 = Gst.ElementFactory.make("nvvideoconvert", "nvvideoconvert1") + + capsfilter = Gst.ElementFactory.make("capsfilter", "capsfilter") + capsfilter.set_property("caps", Gst.Caps.from_string("video/x-raw(memory:NVMM), format=RGBA")) + + streammux = Gst.ElementFactory.make("nvstreammux", "streammux") + streammux.set_property("batch-size", 1) + streammux.set_property("width", 640) + streammux.set_property("height", 480) + + nvvideoconvert2 = Gst.ElementFactory.make("nvvideoconvert", "nvvideoconvert2") + jpegenc = Gst.ElementFactory.make("jpegenc", "jpegenc") + sink = Gst.ElementFactory.make("multifilesink", "sink") + sink.set_property("location", os.path.join(OUTPUT_DIR, "frame_%05d.jpg")) + + elements = [rtspsrc, depay, parse, decode, convert1, nvvideoconvert1, + capsfilter, streammux, nvvideoconvert2, jpegenc, sink] + + if not all(elements): + print("ERROR: Failed to create one or more GStreamer elements", file=sys.stderr) + sys.exit(1) + + return { + "rtspsrc": rtspsrc, + "depay": depay, + "parse": parse, + "decode": decode, + "convert1": convert1, + "nvvideoconvert1": nvvideoconvert1, + "capsfilter": capsfilter, + "streammux": streammux, + "nvvideoconvert2": nvvideoconvert2, + "jpegenc": jpegenc, + "sink": sink + } + + +def link_pipeline_elements(pipeline, elems): + for elem in [elems["depay"], elems["parse"], elems["decode"], elems["convert1"], + elems["nvvideoconvert1"], elems["capsfilter"], elems["streammux"], + elems["nvvideoconvert2"], elems["jpegenc"], elems["sink"]]: + pipeline.add(elem) + + elems["depay"].link(elems["parse"]) + elems["parse"].link(elems["decode"]) + elems["convert1"].link(elems["nvvideoconvert1"]) + + elems["nvvideoconvert1"].link(elems["capsfilter"]) + sinkpad = elems["streammux"].get_request_pad("sink_0") + srcpad = elems["capsfilter"].get_static_pad("src") + srcpad.link(sinkpad) + + elems["streammux"].link(elems["nvvideoconvert2"]) + elems["nvvideoconvert2"].link(elems["jpegenc"]) + elems["jpegenc"].link(elems["sink"]) + + +def connect_dynamic_links(elems): + def on_pad_added(src, new_pad): + sink_pad = elems["depay"].get_static_pad("sink") + if not sink_pad.is_linked(): + new_pad.link(sink_pad) + + def on_decode_pad_added(src, new_pad): + sink_pad = elems["convert1"].get_static_pad("sink") + if not sink_pad.is_linked(): + new_pad.link(sink_pad) + + elems["rtspsrc"].set_property("location", RTSP_URL) + elems["rtspsrc"].set_property("latency", 200) + elems["rtspsrc"].connect("pad-added", on_pad_added) + elems["decode"].connect("pad-added", on_decode_pad_added) + + + +def attach_buffer_probe(elems, buffer_probe): + srcpad = elems["capsfilter"].get_static_pad("src") + srcpad.add_probe(Gst.PadProbeType.BUFFER, buffer_probe) + +def setup_bus_and_loop(pipeline): + bus = pipeline.get_bus() + bus.add_signal_watch() + loop = GLib.MainLoop() + bus.connect("message", on_message, loop) + return loop + + +def main(): + os.makedirs(OUTPUT_DIR, exist_ok=True) + Gst.init(None) + + pipeline = Gst.Pipeline.new("simple-pipeline") + elems = create_elements() + link_pipeline_elements(pipeline, elems) + connect_dynamic_links(elems) + attach_buffer_probe(elems, buffer_probe) + + pipeline.add(elems["rtspsrc"]) + loop = setup_bus_and_loop(pipeline) + + print("Starting pipeline...") + pipeline.set_state(Gst.State.PLAYING) + + try: + loop.run() + except KeyboardInterrupt: + print("\nStopping...") + finally: + pipeline.set_state(Gst.State.NULL) + print("Pipeline stopped") + +if __name__ == "__main__": + main()