measuring FPS of a source live stream (using identity, elements)

[Marianna Smidth Buschle]

Hi,

I have made some python wrappers for easier building gstreamer applications.
And in there I have added some debug/performance measure tools.
I have done that using buffer probes.

I think your problem is that the buffers you get from 'rtspsrc' are not 
necessarily frames (I guess depending on format, packet sizes, ...).
So I would recommend you measure after 'decodebin', there you should get 
frames.

This is how I measured "performance" using buffer probes on specific 
elements:

defmeasure_elem_stats(self, elem: Gst.Element) -> None:
"""Measure bitrate, fps and latency of gstreamer element"""
stats= GstPipeline.GstElementStats(elem)
add_buffer_probe(elem, "src", self._calc_stats, stats)
add_buffer_probe(elem, "sink", self._probe_in, stats)
defadd_buffer_probe(
elem: Gst.Element,
pad_name: str,
func,
data: Optional[object] = None,
) -> bool:
"""Add a buffer probe to pad of element
:param elem: element to probe
:param pad_name: name of the pad to probe, normally 'sink' or 'src'
:param func: callback function called when a buffer is received on the pad
:param data: optional data to pass to probe function, defaults to None
:return: True if the probe was sucessfully installed,
False otherwise.
"""
ifnotelem:
returnFalse
pad= elem.get_static_pad(pad_name)
ifnotpad:
# print(f"Couldn't find {elem.name}::{pad_name}")
returnFalse
pad.add_probe(Gst.PadProbeType.BUFFER, func, data)
returnTrue
def_calc_stats(
self, pad: Gst.Pad, info: Gst.PadProbeInfo, stats: GstElementStats
) -> Gst.PadProbeReturn:
"""Callback of buffer probe for measuring bitrate, fps and latency"""
buf= info.get_buffer()
time= self.get_pipe_timestamp()
diff= time- buf.pts # total latency from src
elapsed= time- stats.perf_sink.prev_time
ifnotis_clock_valid(stats.perf_sink.prev_time):
stats.perf_sink.prev_time= time
elifis_clock_valid(time) andelapsed> Gst.SECOND:
# frames per second
factor_n= elapsed/ Gst.MSECOND
factor_d= Gst.SECOND / Gst.MSECOND
# bytes per second
bps= stats.perf_sink.bps* factor_d/ factor_n
stats.bitrate= bps/ (1024* 1024/ 8) # Mbps
stats.fps= stats.perf_sink.frame_count* factor_d/ factor_n
latency= stats.perf_sink.latency/ stats.perf_sink.frame_count
stats.elem_latency= latency- stats.src_latency
self.__logger.debug(
f"{stats.elem_name}: "
f"bitrate={stats.bitrate}Mbps "
f"fps={stats.fps}"
f"latency elem={timestamp_msecs(stats.elem_latency)}ms "
f"total={timestamp_msecs(latency)}ms"
)
self._plot_stats(time, stats)
stats.perf_sink.reset(time)
stats.perf_sink.frame_count+= 1
stats.perf_sink.bps+= buf.get_size()
stats.perf_sink.latency+= diff
returnGst.PadProbeReturn.OK
def_probe_in(
self, pad: Gst.Pad, info: Gst.PadProbeInfo, stats: GstElementStats
) -> Gst.PadProbeReturn:
"""Callback of buffer probe for measuring processing time"""
buf= info.get_buffer()
time= self.get_pipe_timestamp()
diff= time- buf.pts # latency from src
elapsed= time- stats.perf_src.prev_time
ifnotis_clock_valid(stats.perf_src.prev_time):
stats.perf_src.prev_time= time
elifis_clock_valid(time) andelapsed> Gst.SECOND:
stats.src_latency= stats.perf_src.latency/ stats.perf_src.frame_count
# self.__logger.debug(
# f"{stats.elem_name}: src_latency={latency / Gst.MSECOND} ms"
# )
stats.perf_src.reset(time)
stats.perf_src.frame_count+= 1
stats.perf_src.latency+= diff
returnGst.PadProbeReturn.OK
classGstElementStats:
"""Statistic values for Gst.Element
:param elem: gstreamer element to measure statistics on
"""
classPerfData:
"""Used for measuring bitrate, fps and latencies in buffer probes"""
def__init__(self):
"""Constructor"""
self.prev_time= Gst.CLOCK_TIME_NONE
self.frame_count= 0
self.bps= 0
self.latency= 0
defreset(self, time: int) -> None:
"""Reset values"""
self.frame_count= 0
self.bps= 0
self.latency= 0
self.prev_time= time
def__init__(self, elem: Gst.Element):
"""Constructor"""
self.bitrate= 0# in Mbps
self.fps= 0
self.src_latency= 0# in ms
self.elem_latency= 0# in ms
self.elem_name= elem.name
self.perf_src= GstPipeline.GstElementStats.PerfData()
self.perf_sink= GstPipeline.GstElementStats.PerfData()


Otherwise it is also possible to use GST_TRACERS (either the standard 
gstreamer ones or possible also the ones from gst-shark from ridgeRun) 
and get the values inside your application using a custom log function:

