gsttensor_demux.c

Go to the documentation of this file.

 
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
 
#include <string.h>
#include <gst/gst.h>
#include <glib.h>
#include <nnstreamer_util.h>
 
#include "gsttensor_demux.h"
 
GST_DEBUG_CATEGORY_STATIC (gst_tensor_demux_debug);
#define GST_CAT_DEFAULT gst_tensor_demux_debug
 
#define CAPS_STRING_SINK GST_TENSORS_CAP_MAKE ("{ static, flexible }")
 
#define CAPS_STRING_SRC GST_TENSOR_CAP_DEFAULT ";" GST_TENSORS_CAP_MAKE ("{ static, flexible }")
 
enum
{
  PROP_0,
  PROP_SILENT,
  PROP_TENSORPICK
};
 
static GstStaticPadTemplate src_templ = GST_STATIC_PAD_TEMPLATE ("src_%u",
    GST_PAD_SRC,
    GST_PAD_SOMETIMES,
    GST_STATIC_CAPS (CAPS_STRING_SRC)
    );
 
static GstStaticPadTemplate sink_templ = GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS (CAPS_STRING_SINK)
    );
 
static GstFlowReturn gst_tensor_demux_chain (GstPad * pad, GstObject * parent,
    GstBuffer * buf);
static gboolean gst_tensor_demux_event (GstPad * pad, GstObject * parent,
    GstEvent * event);
static GstStateChangeReturn gst_tensor_demux_change_state (GstElement * element,
    GstStateChange transition);
static void gst_tensor_demux_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec);
static void gst_tensor_demux_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec);
static void gst_tensor_demux_dispose (GObject * object);
#define gst_tensor_demux_parent_class parent_class
G_DEFINE_TYPE (GstTensorDemux, gst_tensor_demux, GST_TYPE_ELEMENT);
 
 
static void
gst_tensor_demux_class_init (GstTensorDemuxClass * klass)
{
  GObjectClass *gobject_class;
  GstElementClass *gstelement_class;
 
  GST_DEBUG_CATEGORY_INIT (gst_tensor_demux_debug, "tensor_demux", 0,
      "Element to demux tensors to tensor stream");
 
  gobject_class = (GObjectClass *) klass;
  gstelement_class = (GstElementClass *) klass;
 
  parent_class = g_type_class_peek_parent (klass);
 
  gobject_class->dispose = gst_tensor_demux_dispose;
  gobject_class->get_property = gst_tensor_demux_get_property;
  gobject_class->set_property = gst_tensor_demux_set_property;
 
  g_object_class_install_property (gobject_class, PROP_SILENT,
      g_param_spec_boolean ("silent", "Silent", "Produce verbose output ?",
          TRUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (gobject_class, PROP_TENSORPICK,
      g_param_spec_string ("tensorpick", "TensorPick",
          "Choose nth tensor among tensors ?", "",
          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  gstelement_class->change_state =
      GST_DEBUG_FUNCPTR (gst_tensor_demux_change_state);
 
  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&sink_templ));
  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&src_templ));
 
  gst_element_class_set_details_simple (gstelement_class,
      "TensorDemux",
      "Demuxer/Tensor",
      "Demux tensors stream to other/tensor stream",
      "Jijoong Moon <jijoong.moon@samsung.com>");
}
 
static void
gst_tensor_demux_init (GstTensorDemux * tensor_demux)
{
  tensor_demux->sinkpad =
      gst_pad_new_from_static_template (&sink_templ, "sink");
  gst_element_add_pad (GST_ELEMENT_CAST (tensor_demux), tensor_demux->sinkpad);
  gst_pad_set_chain_function (tensor_demux->sinkpad,
      GST_DEBUG_FUNCPTR (gst_tensor_demux_chain));
  gst_pad_set_event_function (tensor_demux->sinkpad,
      GST_DEBUG_FUNCPTR (gst_tensor_demux_event));
 
  tensor_demux->num_srcpads = 0;
  tensor_demux->silent = TRUE;
  tensor_demux->tensorpick = NULL;
  tensor_demux->have_group_id = FALSE;
  tensor_demux->group_id = G_MAXUINT;
  tensor_demux->srcpads = NULL;
 
  gst_tensors_config_init (&tensor_demux->tensors_config);
}
 
static void
gst_tensor_demux_remove_src_pads (GstTensorDemux * tensor_demux)
{
  while (tensor_demux->srcpads != NULL) {
    GstTensorPad *tensor_pad = tensor_demux->srcpads->data;
    gst_element_remove_pad (GST_ELEMENT (tensor_demux), tensor_pad->pad);
    g_free (tensor_pad);
    tensor_demux->srcpads =
        g_slist_delete_link (tensor_demux->srcpads, tensor_demux->srcpads);
  }
  tensor_demux->srcpads = NULL;
  tensor_demux->num_srcpads = 0;
 
  gst_tensors_config_free (&tensor_demux->tensors_config);
  gst_tensors_config_init (&tensor_demux->tensors_config);
}
 
