How te retrieve stream statistics in Gstreamer?

Question

First of all I would like to start by saying that I'm really new to Gstreamer and its capabilities so pardon my ignorance if my understanding or implementation was wrong, I am still learning .

I would like to build a small streaming application in PYTHON or JAVA (as I am not proficient in C) with QOS integrated in it, especially for the package drop count statistics, and RTP and RTCP seem the perfect match.

For this purpose I implemented a SERVER

#! /usr/bin/env python
import gi
import sys
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst

#gst-launch -v rtpbin name=rtpbin audiotestsrc ! audioconvert ! alawenc ! rtppcmapay ! rtpbin.send_rtp_sink_0 \
#                rtpbin.send_rtp_src_0 ! udpsink port=10000 host=xxx.xxx.xxx.xxx \
#                rtpbin.send_rtcp_src_0 ! udpsink port=10001 host=xxx.xxx.xxx.xxx sync=false async=false \
#                udpsrc port=10002 ! rtpbin.recv_rtcp_sink_0

DEST_HOST = '127.0.0.1'

AUDIO_SRC = 'audiotestsrc'
AUDIO_ENC = 'alawenc'
AUDIO_PAY = 'rtppcmapay'

RTP_SEND_PORT = 5002
RTCP_SEND_PORT = 5003
RTCP_RECV_PORT = 5007

GObject.threads_init()
Gst.init(sys.argv)

# the pipeline to hold everything
pipeline = Gst.Pipeline.new('rtp_server')

# the pipeline to hold everything
audiosrc = Gst.ElementFactory.make(AUDIO_SRC, 'audiosrc')
audioconv = Gst.ElementFactory.make('audioconvert', 'audioconv')
audiores = Gst.ElementFactory.make('audioresample', 'audiores')

# the pipeline to hold everything
audioenc = Gst.ElementFactory.make(AUDIO_ENC, 'audioenc')
audiopay = Gst.ElementFactory.make(AUDIO_PAY, 'audiopay')

# add capture and payloading to the pipeline and link
pipeline.add(audiosrc)
pipeline.add(audioconv)
pipeline.add(audiores)
pipeline.add(audioenc)
pipeline.add(audiopay)


audiosrc.link(audioconv)
audioconv.link(audiores)
audiores.link(audioenc)
audioenc.link(audiopay)

# the rtpbin element
rtpbin = Gst.ElementFactory.make('rtpbin', 'rtpbin')

pipeline.add(rtpbin)

# the udp sinks and source we will use for RTP and RTCP
rtpsink = Gst.ElementFactory.make('udpsink', 'rtpsink')
rtpsink.set_property('port', RTP_SEND_PORT)
rtpsink.set_property('host', DEST_HOST)

rtcpsink = Gst.ElementFactory.make('udpsink', 'rtcpsink')
rtcpsink.set_property('port', RTCP_SEND_PORT)
rtcpsink.set_property('host', DEST_HOST)
# no need for synchronisation or preroll on the RTCP sink
rtcpsink.set_property('async', False)
rtcpsink.set_property('sync', False)

rtcpsrc = Gst.ElementFactory.make('udpsrc', 'rtcpsrc')
rtcpsrc.set_property('port', RTCP_RECV_PORT)

pipeline.add(rtpsink)
pipeline.add(rtcpsink)
pipeline.add(rtcpsrc)

# now link all to the rtpbin, start by getting an RTP sinkpad for session 0
sinkpad = Gst.Element.get_request_pad(rtpbin, 'send_rtp_sink_0')
srcpad = Gst.Element.get_static_pad(audiopay, 'src')
lres = Gst.Pad.link(srcpad, sinkpad)

# get the RTP srcpad that was created when we requested the sinkpad above and
# link it to the rtpsink sinkpad
srcpad = Gst.Element.get_static_pad(rtpbin, 'send_rtp_src_0')
sinkpad = Gst.Element.get_static_pad(rtpsink, 'sink')
lres = Gst.Pad.link(srcpad, sinkpad)

# get an RTCP srcpad for sending RTCP to the receiver
srcpad = Gst.Element.get_request_pad(rtpbin, 'send_rtcp_src_0')
sinkpad = Gst.Element.get_static_pad(rtcpsink, 'sink')
lres = Gst.Pad.link(srcpad, sinkpad)

# we also want to receive RTCP, request an RTCP sinkpad for session 0 and
# link it to the srcpad of the udpsrc for RTCP
srcpad = Gst.Element.get_static_pad(rtcpsrc, 'src')
sinkpad = Gst.Element.get_request_pad(rtpbin, 'recv_rtcp_sink_0')
lres = Gst.Pad.link(srcpad, sinkpad)