Gstreamer Tracers can be enabled by setting them through the correct
environment variable (`GST_TRACERS`).
See https://gstreamer.freedesktop.org/documentation/additional/design/
tracing.html?gi-language=python#latency
Latency tracer:
GST_TRACERS="latency(flags=pipeline+element+reported)" python3 pipeline.py
CPU usage tracer:
GST_TRACERS="stats;rusage" python3 pipeline.py
# tracer: must be set before 'Gst.init()': no runtime control
# Example tracers
LATENCY_TRACER= "latency(flags=pipeline+element+reported)"
CPU_USAGE_TRACER= "stats;rusage"
DESIRED_TRACERS= None
ifnotos.getenv("GST_TRACERS") andDESIRED_TRACERS:
os.environ["GST_TRACERS"] = DESIRED_TRACERS
def_on_debug(
self,
category: Gst.DebugCategory,
level: Gst.DebugLevel,
file: str,
function: str,
line: int,
source: GObject.Object,
message: Gst.DebugMessage,
user_data: Optional[object] = None,
):
"""Custom gstreamer log function"""
level_name= Gst.DebugLevel.get_name(level)
text= f"Gstreamer {level_name}{category.get_name()}"
ifsource:
file_name= os.path.basename(file)
text+= f"{file_name}:{line}:{function}<{source.name}>"
text+= f": {message.get()}"
# currently ignoring gst debug log level below DEBUG
iflevel== Gst.DebugLevel.ERROR:
self.__logger.error(text)
eliflevel== Gst.DebugLevel.WARNING:
self.__logger.warning(text)
eliflevel== Gst.DebugLevel.INFO:
self.__logger.info(text)
eliflevel== Gst.DebugLevel.DEBUG:
self.__logger.debug(text)
# handle GST_TRACERS
eliflevel== Gst.DebugLevel.TRACE:
# self.__logger.debug(text)
self._handle_tracer_output(category, message)
def_handle_tracer_output(
self, category: Gst.DebugCategory, message: Gst.DebugMessage
) -> None:
"""Handle the output generated by Gstreamer Tracers
Detect the type of tracer and generate the relevant internal Event
"""
ifcategory.get_name() == "GST_TRACER":
tracer= self._parse_trace_msg(message.get())
# 'Gst.Structure.from_string' fails for some types of tracer messages
iftracer== "message"ortracer== "query":
# self.__logger.debug(f"{category.get_name()}: {message.get()}")
return
try:
s, _= Gst.Structure.from_string(message.get())
# tracer = s.get_name()
exceptException:
self.__logger.warning("Failed parsing structure")
self.__logger.debug(f"{category.get_name()}: {message.get()}")
return
iftracer== "element-latency":
type= GstPipeline.GstEvent.GstTracerEventType.ELEM_LATENCY
self._plot_elem_latency(s)
eliftracer== "latency":
type= GstPipeline.GstEvent.GstTracerEventType.LATENCY
eliftracer== "thread-rusage":
# self._plot_rusage(s)
type= GstPipeline.GstEvent.GstTracerEventType.THREAD_RUSAGE
eliftracer== "proc-rusage":
# self._plot_rusage(s)
type= GstPipeline.GstEvent.GstTracerEventType.PROC_RUSAGE
else:
type= GstPipeline.GstEvent.GstTracerEventType.TRACER
self.queue_event(
GstPipeline.GstEvent(
type,
f"{s.to_string()}",
)
)
def_config_gst_log(self) -> None:
"""Configure custom gstreamer logging function
If 'GST_DEBUG' is set though the command line the custom 'on_debug()'
will be ignored and the default gstreamer logging will be used instead.
Also enable the gstreamer tracing system (when required).
"""
ifnotGst.debug_is_active():
Gst.debug_remove_log_function(None)
Gst.debug_add_log_function(self._on_debug, None)
level= Gst.debug_get_default_threshold()
iflevel< Gst.DebugLevel.ERROR:
Gst.debug_set_default_threshold(Gst.DebugLevel.WARNING)
# enable tracer
ifself.DESIRED_TRACERS:
Gst.debug_set_threshold_for_name("GST_TRACER", Gst.DebugLevel.TRACE)
Gst.debug_set_active(True)

Best Regards

Marianna

On 17.03.2023 05.04, gstreamer-devel-request at lists.freedesktop.org wrote:
> hello all,
>
> i am attempting to measure FPS of a source stream (e.g. rtsp or hls). I am
> currently using an identity element immediately after the source element
> (e.g. rtspsrc -> identity -> decodebin -> fakesink). Using the handoff
> signal on identity, I am able to get a good calculation of buffers per
> second.
>
> However, it seems that this value is not always 1:1 with frames per second.
> It very frequently is on stable streams (e.g. I often see 29-31 buffers/sec
> on a 30fps stream). However, on streams with intermittent connectivity, it
> doesn't seem to tell the whole story.
>
> I've seen fpsdisplaysink which actually works really well (I am able to see
> "real" fps values that are dynamically changing as the pipeline plays).
> However, a) I don't want to measure the fps at the sink (i have a tee'd
> pipeline with multiple sinks and more interesting in measuring source fps
> than that of my pipeline) and b) this element seems to just overlay it ?
> ideally i would be able to get the average "recent" fps as a property.
>
> I've seen ridgerun's gst-perf element, but that only seems to log values.
>
> Is there a recommended approach to monitoring "real" FPS of an incoming
> live stream? Ideally using identity, but open to other options.
>
> Thanks,
> Matt

-- 
Best regards / Med venlig hilsen
“Marianna Smidth Buschle”
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.freedesktop.org/archives/gstreamer-devel/attachments/20230317/cabf604a/attachment-0001.htm>