static void
gst_tensor_demux_dispose (GObject * object)
{
  GstTensorDemux *tensor_demux = GST_TENSOR_DEMUX (object);
 
  gst_tensor_demux_remove_src_pads (tensor_demux);
  g_list_free_full (tensor_demux->tensorpick, g_free);
  G_OBJECT_CLASS (parent_class)->dispose (object);
}
 
static gboolean
gst_tensor_demux_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
  GstTensorDemux *tensor_demux;
  tensor_demux = GST_TENSOR_DEMUX (parent);
 
  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_CAPS:
    {
      GstCaps *caps;
      gst_event_parse_caps (event, &caps);
      gst_tensors_config_from_cap (&tensor_demux->tensors_config, caps);
      break;
    }
    case GST_EVENT_EOS:
      if (!tensor_demux->srcpads) {
        GST_ELEMENT_ERROR (tensor_demux, STREAM, WRONG_TYPE,
            ("This stream contains no valid stremas."),
            ("Got EOS before adding any pads"));
        gst_event_unref (event);
        return FALSE;
      }
      break;
    default:
      break;
  }
 
  return gst_pad_event_default (pad, parent, event);
}
 
static gboolean
gst_tensor_demux_get_tensor_config (GstTensorDemux * tensor_demux,
    GstTensorsConfig * config, const guint nth, const guint total)
{
  gst_tensors_config_init (config);
 
  if (tensor_demux->tensorpick != NULL) {
    gchar *selected_tensor;
    gchar **strv;
    guint i, num, idx;
 
    g_assert (g_list_length (tensor_demux->tensorpick) >= nth);
 
    selected_tensor = (gchar *) g_list_nth_data (tensor_demux->tensorpick, nth);
    strv = g_strsplit_set (selected_tensor, ":+", -1);
    num = g_strv_length (strv);
 
    for (i = 0; i < num; i++) {
      idx = (guint) g_ascii_strtoll (strv[i], NULL, 10);
 
      /* Internal error, handle invalid index. */
      if (idx >= total) {
        g_strfreev (strv);
        return FALSE;
      }
 
      gst_tensor_info_copy (gst_tensors_info_get_nth_info (&config->info, i),
          gst_tensors_info_get_nth_info (&tensor_demux->tensors_config.info,
              idx));
    }
 
    config->info.num_tensors = num;
    g_strfreev (strv);
  } else {
    /* Internal error, handle invalid index. */
    if (nth >= total)
      return FALSE;
 
    config->info.num_tensors = 1;
    gst_tensor_info_copy (&config->info.info[0],
        gst_tensors_info_get_nth_info (&tensor_demux->tensors_config.info,
            nth));
  }
 
  config->info.format = tensor_demux->tensors_config.info.format;
  config->rate_n = tensor_demux->tensors_config.rate_n;
  config->rate_d = tensor_demux->tensors_config.rate_d;
  return TRUE;
}
 
