Access the raw frames

.gitignore

@@ -6,7 +6,7 @@ __pycache__/
 .Python
 
 # Outputs
-outputs/
+output/
 dataset/
 data/classification_dataset
 data/output_images/

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.

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
 

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()