Merge branch 'mpegtsbase-stats' into 'main'

Draft: mpegtsdemux: Add stats to mpegtsbase

See merge request gstreamer/gstreamer!3455

subprojects/gst-plugins-bad/gst/mpegtsdemux/mpegtsbase.c

@@ -60,12 +60,15 @@ static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
     );
 
 #define DEFAULT_IGNORE_PCR FALSE
+#define DEFAULT_ENABLE_STATS FALSE
 
 enum
 {
   PROP_0,
   PROP_PARSE_PRIVATE_SECTIONS,
   PROP_IGNORE_PCR,
+  PROP_ENABLE_STATS,
+  PROP_STATS,
   /* FILL ME */
 };
 
@@ -98,6 +101,7 @@ static gboolean mpegts_base_parse_atsc_mgt (MpegTSBase * base,
     GstMpegtsSection * section);
 static void remove_each_program (MpegTSBaseProgram * program,
     MpegTSBase * base);
+static GstStructure *mpegts_base_get_stats (MpegTSBase * base);
 
 static void
 _extra_init (void)
@@ -161,6 +165,25 @@ mpegts_base_class_init (MpegTSBaseClass * klass)
           "Ignore PCR stream for timing", DEFAULT_IGNORE_PCR,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  /**
+   * GstMpegtsBase:enable-stats:
+   *
+   * Collect stats to be retrieved via the stats property.
+   */
+  g_object_class_install_property (gobject_class, PROP_ENABLE_STATS,
+      g_param_spec_boolean ("enable-stats", "Enable statistics",
+          "Enable MPEG-TS statistics collection", DEFAULT_ENABLE_STATS,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstMpegtsBase:stats:
+   *
+   * Retrieve statistics about the MPEG-TS stream, if enable-stats is TRUE.
+   */
+  g_object_class_install_property (gobject_class, PROP_STATS,
+      g_param_spec_boxed ("stats", "Statistics", "MPEG-TS statistics",
+          GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
+
   klass->sink_query = GST_DEBUG_FUNCPTR (mpegts_base_default_sink_query);
   klass->handle_psi = NULL;
 
@@ -180,6 +203,11 @@ mpegts_base_set_property (GObject * object, guint prop_id,
     case PROP_IGNORE_PCR:
       base->ignore_pcr = g_value_get_boolean (value);
       break;
+    case PROP_ENABLE_STATS:
+      base->enable_stats = g_value_get_boolean (value);
+      mpegts_packetizer_set_packet_counts_enabled (base->packetizer,
+          base->enable_stats);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
   }
@@ -198,6 +226,12 @@ mpegts_base_get_property (GObject * object, guint prop_id,
     case PROP_IGNORE_PCR:
       g_value_set_boolean (value, base->ignore_pcr);
       break;
+    case PROP_ENABLE_STATS:
+      g_value_set_boolean (value, base->enable_stats);
+      break;
+    case PROP_STATS:
+      g_value_take_boxed (value, mpegts_base_get_stats (base));
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
   }
@@ -282,6 +316,8 @@ mpegts_base_init (MpegTSBase * base)
   base->push_section = TRUE;
   base->ignore_pcr = DEFAULT_IGNORE_PCR;
 
+  base->enable_stats = DEFAULT_ENABLE_STATS;
+
   mpegts_base_reset (base);
 }
 
@@ -1405,6 +1441,26 @@ remove_each_program (MpegTSBaseProgram * program, MpegTSBase * base)
   mpegts_base_deactivate_program (base, program);
 }
 
+static GstStructure *
+mpegts_base_get_stats (MpegTSBase * base)
+{
+  GstStructure *s;
+
+  s = gst_structure_new_empty ("application/x-gst-mpegts-stats");
+
+  if (base->enable_stats) {
+    GstStructure *counts;
+    GValue counts_v = G_VALUE_INIT;
+
+    counts = mpegts_packetizer_get_packet_counts (base->packetizer);
+    g_value_init (&counts_v, GST_TYPE_STRUCTURE);
+    g_value_take_boxed (&counts_v, counts);
+    gst_structure_take_value (s, "packet-counts", &counts_v);
+  }
+
+  return s;
+}
+
 static inline GstFlowReturn
 mpegts_base_drain (MpegTSBase * base)
 {

subprojects/gst-plugins-bad/gst/mpegtsdemux/mpegtsbase.h

@@ -176,6 +176,9 @@ struct _MpegTSBase {
 
   /* Used for delayed seek events */
   GstEvent *seek_event;
+
+  /* Collect statistics */
+  gboolean enable_stats;
 };
 
 struct _MpegTSBaseClass {

subprojects/gst-plugins-bad/gst/mpegtsdemux/mpegtspacketizer.c

@@ -58,6 +58,8 @@ static GstClockTime calculate_skew (MpegTSPacketizer2 * packetizer,
 static void _close_current_group (MpegTSPCR * pcrtable);
 static void record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
     guint64 pcr, guint64 offset);
+static void mpegts_packetizer_count_packet (MpegTSPacketizer2 * packetizer,
+    MpegTSPacketizerPacket * packet);
 
 #define CONTINUITY_UNSET 255
 #define VERSION_NUMBER_UNSET 255
@@ -275,6 +277,9 @@ mpegts_packetizer_init (MpegTSPacketizer2 * packetizer)
   packetizer->last_pts = GST_CLOCK_TIME_NONE;
   packetizer->last_dts = GST_CLOCK_TIME_NONE;
   packetizer->extra_shift = 0;
+
+  g_mutex_init (&packetizer->pid_stats_lock);
+  packetizer->pid_stats = NULL;
 }
 
 static void
@@ -302,6 +307,9 @@ mpegts_packetizer_dispose (GObject * object)
     packetizer->empty = TRUE;
 
     flush_observations (packetizer);
+
+    mpegts_packetizer_set_packet_counts_enabled (packetizer, FALSE);
+    g_mutex_clear (&packetizer->pid_stats_lock);
   }
 
   if (G_OBJECT_CLASS (mpegts_packetizer_parent_class)->dispose)
@@ -503,6 +511,7 @@ mpegts_packetizer_parse_packet (MpegTSPacketizer2 * packetizer,
   else
     packet->payload = NULL;
 
+  mpegts_packetizer_count_packet (packetizer, packet);
   return PACKET_OK;
 }
 
@@ -2603,3 +2612,84 @@ mpegts_packetizer_set_current_pcr_offset (MpegTSPacketizer2 * packetizer,
   }
   PACKETIZER_GROUP_UNLOCK (packetizer);
 }
+
+void
+mpegts_packetizer_set_packet_counts_enabled (MpegTSPacketizer2 * packetizer,
+    gboolean enabled)
+{
+  g_mutex_lock (&packetizer->pid_stats_lock);
+
+  if (enabled && !packetizer->pid_stats)
+    packetizer->pid_stats = g_array_new (FALSE, FALSE, sizeof (MpegTSPIDStats));
+  else if (!enabled && packetizer->pid_stats)
+    g_clear_pointer (&packetizer->pid_stats, g_array_unref);
+
+  g_mutex_unlock (&packetizer->pid_stats_lock);
+}
+
+static void
+mpegts_packetizer_count_packet (MpegTSPacketizer2 * packetizer,
+    MpegTSPacketizerPacket * packet)
+{
+  GArray *array;
+  gint16 pid;
+  MpegTSPIDStats *stats = NULL;
+  gint i;
+
+  if (!packetizer->pid_stats)
+    return;
+
+  g_mutex_lock (&packetizer->pid_stats_lock);
+
+  array = packetizer->pid_stats;
+  if (!array)
+    goto out;
+
+  pid = packet->pid;
+
+  for (i = 0; i < array->len; i++) {
+    stats = &g_array_index (array, MpegTSPIDStats, i);
+
+    if (stats->pid == pid)
+      break;
+  }
+
+  if (i == array->len) {
+    MpegTSPIDStats new_stats = {.pid = pid };
+    g_array_append_val (array, new_stats);
+    stats = &g_array_index (array, MpegTSPIDStats, i);
+  }
+
+  stats->packets += 1;
+
+out:
+  g_mutex_unlock (&packetizer->pid_stats_lock);
+}
+
+GstStructure *
+mpegts_packetizer_get_packet_counts (MpegTSPacketizer2 * packetizer)
+{
+  GArray *array;
+  GstStructure *s;
+  gint i;
+
+  s = gst_structure_new_empty ("application/x-gst-mpegts-packet-counts");
+
+  g_mutex_lock (&packetizer->pid_stats_lock);
+
+  array = packetizer->pid_stats;
+  if (!array)
+    goto out;
+
+  for (i = 0; i < array->len; i++) {
+    MpegTSPIDStats *stats = &g_array_index (array, MpegTSPIDStats, i);
+    gchar field_name[7];
+
+    g_snprintf (field_name, sizeof field_name, "0x%04x", stats->pid);
+    gst_structure_set (s, field_name, G_TYPE_UINT64, stats->packets, NULL);
+  }
+
+out:
+  g_mutex_unlock (&packetizer->pid_stats_lock);
+  return s;
+}

subprojects/gst-plugins-bad/gst/mpegtsdemux/mpegtspacketizer.h

@@ -241,6 +241,13 @@ typedef struct _MpegTSPCR
   PCROffsetCurrent *current;
 } MpegTSPCR;
 
