Skip to content

Plugin metadata

Method Same Process Network Python Friendly
Custom GstMeta (c or rust) ✅ Best ❌ ⚠ Requires binding
GstCustomMeta ✅ Very Good ❌ ⚠ Limited GI support
Bus Message (GstStructure) ✅ ❌ ✅ Easy

GstStructure

plugin 
 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
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")

from gi.repository import Gst, GstBase

Gst.init(None)


class SimpleMeta(GstBase.BaseTransform):

    __gstmetadata__ = (
        "SimpleMeta",
        "Transform",
        "Posts metadata on the bus",
        "Amir"
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any()
        ),
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any()
        ),
    )

    def do_transform_ip(self, buf):

        s = Gst.Structure.new_empty("tracker")

        s.set_value("x", 100)
        s.set_value("y", 200)

        msg = Gst.Message.new_application(
            self,
            s
        )

        self.post_message(msg)

        return Gst.FlowReturn.OK


GObject = gi.repository.GObject

GObject.type_register(SimpleMeta)

__gstelementfactory__ = (
    "simplemeta",
    Gst.Rank.NONE,
    SimpleMeta
)

The simplemeta element posts a Gst.MessageType.APPLICATION message on the pipeline bus for each buffer. The message carries a GstStructure named tracker with the metadata fields x and y. The application reads messages from the bus, filters application messages, extracts the structure, and prints those values.

usage 
 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
import gi

gi.require_version("Gst", "1.0")
from gi.repository import Gst

Gst.init(None)

pipeline = Gst.parse_launch(
    "videotestsrc ! simplemeta ! fakesink"
)

bus = pipeline.get_bus()

pipeline.set_state(Gst.State.PLAYING)

while True:

    msg = bus.timed_pop(
        Gst.SECOND
    )

    if not msg:
        continue

    if msg.type == Gst.MessageType.APPLICATION:

        st = msg.get_structure()

        print(
            st.get_name(),
            st.get_value("x"),
            st.get_value("y")
        )

    elif msg.type == Gst.MessageType.EOS:
        break

pipeline.set_state(Gst.State.NULL)

GstCustomMeta

GstMeta is the base mechanism GStreamer uses to attach metadata directly to a Gst.Buffer. A custom GstMeta implementation defines its own metadata type, allocation logic, transform/copy behavior, and usually needs C or Rust code plus language bindings before Python can use it comfortably.

GstCustomMeta is a built-in GstMeta implementation whose payload is a named Gst.Structure. Instead of defining a new native metadata type, the plugin registers a custom metadata name and stores fields in that structure. This makes it useful for small, structured values that need to travel with the buffer inside the same process, while avoiding most of the boilerplate required by a full custom GstMeta.

The tradeoff is that GstCustomMeta is less flexible than a native custom GstMeta: the data model is limited to Gst.Structure fields, transform behavior is generic, and Python GI access can be limited depending on the GStreamer version and bindings.

plugin 
  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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
import ctypes
import ctypes.util

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")

from gi.repository import GObject, Gst, GstBase

Gst.init(None)

META_NAME = "simple-tracker-meta"
G_TYPE_INT = 24


class GValue(ctypes.Structure):
    _fields_ = (
        ("g_type", ctypes.c_size_t),
        ("data", ctypes.c_uint64 * 2),
    )


_gst = ctypes.CDLL(ctypes.util.find_library("gstreamer-1.0"))
_gobject = ctypes.CDLL(ctypes.util.find_library("gobject-2.0"))

_gst.gst_custom_meta_get_structure.argtypes = [ctypes.c_void_p]
_gst.gst_custom_meta_get_structure.restype = ctypes.c_void_p
_gst.gst_structure_set_value.argtypes = [
    ctypes.c_void_p,
    ctypes.c_char_p,
    ctypes.c_void_p,
]
_gst.gst_structure_set_value.restype = None

_gobject.g_value_init.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
_gobject.g_value_init.restype = ctypes.c_void_p
_gobject.g_value_set_int.argtypes = [ctypes.c_void_p, ctypes.c_int]
_gobject.g_value_set_int.restype = None
_gobject.g_value_unset.argtypes = [ctypes.c_void_p]
_gobject.g_value_unset.restype = None

Gst.Meta.register_custom_simple(META_NAME)


def set_meta_int(custom_meta, name, value):
    structure = _gst.gst_custom_meta_get_structure(hash(custom_meta))
    gvalue = GValue()

    _gobject.g_value_init(ctypes.byref(gvalue), G_TYPE_INT)
    _gobject.g_value_set_int(ctypes.byref(gvalue), value)
    _gst.gst_structure_set_value(
        structure,
        name.encode("utf-8"),
        ctypes.byref(gvalue),
    )
    _gobject.g_value_unset(ctypes.byref(gvalue))


