Skip to content

In Python GStreamer bindings, emit-signals is a boolean property that tells elements such as appsink to emit high-level GObject signals instead of requiring the application to poll for data. For appsink, enabling this property lets you handle buffers through signals like new-sample.

AppSink callback

appsink can emit a new-sample signal whenever a buffer is ready. In Python, connect that signal to a callback and pull the sample from the sink inside the callback.

The pipeline must set emit-signals=true on appsink; otherwise appsink will not emit new-sample, and the callback will not run.

Callback example 
 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
import gi

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

Gst.init(None)

pipeline = Gst.parse_launch(
    "videotestsrc ! videoconvert ! appsink name=sink emit-signals=true"
)

appsink = pipeline.get_by_name("sink")


def on_new_sample(sink):
    sample = sink.emit("pull-sample")
    print("Frame")
    return Gst.FlowReturn.OK


appsink.connect("new-sample", on_new_sample)

pipeline.set_state(Gst.State.PLAYING)

GLib.MainLoop().run()

AppSink pull mode

In pull mode, the application asks appsink for samples directly. Keep emit-signals=false; no new-sample callback is used.

pull-sample is the blocking version. It has no timeout: the call waits until a sample is available or until the stream reaches EOS. If it returns None, the stream is done or the sink is shutting down, so the loop should stop instead of continuing to wait for another frame.

Blocking pull example 
 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
import gi

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

Gst.init(None)

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

appsink = pipeline.get_by_name("sink")

pipeline.set_state(Gst.State.PLAYING)

while True:
    sample = appsink.emit("pull-sample")
    if sample is None:
        # pull-sample blocks until a sample is ready. None means EOS or shutdown,
        # not "no sample yet", so continuing would loop after the stream ended.
        break

    buffer = sample.get_buffer()
    print("Frame pts:", buffer.pts)

pipeline.set_state(Gst.State.NULL)

try-pull-sample is the timeout-based style. It waits only for the timeout you pass, returns None when no sample is ready yet, and lets the application do other work or check the bus before trying again. In that case, continue is the right behavior unless the bus reports EOS or an error.

Non-blocking pull example 
 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
import gi

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

Gst.init(None)

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

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

pipeline.set_state(Gst.State.PLAYING)

while True:
    sample = appsink.emit("try-pull-sample", 100 * Gst.MSECOND)
    if sample is not None:
        buffer = sample.get_buffer()
        print("Frame pts:", buffer.pts)
        continue

    message = bus.timed_pop_filtered(
        0,
        Gst.MessageType.ERROR | Gst.MessageType.EOS,
    )
    if message is None:
        print("No sample ready yet")
        continue

    # Stop only when the pipeline reports EOS or ERROR on the bus.
    if message.type == Gst.MessageType.ERROR:
        error, debug = message.parse_error()
        print("Error:", error, debug)

    break

pipeline.set_state(Gst.State.NULL)