static GstTensorPad *
gst_tensor_demux_get_tensor_pad (GstTensorDemux * tensor_demux,
    gboolean * created, const guint nth, const guint total)
{
  GstElement *element = GST_ELEMENT_CAST (tensor_demux);
  g_autofree gchar *element_name = gst_element_get_name (element);
  GSList *walk;
  GstPad *pad;
  GstTensorPad *tensorpad;
  gchar *name;
  GstEvent *event;
  gchar *stream_id;
  GstCaps *caps = NULL;
  GstTensorsConfig config;
 
  walk = tensor_demux->srcpads;
  while (walk) {
    GstTensorPad *pad = (GstTensorPad *) walk->data;
    if (nth == pad->nth) {
      if (created) {
        *created = FALSE;
      }
      return pad;
    }
    walk = walk->next;
  }
 
  tensorpad = g_new0 (GstTensorPad, 1);
  g_assert (tensorpad != NULL);
  GST_DEBUG_OBJECT (tensor_demux, "creating pad: %d(%dth)",
      tensor_demux->num_srcpads, nth);
 
  name = g_strdup_printf ("src_%u", tensor_demux->num_srcpads);
  pad = gst_pad_new_from_static_template (&src_templ, name);
  stream_id = gst_pad_create_stream_id_printf (pad, element,
      "%s-nnsdemux-%s-%08x", element_name, name, g_random_int ());
  g_free (name);
 
  tensorpad->pad = pad;
  tensorpad->nth = nth;
  tensorpad->last_ret = GST_FLOW_OK;
  tensorpad->last_ts = GST_CLOCK_TIME_NONE;
 
  tensor_demux->srcpads = g_slist_append (tensor_demux->srcpads, tensorpad);
  tensor_demux->num_srcpads++;
 
  gst_pad_use_fixed_caps (pad);
  gst_pad_set_active (pad, TRUE);
  gst_element_add_pad (GST_ELEMENT_CAST (tensor_demux), pad);
 
  if (!tensor_demux->have_group_id) {
    event =
        gst_pad_get_sticky_event (tensor_demux->sinkpad, GST_EVENT_STREAM_START,
        0);
    if (event) {
      tensor_demux->have_group_id =
          gst_event_parse_group_id (event, &tensor_demux->group_id);
      gst_event_unref (event);
    } else if (!tensor_demux->have_group_id) {
      tensor_demux->have_group_id = TRUE;
      tensor_demux->group_id = gst_util_group_id_next ();
    }
  }
 
  event = gst_event_new_stream_start (stream_id);
  if (tensor_demux->have_group_id)
    gst_event_set_group_id (event, tensor_demux->group_id);
 
  gst_pad_store_sticky_event (pad, event);
  g_free (stream_id);
  gst_event_unref (event);
 
  /* configure nth pad caps */
  if (gst_tensor_demux_get_tensor_config (tensor_demux, &config, nth, total)) {
    caps = gst_tensor_pad_caps_from_config (pad, &config);
 
    gst_pad_set_caps (pad, caps);
    gst_caps_unref (caps);
  } else {
    GST_WARNING_OBJECT (tensor_demux, "Unable to set pad caps");
  }
 
  if (created) {
    *created = TRUE;
  }
 
  if (tensor_demux->tensorpick != NULL) {
    GST_DEBUG_OBJECT (tensor_demux, "TensorPick is set! : %dth tensor\n", nth);
    if (g_list_length (tensor_demux->tensorpick) == tensor_demux->num_srcpads) {
      gst_element_no_more_pads (GST_ELEMENT_CAST (tensor_demux));
    }
  }
 
  gst_tensors_config_free (&config);
  return tensorpad;
}
 
static GstFlowReturn
gst_tensor_demux_combine_flows (GstTensorDemux * tensor_demux,
    GstTensorPad * pad, GstFlowReturn ret)
{
  GSList *walk;
  pad->last_ret = ret;
 
  if (ret != GST_FLOW_NOT_LINKED)
    goto done;
 
  for (walk = tensor_demux->srcpads; walk; walk = g_slist_next (walk)) {
    GstTensorPad *opad = (GstTensorPad *) walk->data;
    ret = opad->last_ret;
    if (ret != GST_FLOW_NOT_LINKED)
      goto done;
  }
done:
  return ret;
}
 
