Python send video to multiple sinks

Continuing the discussion from How to create an rtsp sink with deepstream python program?:

Basically I want to create two sinks, one to rtsp stream and one to local display. Maybe in future I’ll also create a third to save to mp4.

This links gives a good explanation on how to create multiple sinks in python through tee and multiple queues.

I found an example code in the python sample apps deepstream_test_4.py which uses tee to split the pipeline to multiple sinks, but am having difficulty in understanding how the tee is linked to the two queues. Can someone help explain the code clearly ?

The relevant code from the nvidia example app (https://github.com/NVIDIA-AI-IOT/deepstream_python_apps/blob/master/apps/deepstream-test4/deepstream_test_4.py) is here for reference.

tee=Gst.ElementFactory.make("tee", "nvsink-tee")
if not tee:
    sys.stderr.write(" Unable to create tee \n")

queue1=Gst.ElementFactory.make("queue", "nvtee-que1")
if not queue1:
    sys.stderr.write(" Unable to create queue1 \n")

queue2=Gst.ElementFactory.make("queue", "nvtee-que2")
if not queue2:
    sys.stderr.write(" Unable to create queue2 \n")

if (no_display) :
    print("Creating FakeSink \n")
    sink = Gst.ElementFactory.make("fakesink", "fakesink")
    if not sink:
        sys.stderr.write(" Unable to create fakesink \n")
else:
    if is_aarch64():
        transform = Gst.ElementFactory.make("nvegltransform", "nvegl-transform")

    print("Creating EGLSink \n")
    sink = Gst.ElementFactory.make("nveglglessink", "nvvideo-renderer")
    if not sink:
        sys.stderr.write(" Unable to create egl sink \n")

print("Playing file %s " %input_file)
source.set_property('location', input_file)
streammux.set_property('width', 1920)
streammux.set_property('height', 1080)
streammux.set_property('batch-size', 1)
streammux.set_property('batched-push-timeout', 4000000)
pgie.set_property('config-file-path', PGIE_CONFIG_FILE)
msgconv.set_property('config',MSCONV_CONFIG_FILE)
msgconv.set_property('payload-type', schema_type)
msgbroker.set_property('proto-lib', proto_lib)
msgbroker.set_property('conn-str', conn_str)
if cfg_file is not None:
    msgbroker.set_property('config', cfg_file)
if topic is not None:
    msgbroker.set_property('topic', topic)
msgbroker.set_property('sync', False)

print("Adding elements to Pipeline \n")
pipeline.add(source)
pipeline.add(h264parser)
pipeline.add(decoder)
pipeline.add(streammux)
pipeline.add(pgie)
pipeline.add(nvvidconv)
pipeline.add(nvosd)
pipeline.add(tee)
pipeline.add(queue1)
pipeline.add(queue2)
pipeline.add(msgconv)
pipeline.add(msgbroker)
pipeline.add(sink)
if is_aarch64() and not no_display:
    pipeline.add(transform)

print("Linking elements in the Pipeline \n")
source.link(h264parser)
h264parser.link(decoder)

sinkpad = streammux.get_request_pad("sink_0")
if not sinkpad:
    sys.stderr.write(" Unable to get the sink pad of streammux \n")
srcpad = decoder.get_static_pad("src")
if not srcpad:
    sys.stderr.write(" Unable to get source pad of decoder \n")
srcpad.link(sinkpad)

streammux.link(pgie)
pgie.link(nvvidconv)
nvvidconv.link(nvosd)
nvosd.link(tee)
queue1.link(msgconv)
msgconv.link(msgbroker)
if is_aarch64() and not no_display:
    queue2.link(transform)
    transform.link(sink)
else:
    queue2.link(sink)
sink_pad=queue1.get_static_pad("sink")
tee_msg_pad=tee.get_request_pad('src_%u')
tee_render_pad=tee.get_request_pad("src_%u")
if not tee_msg_pad or not tee_render_pad:
    sys.stderr.write("Unable to get request pads\n")
tee_msg_pad.link(sink_pad)
sink_pad=queue2.get_static_pad("sink")
tee_render_pad.link(sink_pad)

# create an event loop and feed gstreamer bus mesages to it
loop = GObject.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect ("message", bus_call, loop)

osdsinkpad = nvosd.get_static_pad("sink")
if not osdsinkpad:
    sys.stderr.write(" Unable to get sink pad of nvosd \n")

osdsinkpad.add_probe(Gst.PadProbeType.BUFFER, osd_sink_pad_buffer_probe, 0)

Hi,
The tee element basically duplicates incoming buffers to all of its src_pads and can create multiple sources on request:

tee

The example is requesting two pads with get_request_pad and linking them to two different branches. you can repeat this process to add another sink and end up with something like this:

In that example, we use GstInterpipe because it helps isolate each sub-pipeline to handle stop, play, and errors individually but it is the same idea as using a tee.

2 Likes

Thank you Miguel, this is an excellent answer and links.

Great sharing, @miguel.taylor!

Thanks!

@tttuser Were you able to do it? I am trying to do something similar but instead of local display, I am trying to one queue for msgconv and other for rtsp sink.

print(“Adding elements to Pipeline \n”)
pipeline.add(pgie)
pipeline.add(tracker)
pipeline.add(tiler)
pipeline.add(nvvidconv)
pipeline.add(nvosd)

pipeline.add(tee)
pipeline.add(queue1)
pipeline.add(queue2)
pipeline.add(msgconv)
pipeline.add(msgbroker)

pipeline.add(nvvidconv_postosd)
pipeline.add(caps)
pipeline.add(encoder)
pipeline.add(rtppay)
if is_aarch64():
    pipeline.add(transform)
pipeline.add(sink)
pipeline.add(fakesink)

print("Linking elements in the Pipeline \n")
streammux.link(pgie)
pgie.link(tracker)
tracker.link(tiler)
tiler.link(nvvidconv)
nvvidconv.link(nvosd)
nvosd.link(tee)

queue1.link(msgconv)
msgconv.link(msgbroker)
queue2.link(nvvidconv_postosd)    
nvvidconv_postosd.link(caps)
caps.link(encoder)
encoder.link(rtppay)
rtppay.link(sink)
if is_aarch64():
    rtppay.link(transform)
    transform.link(sink)
else:
    rtppay.link(sink)  


tee_msg_pad = tee.get_request_pad('src_%u')
tee_render_pad = tee.get_request_pad("src_%u")
if not tee_msg_pad or not tee_render_pad :
    sys.stderr.write("Unable to get request pads \n")

msg_sink_pad = queue1.get_static_pad("sink")
tee_msg_pad.link(msg_sink_pad)
vid_sink_pad = queue2.get_static_pad("sink")
tee_render_pad.link(vid_sink_pad)