class SimpleGstMeta(GstBase.BaseTransform):

    __gstmetadata__ = (
        "SimpleGstMeta",
        "Transform",
        "Attaches tracker metadata to buffers with GstMeta",
        "Amir",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
    )

    def do_transform_ip(self, buf):
        meta = buf.add_custom_meta(META_NAME)

        if meta is None:
            return Gst.FlowReturn.ERROR

        set_meta_int(meta, "x", 100)
        set_meta_int(meta, "y", 200)

        return Gst.FlowReturn.OK


GObject.type_register(SimpleGstMeta)

__gstelementfactory__ = (
    "simplegstmeta",
    Gst.Rank.NONE,
    SimpleGstMeta,
)
usage 
 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
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstApp", "1.0")

from gi.repository import Gst, GstApp

Gst.init(None)

META_NAME = "simple-tracker-meta"


def print_buffer_meta(buf):
    meta = buf.get_custom_meta(META_NAME)

    if meta is None:
        return

    st = meta.get_structure()

    print(
        st.get_name(),
        st.get_value("x"),
        st.get_value("y"),
    )


pipeline = Gst.parse_launch(
    "videotestsrc num-buffers=5 ! "
    "simplegstmeta ! "
    "appsink name=sink emit-signals=false sync=false"
)

sink = pipeline.get_by_name("sink")
bus = pipeline.get_bus()

pipeline.set_state(Gst.State.PLAYING)

while True:
    msg = bus.timed_pop_filtered(
        0,
        Gst.MessageType.ERROR | Gst.MessageType.EOS,
    )

    if msg is not None:
        if msg.type == Gst.MessageType.ERROR:
            err, debug = msg.parse_error()
            print(f"ERROR: {err}")
            print(f"DEBUG: {debug}")
            break

        if msg.type == Gst.MessageType.EOS:
            break

    sample = sink.try_pull_sample(100 * Gst.MSECOND)

    if sample is None:
        continue

    buf = sample.get_buffer()

    if buf is None:
        continue

    print_buffer_meta(buf)

pipeline.set_state(Gst.State.NULL)

Demo

  • Add another plugin that use the data
Plugin 
 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
import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstBase", "1.0")

from gi.repository import GObject, Gst, GstBase

Gst.init(None)

META_NAME = "simple-tracker-meta"


class UseGstMeta(GstBase.BaseTransform):

    __gstmetadata__ = (
        "UseGstMeta",
        "Transform",
        "Reads tracker metadata from buffers",
        "Amir",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
    )

    def do_transform_ip(self, buf):
        meta = buf.get_custom_meta(META_NAME)

        if meta is None:
            print("usegstmeta: no meta")
            return Gst.FlowReturn.OK

        st = meta.get_structure()
        x = st.get_value("x")
        y = st.get_value("y")
        total = x + y

        print(f"usegstmeta: x={x} y={y} total={total}")

        msg_struct = Gst.Structure.new_empty("used-tracker-meta")
        msg_struct.set_value("x", x)
        msg_struct.set_value("y", y)
        msg_struct.set_value("total", total)

        self.post_message(
            Gst.Message.new_application(self, msg_struct)
        )

        return Gst.FlowReturn.OK


GObject.type_register(UseGstMeta)

__gstelementfactory__ = (
    "usegstmeta",
    Gst.Rank.NONE,
    UseGstMeta,
)
usage 
 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
import gi

gi.require_version("Gst", "1.0")
from gi.repository import Gst

Gst.init(None)

pipeline = Gst.parse_launch(
    "videotestsrc num-buffers=5 ! "
    "simplegstmeta ! "
    "usegstmeta ! "
    "fakesink"
)

bus = pipeline.get_bus()

pipeline.set_state(Gst.State.PLAYING)

while True:
    msg = bus.timed_pop_filtered(
        Gst.SECOND,
        (
            Gst.MessageType.ERROR
            | Gst.MessageType.EOS
            | Gst.MessageType.APPLICATION
        ),
    )

    if msg is None:
        continue

    if msg.type == Gst.MessageType.ERROR:
        err, debug = msg.parse_error()
        print(f"ERROR: {err}")
        print(f"DEBUG: {debug}")
        break

    if msg.type == Gst.MessageType.EOS:
        break

    if msg.type == Gst.MessageType.APPLICATION:
        st = msg.get_structure()

        if st.get_name() != "used-tracker-meta":
            continue

        print(
            "bus:",
            st.get_value("x"),
            st.get_value("y"),
            st.get_value("total"),
        )

pipeline.set_state(Gst.State.NULL)

the python binding get the metadata from the usegstmeta plugin in that resend the data from simplegstmeta