How to make gstreamer buffer writable in Python?

      Want to modify Gst.Buffer in your Gstreamer application? Next information could be very useful. Estimated reading time: 10m, with code, no images:(

      Gstreamer in Python has some issues when it comes to advanced filters, plugins or behaviour. Most common issues connected to Gst.Buffer. 

#1. Problem

      Let’s create Gst.Buffer from simple string.

buffer = Gst.Buffer.new_wrapped(b"lifestyletransfer")

      In order to change something in buffer it’s required to map buffer to Gst.MapInfo with read/write flags.

ret, map_info = buffer.map(Gst.MapFlags.READ | Gst.MapFlags.WRITE)

      Data itself is accessible in data field.

print(map_info.data)
- b'lifestyletransfer'

     Now, let replace lower case ‘l‘ on upper case ‘L’:

map_info.data[0] = "L"

     But you’ll get and error:

'bytes' object does not support item assignment

     This is because PyGObject does not properly expose struct MapInfo. Look at this issue.

#2. Solution

     Instead of calling Gst.Buffer.map(), use ctypes to call gst_buffer_map() and gst_buffer_unmap(). The solution is taken from stb-tester’s open source projects (Github)

#1. Define Gst.MapInfo in proper way (using C struct):

from ctypes import *

class GstMapInfo(Structure):
    _fields_ = [("memory", c_void_p),        # GstMemory *memory
                ("flags", c_int),            # GstMapFlags flags
                ("data", POINTER(c_byte)),   # guint8 *data
                ("size", c_size_t),          # gsize size
                ("maxsize", c_size_t),       # gsize maxsize
                ("user_data", c_void_p * 4), # gpointer user_data[4]
                ("_gst_reserved", c_void_p * GST_PADDING)]

    Notes:

• GST_PADDING is default padding for structues (from gstconfig). (GST_PADDING = 4)
• To understand why used those fields check next example:

#define GST_MAP_INFO_INIT { NULL, # c_void_p                  
                            (GstMapFlags) 0, # c_int
                            NULL,# POINTER(c_byte)
                            0, # c_size_t
                            0, # c_size_t
                            {NULL, NULL, NULL, NULL}, # c_void_p * 4
                            {NULL, NULL, NULL, NULL}  # c_void_p * GST_PADDING }

#2. Load Gstreamer from shared library

libgst = ctypes.CDLL("libgstreamer-1.0.so.0")

# gst_buffer_map
libgst.gst_buffer_map.argtypes = [c_void_p, GST_MAP_INFO_POINTER, c_int]
libgst.gst_buffer_map.restype = c_int

# gst_buffer_unmap
libgst.gst_buffer_unmap.argtypes = [c_void_p, GST_MAP_INFO_POINTER]
libgst.gst_buffer_unmap.restype = None

# gst_mini_object_is_writable
libgst.gst_mini_object_is_writable.argtypes = [c_void_p]
libgst.gst_mini_object_is_writable.restype = c_int

@contextmanager
def map_gst_buffer(pbuffer, flags):
    if pbuffer is None:
        raise TypeError("Cannot pass NULL to map_gst_buffer")

    ptr = hash(pbuffer)
    if flags & Gst.MapFlags.WRITE and libgst.gst_mini_object_is_writable(ptr) == 0:
        raise ValueError("Writable array requested but buffer is not writeable")

    mapping = GstMapInfo()
    success = libgst.gst_buffer_map(ptr, mapping, flags)

    if not success:
        raise RuntimeError("Couldn't map buffer")

    try:
        # Cast POINTER(c_byte) to POINTER to array of c_byte with size mapping.size
        # Returns not pointer but the object to which pointer points
        yield cast(mapping.data, POINTER(c_byte * mapping.size)).contents
    finally:
        libgst.gst_buffer_unmap(ptr, mapping)

Notes:

• Implemented with @contextmanager in order to simplify function call using “with” statement

 #5. Run tests

with map_gst_buffer(buffer, Gst.MapFlags.READ | Gst.MapFlags.WRITE) as mapped:
    mapped[0] = ord('L')
    mapped[4] = ord('S')
    mapped[9] = ord('T')

      Result:

ret, map_info = buffer.map(Gst.MapFlags.READ)
print(map_info.data)
-------------------
b'LifeStyleTransfer'

      Hope it’s works for you too. Code is available here 🙂

Additional:

• This hack and some more interesting hacks are available here