# set the pipeline to playing
Gst.Element.set_state(pipeline, Gst.State.PLAYING)

# we need to run a GLib main loop to get the messages
mainloop = GObject.MainLoop()
mainloop.run()
Gst.Element.set_state(pipeline, Gst.State.NULL)

and a Client

#! /usr/bin/env python

import gi
import sys
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst

#
# A simple RTP receiver
#
#  receives alaw encoded RTP audio on port 5002, RTCP is received on  port 5003.
#  the receiver RTCP reports are sent to port 5007
#
#             .-------.      .----------.     .---------.   .-------.   .--------.
#  RTP        |udpsrc |      | rtpbin   |     |pcmadepay|   |alawdec|   |alsasink|
#  port=5002  |      src->recv_rtp recv_rtp->sink     src->sink   src->sink      |
#             '-------'      |          |     '---------'   '-------'   '--------'
#                            |          |
#                            |          |     .-------.
#                            |          |     |udpsink|  RTCP
#                            |    send_rtcp->sink     | port=5007
#             .-------.      |          |     '-------' sync=false
#  RTCP       |udpsrc |      |          |               async=false
#  port=5003  |     src->recv_rtcp      |
#             '-------'      '----------'

AUDIO_CAPS = 'application/x-rtp,media=(string)audio,clock-rate=(int)8000,encoding-name=(string)PCMA'
AUDIO_DEPAY = 'rtppcmadepay'
AUDIO_DEC = 'alawdec'
AUDIO_SINK = 'autoaudiosink'

DEST = '127.0.0.1'

RTP_RECV_PORT = 5002
RTCP_RECV_PORT = 5003
RTCP_SEND_PORT = 5007

GObject.threads_init()
Gst.init(sys.argv)

#gst-launch -v rtpbin name=rtpbin                                                \
#       udpsrc caps=$AUDIO_CAPS port=$RTP_RECV_PORT ! rtpbin.recv_rtp_sink_0              \
#             rtpbin. ! rtppcmadepay ! alawdec ! audioconvert ! audioresample ! autoaudiosink \
#           udpsrc port=$RTCP_RECV_PORT ! rtpbin.recv_rtcp_sink_0                              \
#         rtpbin.send_rtcp_src_0 ! udpsink port=$RTCP_SEND_PORT host=$DEST sync=false async=false

def pad_added_cb(rtpbin, new_pad, depay):
    sinkpad = Gst.Element.get_static_pad(depay, 'sink')
    lres = Gst.Pad.link(new_pad, sinkpad)

# the pipeline to hold eveything
pipeline = Gst.Pipeline.new('rtp_client')

# the udp src and source we will use for RTP and RTCP
rtpsrc = Gst.ElementFactory.make('udpsrc', 'rtpsrc')
rtpsrc.set_property('port', RTP_RECV_PORT)

# we need to set caps on the udpsrc for the RTP data
caps = Gst.caps_from_string(AUDIO_CAPS)
rtpsrc.set_property('caps', caps)

rtcpsrc = Gst.ElementFactory.make('udpsrc', 'rtcpsrc')
rtcpsrc.set_property('port', RTCP_RECV_PORT)

rtcpsink = Gst.ElementFactory.make('udpsink', 'rtcpsink')
rtcpsink.set_property('port', RTCP_SEND_PORT)
rtcpsink.set_property('host', DEST)

# no need for synchronisation or preroll on the RTCP sink
rtcpsink.set_property('async', False)
rtcpsink.set_property('sync', False)

pipeline.add(rtpsrc)
pipeline.add(rtcpsrc)
pipeline.add(rtcpsink)

# the depayloading and decoding
audiodepay = Gst.ElementFactory.make(AUDIO_DEPAY, 'audiodepay')
audiodec = Gst.ElementFactory.make(AUDIO_DEC, 'audiodec')

# the audio playback and format conversion
audioconv = Gst.ElementFactory.make('audioconvert', 'audioconv')
audiores = Gst.ElementFactory.make('audioresample', 'audiores')
audiosink = Gst.ElementFactory.make(AUDIO_SINK, 'audiosink')

# add depayloading and playback to the pipeline and link
pipeline.add(audiodepay)
pipeline.add(audiodec)
pipeline.add(audioconv)
pipeline.add(audiores)
pipeline.add(audiosink)


audiodepay.link(audiodec)
audiodec.link(audioconv)
audioconv.link(audiores)
audiores.link(audiosink)

# the rtpbin element
rtpbin = Gst.ElementFactory.make('rtpbin', 'rtpbin')

pipeline.add(rtpbin)

