Skip to content

Python source element

This example creates a custom GStreamer source element in Python. The element is named simplepyimagesrc and generates raw RGB video frames without reading from a camera, file, or network stream.

The plugin is useful when you need a synthetic source for tests, demos, or for feeding generated frames into a normal GStreamer pipeline.

Full code

imagesrc.py 
 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
import gi

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

from gi.repository import GObject, Gst, GstBase

Gst.init(None)

WIDTH = 320
HEIGHT = 240
FPS = 30
CHANNELS = 3
FRAME_SIZE = WIDTH * HEIGHT * CHANNELS
DURATION = Gst.SECOND // FPS

CAPS = Gst.Caps.from_string(
    f"video/x-raw,format=RGB,width={WIDTH},height={HEIGHT},framerate={FPS}/1"
)


class SimplePyImageSrc(GstBase.BaseSrc):

    __gstmetadata__ = (
        "SimplePyImageSrc",
        "Source/Video",
        "Generates simple RGB image buffers",
        "Amir",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            CAPS,
        ),
    )

    def __init__(self):
        super().__init__()
        self.frame_number = 0
        self.set_format(Gst.Format.TIME)
        self.set_caps(CAPS)

    def do_start(self):
        self.frame_number = 0
        return True

    def do_is_seekable(self):
        return False

    def do_create(self, offset, size, buf):
        data = bytearray(FRAME_SIZE)
        shift = self.frame_number % 256

        for y in range(HEIGHT):
            for x in range(WIDTH):
                index = (y * WIDTH + x) * CHANNELS
                data[index] = (x + shift) % 256
                data[index + 1] = (y + shift) % 256
                data[index + 2] = shift

        buffer = Gst.Buffer.new_allocate(None, FRAME_SIZE, None)
        buffer.fill(0, data)
        buffer.pts = self.frame_number * DURATION
        buffer.dts = buffer.pts
        buffer.duration = DURATION
        buffer.offset = self.frame_number

        self.frame_number += 1

        return Gst.FlowReturn.OK, buffer


GObject.type_register(SimplePyImageSrc)

__gstelementfactory__ = (
    "simplepyimagesrc",
    Gst.Rank.NONE,
    SimplePyImageSrc,
)

Element type

SimplePyImageSrc inherits from GstBase.BaseSrc. A source element has only a source pad and produces buffers when the pipeline asks for more data.

1
class SimplePyImageSrc(GstBase.BaseSrc):

Use BaseSrc when the plugin creates data. Use BaseTransform when the plugin receives buffers, modifies or inspects them, and pushes them downstream.

Caps and timing

The constants at the top define the stream shape:

  • WIDTH, HEIGHT, and CHANNELS define the frame size.
  • FPS defines the frame rate.
  • FRAME_SIZE is the number of bytes in each RGB frame.
  • DURATION is the timestamp distance between two frames.

CAPS advertises the exact format the element produces:

1
2
3
CAPS = Gst.Caps.from_string(
    f"video/x-raw,format=RGB,width={WIDTH},height={HEIGHT},framerate={FPS}/1"
)

Downstream elements use these caps to know that the buffers are raw RGB video at 320x240 and 30/1 fps.

Pad template

The source pad template declares that this element always has one src pad and that the pad produces the caps defined above.

1
2
3
4
5
6
Gst.PadTemplate.new(
    "src",
    Gst.PadDirection.SRC,
    Gst.PadPresence.ALWAYS,
    CAPS,
)

This is the only pad template needed for a pure source element.

Startup

The constructor configures the source to use time-based offsets and sets the source caps.

1
2
self.set_format(Gst.Format.TIME)
self.set_caps(CAPS)

do_start() resets frame_number each time the element starts. That makes the first generated buffer start at timestamp 0.

The element is not seekable:

1
2
def do_is_seekable(self):
    return False

That is correct for a live-style generated stream that cannot jump to an arbitrary position.

Buffer creation

do_create() is the core method. GStreamer calls it whenever it needs the next buffer from the source.

The example fills a bytearray with an RGB gradient:

1
2
3
data[index] = (x + shift) % 256
data[index + 1] = (y + shift) % 256
data[index + 2] = shift

shift changes with frame_number, so each frame is slightly different.

Then the code allocates a Gst.Buffer, copies the image bytes into it, and sets the timing metadata:

1
2
3
4
buffer.pts = self.frame_number * DURATION
buffer.dts = buffer.pts
buffer.duration = DURATION
buffer.offset = self.frame_number

Correct timestamps are important because downstream elements, sinks, encoders, and muxers use them for playback timing and synchronization.

Finally, the method returns Gst.FlowReturn.OK and the new buffer:

1
return Gst.FlowReturn.OK, buffer

Register the plugin

The last lines register the Python class with GObject and expose it to GStreamer as an element factory named simplepyimagesrc.

1
2
3
4
5
6
GObject.type_register(SimplePyImageSrc)
__gstelementfactory__ = (
    "simplepyimagesrc",
    Gst.Rank.NONE,
    SimplePyImageSrc,
)

The factory name is the name used in gst-launch-1.0 pipelines.

Run

Put the plugin file where the Python plugin loader can discover it, then inspect the element:

1
GST_PLUGIN_PATH=$PWD gst-inspect-1.0 simplepyimagesrc

Run a preview pipeline:

1
GST_PLUGIN_PATH=$PWD gst-launch-1.0 simplepyimagesrc ! videoconvert ! autovideosink

For a headless test:

1
GST_PLUGIN_PATH=$PWD gst-launch-1.0 simplepyimagesrc num-buffers=30 ! fakesink