+typedef struct _MpegTSPIDStats {
+  guint16 pid;
+
+  /* Number of packets seen */
+  guint64 packets;
+} MpegTSPIDStats;
+
 struct _MpegTSPacketizer2 {
   GObject     parent;
 
@@ -291,6 +298,10 @@ struct _MpegTSPacketizer2 {
   /* Extra time offset to handle values before initial PCR.
    * This will be added to all converted timestamps */
   GstClockTime extra_shift;
+
+  /* Tracks stats for each PID */
+  GMutex pid_stats_lock;
+  GArray *pid_stats;
 };
 
 struct _MpegTSPacketizer2Class {
@@ -387,6 +398,12 @@ mpegts_packetizer_set_reference_offset (MpegTSPacketizer2 * packetizer,
 G_GNUC_INTERNAL void
 mpegts_packetizer_set_pcr_discont_threshold (MpegTSPacketizer2 * packetizer,
 					GstClockTime threshold);
+
+G_GNUC_INTERNAL void
+mpegts_packetizer_set_packet_counts_enabled (MpegTSPacketizer2 * packetizer,
+					     gboolean enabled);
+G_GNUC_INTERNAL GstStructure *
+mpegts_packetizer_get_packet_counts (MpegTSPacketizer2 * packetizer);
 G_END_DECLS
 
 #endif /* GST_MPEGTS_PACKETIZER_H */