# now link all to the rtpbin, start by getting an RTP sinkpad for session 0
srcpad = Gst.Element.get_static_pad(rtpsrc, 'src')
sinkpad = Gst.Element.get_request_pad(rtpbin, 'recv_rtp_sink_0')
lres = Gst.Pad.link(srcpad, sinkpad)

# get an RTCP sinkpad in session 0
srcpad = Gst.Element.get_static_pad(rtcpsrc, 'src')
sinkpad = Gst.Element.get_request_pad(rtpbin, 'recv_rtcp_sink_0')
lres = Gst.Pad.link(srcpad, sinkpad)

# get an RTCP srcpad for sending RTCP back to the sender
srcpad = Gst.Element.get_request_pad(rtpbin, 'send_rtcp_src_0')
sinkpad = Gst.Element.get_static_pad(rtcpsink, 'sink')
lres = Gst.Pad.link(srcpad, sinkpad)

rtpbin.connect('pad-added', pad_added_cb, audiodepay)



# def newManager():
    # rtpbin.connect('on-ssrc-active', onSSRCActive)

# def onSSRCActive(self):
    # print (self)

# rtpsrc.connect('new-manager', newManager)

Gst.Element.set_state(pipeline, Gst.State.PLAYING)

mainloop = GObject.MainLoop()
mainloop.run()

Gst.Element.set_state(pipeline, Gst.State.NULL)

While the server and client works without problems I cannot find a way or information on how to retrieve the dropped package statistics from RTCP for this RTP stream even if it is in command line.

And any help in that regard would be appreciated!

Ensoll

The pipeline has gstbus to which you normally connect with gst_bus_add_signal_watch () this allows you to connect messages that you are interested in watching for , to a callback. see: gstreamer.freedesktop.org/data/doc/gstreamer/head/gstreamer/… and gstreamer.freedesktop.org/documentation/application-development/… – Roi

@RolfofSaxony thank you for your answer! I will have a look at it right away, also please don't forget to write this comment as an answer as it may help someone else in the future :) – Ensoll

Answer 1

Here is a mock up, on Linux, just to illustrate a method of listening to the bus messages.
I haven't the time to sort out the QOS Gst.Message, so I hacked something in there to force a QOS, so that you at least see it.
Hopefully this will be enough to get you going in the right direction.

#!/usr/bin/python3

from os import path
import time
import gi
gi.require_version('Gst', '1.0')
gi.require_version('Gtk', '3.0')
gi.require_version('GstVideo', '1.0')
from gi.repository import GObject, Gst, Gtk
from gi.repository import GdkX11, GstVideo


GObject.threads_init()
Gst.init(None)
filename = path.join(path.dirname(path.abspath(__file__)), '../H.mkv')
uri = 'file://' + filename


class Player(object):
    def __init__(self):
        self.window = Gtk.Window()
        self.window.connect('destroy', self.quit)
        self.window.set_default_size(800, 450)

        self.drawingarea = Gtk.DrawingArea()
        self.window.add(self.drawingarea)

        # Create GStreamer pipeline
        self.pipeline = Gst.Pipeline()

        # Create bus to get events from GStreamer pipeline
        self.bus = self.pipeline.get_bus()
        self.bus.add_signal_watch()
        self.bus.connect('message::eos', self.on_eos)
        self.bus.connect('message::error', self.on_error)
        self.bus.connect('message::qos', self.on_quality_of_service)

        self.bus.enable_sync_message_emission()
        self.bus.connect('sync-message::element', self.on_sync_message)

        # Create GStreamer elements
        self.playbin = Gst.ElementFactory.make('playbin', None)

        # Add playbin to the pipeline
        self.pipeline.add(self.playbin)

        # Set properties
        self.playbin.set_property('uri', uri)

    def run(self):
        self.window.show_all()
        self.xid = self.drawingarea.get_property('window').get_xid()
        self.pipeline.set_state(Gst.State.PLAYING)
        time.sleep(2)
        self.bus.emit('message::qos',Gst.Message('xxxxx'))
        Gtk.main()

    def quit(self, window):
        self.pipeline.set_state(Gst.State.NULL)
        Gtk.main_quit()

    def on_sync_message(self, bus, msg):
        if msg.get_structure().get_name() == 'prepare-window-handle':
            print('prepare-window-handle')
            msg.src.set_window_handle(self.xid)

    def on_eos(self, bus, msg):
        print('End of Service')

    def on_error(self, bus, msg):
        print('Error', msg.parse_error())

    def on_quality_of_service(self, bus, msg):
        print('Qos Message:', msg.parse_qos())

p = Player()
p.run()

Roi

Thank you! This is exactly what i needed :) – Ensoll