static GstFlowReturn
gst_tensor_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
{
  guint num_tensors, num_srcs, i, idx;
  GstFlowReturn res = GST_FLOW_OK;
  GstTensorDemux *tensor_demux;
  GList *list = NULL;
  GstTensorInfo *_info;
 
  UNUSED (pad);
  tensor_demux = GST_TENSOR_DEMUX (parent);
 
  buf = gst_tensor_buffer_from_config (buf, &tensor_demux->tensors_config);
 
  num_tensors = gst_tensor_buffer_get_count (buf);
  if (gst_tensors_config_is_static (&tensor_demux->tensors_config))
    g_assert (tensor_demux->tensors_config.info.num_tensors == num_tensors);
 
  GST_DEBUG_OBJECT (tensor_demux, " Number of Tensors: %d", num_tensors);
 
  num_srcs = num_tensors;
  if (tensor_demux->tensorpick != NULL) {
    num_srcs = g_list_length (tensor_demux->tensorpick);
    list = tensor_demux->tensorpick;
  }
 
  for (i = 0; i < num_srcs; i++) {
    GstTensorPad *srcpad;
    GstBuffer *outbuf;
    GstMemory *mem;
    gboolean created;
    GstClockTime ts;
 
    srcpad = gst_tensor_demux_get_tensor_pad (tensor_demux, &created, i,
        num_tensors);
    outbuf = gst_buffer_new ();
 
    if (tensor_demux->tensorpick != NULL) {
      guint num, j;
      gchar **strv = g_strsplit_set ((gchar *) list->data, ":+", -1);
 
      num = g_strv_length (strv);
      for (j = 0; j < num; j++) {
        idx = (guint) g_ascii_strtoll (strv[j], NULL, 10);
        _info =
            gst_tensors_info_get_nth_info (&tensor_demux->tensors_config.info,
            idx);
        mem = gst_tensor_buffer_get_nth_memory (buf, idx);
        if (!gst_tensor_buffer_append_memory (outbuf, mem, _info)) {
          gst_buffer_unref (outbuf);
          res = GST_FLOW_ERROR;
          goto error;
        }
      }
      g_strfreev (strv);
      list = list->next;
    } else {
      _info =
          gst_tensors_info_get_nth_info (&tensor_demux->tensors_config.info, i);
      mem = gst_tensor_buffer_get_nth_memory (buf, i);
      if (!gst_tensor_buffer_append_memory (outbuf, mem, _info)) {
        gst_buffer_unref (outbuf);
        res = GST_FLOW_ERROR;
        goto error;
      }
    }
 
    ts = GST_BUFFER_TIMESTAMP (buf);
 
    if (created) {
      GstSegment segment;
      gst_segment_init (&segment, GST_FORMAT_TIME);
      gst_pad_push_event (srcpad->pad, gst_event_new_segment (&segment));
    }
 
    outbuf = gst_buffer_make_writable (outbuf);
 
    /* metadata from incoming buffer */
    gst_buffer_copy_into (outbuf, buf, GST_BUFFER_COPY_METADATA, 0, -1);
 
    if (srcpad->last_ts == GST_CLOCK_TIME_NONE || srcpad->last_ts != ts) {
      srcpad->last_ts = ts;
    } else {
      GST_DEBUG_OBJECT (tensor_demux, "invalid timestamp %" GST_TIME_FORMAT,
          GST_TIME_ARGS (ts));
    }
 
    GST_DEBUG_OBJECT (tensor_demux,
        "pushing buffer with timestamp %" GST_TIME_FORMAT,
        GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (outbuf)));
    res = gst_pad_push (srcpad->pad, outbuf);
    res = gst_tensor_demux_combine_flows (tensor_demux, srcpad, res);
 
    if (res != GST_FLOW_OK)
      break;
  }
 
error:
  gst_buffer_unref (buf);
  return res;
}
 
static GstStateChangeReturn
gst_tensor_demux_change_state (GstElement * element, GstStateChange transition)
{
  GstTensorDemux *tensor_demux;
  GstStateChangeReturn ret;
  tensor_demux = GST_TENSOR_DEMUX (element);
  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
  if (ret == GST_STATE_CHANGE_FAILURE)
    return ret;
  switch (transition) {
    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
      break;
    case GST_STATE_CHANGE_PAUSED_TO_READY:
      tensor_demux->group_id = G_MAXUINT;
      tensor_demux->have_group_id = FALSE;
      gst_tensor_demux_remove_src_pads (tensor_demux);
      break;
    case GST_STATE_CHANGE_READY_TO_NULL:
      break;
    default:
      break;
  }
 
  return ret;
}
 
static void
gst_tensor_demux_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstTensorDemux *filter = GST_TENSOR_DEMUX (object);
  switch (prop_id) {
    case PROP_SILENT:
      filter->silent = g_value_get_boolean (value);
      break;
    case PROP_TENSORPICK:
    {
      guint i;
      const gchar *param = g_value_get_string (value);
      gchar **strv = g_strsplit_set (param, ",.;/", -1);
      guint num = g_strv_length (strv);
 
      /* Before setting the new Tensor Pick data, the existing one should be removed. */
      if (filter->tensorpick) {
        g_list_free_full (filter->tensorpick, g_free);
        filter->tensorpick = NULL;
      }
      for (i = 0; i < num; i++) {
        gchar *tmp = g_strdup (strv[i]);
        filter->tensorpick = g_list_append (filter->tensorpick, tmp);
      }
      g_strfreev (strv);
      break;
    }
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}
 
static void
gst_tensor_demux_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec)
{
  GstTensorDemux *filter = GST_TENSOR_DEMUX (object);
  switch (prop_id) {
    case PROP_SILENT:
      g_value_set_boolean (value, filter->silent);
      break;
    case PROP_TENSORPICK:
    {
      GList *list;
      char *p;
      GPtrArray *arr = g_ptr_array_new ();
      gchar **strings;
 
      for (list = filter->tensorpick; list != NULL; list = list->next) {
        g_ptr_array_add (arr, g_strdup_printf ("%s", (gchar *) list->data));
      }
      g_ptr_array_add (arr, NULL);
      strings = (gchar **) g_ptr_array_free (arr, FALSE);
      p = g_strjoinv (",", strings);
      g_strfreev (strings);
      g_value_take_string (value, p);
      break;
    }
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}