gsttensor_aggregator.c

Go to the documentation of this file.

 
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
 
#include <string.h>
#include "gsttensor_aggregator.h"
#include "tensor_meta.h"
#include <nnstreamer_util.h>
 
#ifndef DBG
#define DBG (!self->silent)
#endif
 
#define silent_debug_config(self,c,msg) do { \
  if (DBG) { \
    if (c) { \
      gchar *dim_str; \
      dim_str = gst_tensor_get_dimension_string ((c)->info.info[0].dimension); \
      GST_DEBUG_OBJECT (self, msg " type=%d dim=%s rate=%d/%d", (c)->info.info[0].type, dim_str, (c)->rate_n, (c)->rate_d); \
      g_free (dim_str); \
    } \
  } \
} while (0)
 
GST_DEBUG_CATEGORY_STATIC (gst_tensor_aggregator_debug);
#define GST_CAT_DEFAULT gst_tensor_aggregator_debug
 
enum
{
  PROP_0,
  PROP_FRAMES_IN,
  PROP_FRAMES_OUT,
  PROP_FRAMES_FLUSH,
  PROP_FRAMES_DIMENSION,
  PROP_CONCAT,
  PROP_SILENT
};
 
#define DEFAULT_SILENT TRUE
 
#define DEFAULT_FRAMES_IN 1
 
#define DEFAULT_FRAMES_OUT 1
 
#define DEFAULT_FRAMES_FLUSH 0
 
#define DEFAULT_FRAMES_DIMENSION (NNS_TENSOR_RANK_LIMIT - 1)
 
#define DEFAULT_CONCAT TRUE
 
#define CAPS_STRING GST_TENSOR_CAP_DEFAULT ";" GST_TENSORS_CAP_WITH_NUM ("1")
 
static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
    GST_PAD_SINK,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS (CAPS_STRING));
 
static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
    GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS (CAPS_STRING));
 
#define gst_tensor_aggregator_parent_class parent_class
G_DEFINE_TYPE (GstTensorAggregator, gst_tensor_aggregator, GST_TYPE_ELEMENT);
 
static void gst_tensor_aggregator_finalize (GObject * object);
static void gst_tensor_aggregator_set_property (GObject * object,
    guint prop_id, const GValue * value, GParamSpec * pspec);
static void gst_tensor_aggregator_get_property (GObject * object,
    guint prop_id, GValue * value, GParamSpec * pspec);
 
static gboolean gst_tensor_aggregator_sink_event (GstPad * pad,
    GstObject * parent, GstEvent * event);
static gboolean gst_tensor_aggregator_sink_query (GstPad * pad,
    GstObject * parent, GstQuery * query);
static gboolean gst_tensor_aggregator_src_query (GstPad * pad,
    GstObject * parent, GstQuery * query);
static GstFlowReturn gst_tensor_aggregator_chain (GstPad * pad,
    GstObject * parent, GstBuffer * buf);
static GstStateChangeReturn
gst_tensor_aggregator_change_state (GstElement * element,
    GstStateChange transition);
 
static void gst_tensor_aggregator_reset (GstTensorAggregator * self);
static GstCaps *gst_tensor_aggregator_query_caps (GstTensorAggregator * self,
    GstPad * pad, GstCaps * filter);
static gboolean gst_tensor_aggregator_parse_caps (GstTensorAggregator * self,
    const GstCaps * caps);
 
static void
gst_tensor_aggregator_class_init (GstTensorAggregatorClass * klass)
{
  GObjectClass *object_class;
  GstElementClass *element_class;
 
  GST_DEBUG_CATEGORY_INIT (gst_tensor_aggregator_debug, "tensor_aggregator", 0,
      "Element to aggregate tensor stream");
 
  object_class = (GObjectClass *) klass;
  element_class = (GstElementClass *) klass;
 
  object_class->set_property = gst_tensor_aggregator_set_property;
  object_class->get_property = gst_tensor_aggregator_get_property;
  object_class->finalize = gst_tensor_aggregator_finalize;
 
  g_object_class_install_property (object_class, PROP_FRAMES_IN,
      g_param_spec_uint ("frames-in", "Frames in input",
          "The number of frames in incoming buffer", 1, G_MAXUINT,
          DEFAULT_FRAMES_IN, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (object_class, PROP_FRAMES_OUT,
      g_param_spec_uint ("frames-out", "Frames in output",
          "The number of frames in outgoing buffer", 1, G_MAXUINT,
          DEFAULT_FRAMES_OUT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (object_class, PROP_FRAMES_FLUSH,
      g_param_spec_uint ("frames-flush", "Frames to flush",
          "The number of frames to flush (0 to flush all output)", 0, G_MAXUINT,
          DEFAULT_FRAMES_FLUSH, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (object_class, PROP_FRAMES_DIMENSION,
      g_param_spec_uint ("frames-dim", "Dimension index of frames",
          "The dimension index of frames in tensor",
          0, (NNS_TENSOR_RANK_LIMIT - 1), DEFAULT_FRAMES_DIMENSION,
          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (object_class, PROP_CONCAT,
      g_param_spec_boolean ("concat", "Concat", "Concatenate output buffer",
          DEFAULT_CONCAT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (object_class, PROP_SILENT,
      g_param_spec_boolean ("silent", "Silent", "Produce verbose output",
          DEFAULT_SILENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  gst_element_class_set_static_metadata (element_class,
      "TensorAggregator",
      "Filter/Tensor",
      "Element to aggregate tensor stream", "Samsung Electronics Co., Ltd.");
 
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&src_template));
  gst_element_class_add_pad_template (element_class,
      gst_static_pad_template_get (&sink_template));
 
  element_class->change_state = gst_tensor_aggregator_change_state;
}
 
static void
gst_tensor_aggregator_init (GstTensorAggregator * self)
{
  self->sinkpad = gst_pad_new_from_static_template (&sink_template, "sink");
  gst_pad_set_event_function (self->sinkpad,
      GST_DEBUG_FUNCPTR (gst_tensor_aggregator_sink_event));
  gst_pad_set_query_function (self->sinkpad,
      GST_DEBUG_FUNCPTR (gst_tensor_aggregator_sink_query));
  gst_pad_set_chain_function (self->sinkpad,
      GST_DEBUG_FUNCPTR (gst_tensor_aggregator_chain));
  GST_PAD_SET_PROXY_CAPS (self->sinkpad);
  gst_element_add_pad (GST_ELEMENT (self), self->sinkpad);
 
  self->srcpad = gst_pad_new_from_static_template (&src_template, "src");
  gst_pad_set_query_function (self->srcpad,
      GST_DEBUG_FUNCPTR (gst_tensor_aggregator_src_query));
  GST_PAD_SET_PROXY_CAPS (self->srcpad);
  gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
 
  self->silent = DEFAULT_SILENT;
  self->frames_in = DEFAULT_FRAMES_IN;
  self->frames_out = DEFAULT_FRAMES_OUT;
  self->frames_flush = DEFAULT_FRAMES_FLUSH;
  self->frames_dim = DEFAULT_FRAMES_DIMENSION;
  self->concat = DEFAULT_CONCAT;
 
  self->tensor_configured = FALSE;
  gst_tensors_config_init (&self->in_config);
  gst_tensors_config_init (&self->out_config);
 
  self->adapter_table = gst_tensor_aggregation_init ();
  gst_tensor_aggregator_reset (self);
}
 
static void
gst_tensor_aggregator_finalize (GObject * object)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (object);
 
  gst_tensor_aggregator_reset (self);
 
  gst_tensors_config_free (&self->in_config);
  gst_tensors_config_free (&self->out_config);
  g_hash_table_destroy (self->adapter_table);
 
  G_OBJECT_CLASS (parent_class)->finalize (object);
}
 
static void
gst_tensor_aggregator_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (object);
 
  switch (prop_id) {
    case PROP_FRAMES_IN:
      self->frames_in = g_value_get_uint (value);
      break;
    case PROP_FRAMES_OUT:
      self->frames_out = g_value_get_uint (value);
      break;
    case PROP_FRAMES_FLUSH:
      self->frames_flush = g_value_get_uint (value);
      break;
    case PROP_FRAMES_DIMENSION:
      self->frames_dim = g_value_get_uint (value);
      break;
    case PROP_CONCAT:
      self->concat = g_value_get_boolean (value);
      break;
    case PROP_SILENT:
      self->silent = g_value_get_boolean (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}
 
static void
gst_tensor_aggregator_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (object);
 
  switch (prop_id) {
    case PROP_FRAMES_IN:
      g_value_set_uint (value, self->frames_in);
      break;
    case PROP_FRAMES_OUT:
      g_value_set_uint (value, self->frames_out);
      break;
    case PROP_FRAMES_FLUSH:
      g_value_set_uint (value, self->frames_flush);
      break;
    case PROP_FRAMES_DIMENSION:
      g_value_set_uint (value, self->frames_dim);
      break;
    case PROP_CONCAT:
      g_value_set_boolean (value, self->concat);
      break;
    case PROP_SILENT:
      g_value_set_boolean (value, self->silent);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}
 
static gboolean
gst_tensor_aggregator_sink_event (GstPad * pad, GstObject * parent,
    GstEvent * event)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (parent);
 
  GST_DEBUG_OBJECT (self, "Received %s event: %" GST_PTR_FORMAT,
      GST_EVENT_TYPE_NAME (event), event);
 
  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_CAPS:
    {
      GstCaps *in_caps;
      GstCaps *out_caps;
 
      gst_event_parse_caps (event, &in_caps);
      silent_debug_caps (self, in_caps, "in-caps");
 
      if (gst_tensor_aggregator_parse_caps (self, in_caps)) {
        gboolean ret = FALSE;
 
        out_caps =
            gst_tensor_pad_caps_from_config (self->srcpad, &self->out_config);
        silent_debug_caps (self, out_caps, "out-caps");
 
        ret = gst_pad_set_caps (self->srcpad, out_caps);
 
        gst_event_unref (event);
        gst_caps_unref (out_caps);
 
        return ret;
      }
      break;
    }
    case GST_EVENT_FLUSH_STOP:
      gst_tensor_aggregator_reset (self);
      break;
    default:
      break;
  }
 
  return gst_pad_event_default (pad, parent, event);
}
 
static gboolean
gst_tensor_aggregator_sink_query (GstPad * pad, GstObject * parent,
    GstQuery * query)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (parent);
 
  GST_DEBUG_OBJECT (self, "Received %s query: %" GST_PTR_FORMAT,
      GST_QUERY_TYPE_NAME (query), query);
 
  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_CAPS:
    {
      GstCaps *caps;
      GstCaps *filter;
 
      gst_query_parse_caps (query, &filter);
      caps = gst_tensor_aggregator_query_caps (self, pad, filter);
 
      gst_query_set_caps_result (query, caps);
      gst_caps_unref (caps);
      return TRUE;
    }
    case GST_QUERY_ACCEPT_CAPS:
    {
      GstCaps *caps;
      GstCaps *template_caps;
      gboolean res = FALSE;
 
      gst_query_parse_accept_caps (query, &caps);
      silent_debug_caps (self, caps, "accept-caps");
 
      if (gst_caps_is_fixed (caps)) {
        template_caps = gst_pad_get_pad_template_caps (pad);
 
        res = gst_caps_can_intersect (template_caps, caps);
        gst_caps_unref (template_caps);
      }
 
      gst_query_set_accept_caps_result (query, res);
      return TRUE;
    }
    default:
      break;
  }
 
  return gst_pad_query_default (pad, parent, query);
}
 
static gboolean
gst_tensor_aggregator_src_query (GstPad * pad, GstObject * parent,
    GstQuery * query)
{
  GstTensorAggregator *self;
 
  self = GST_TENSOR_AGGREGATOR (parent);
 
  GST_DEBUG_OBJECT (self, "Received %s query: %" GST_PTR_FORMAT,
      GST_QUERY_TYPE_NAME (query), query);
 
  switch (GST_QUERY_TYPE (query)) {
    case GST_QUERY_CAPS:
    {
      GstCaps *caps;
      GstCaps *filter;
 
      gst_query_parse_caps (query, &filter);
      caps = gst_tensor_aggregator_query_caps (self, pad, filter);
 
      gst_query_set_caps_result (query, caps);
      gst_caps_unref (caps);
      return TRUE;
    }
    default:
      break;
  }
 
  return gst_pad_query_default (pad, parent, query);
}
 
static GstAdapter *
gst_tensor_aggregator_get_adapter (GstTensorAggregator * self, GstBuffer * buf)
{
  GstMetaQuery *meta;
  guint32 key = 0;
 
  meta = gst_buffer_get_meta_query (buf);
  if (meta)
    key = meta->client_id;
 
  return gst_tensor_aggregation_get_adapter (self->adapter_table, key);
}
 
static gboolean
gst_tensor_aggregator_check_concat_axis (GstTensorAggregator * self,
    const GstTensorInfo * info)
{
  guint i;
 
  g_assert (info != NULL); 
  if (self->concat && self->frames_out > 1) {
    for (i = self->frames_dim + 1; i < NNS_TENSOR_RANK_LIMIT; i++) {
      if (info->dimension[i] > 1) {
        return TRUE;
      }
    }
  }
 
  return FALSE;
}
 
static gboolean
gst_tensor_aggregator_concat (GstTensorAggregator * self, GstBuffer * outbuf,
    const GstTensorInfo * info)
{
  GstBuffer *srcbuf;
  GstMapInfo src_info, dest_info;
  guint f;
  gsize block_size;
  gsize src_idx, dest_idx;
  gsize frame_size;
 
  frame_size = gst_tensor_info_get_size (info);
  g_assert (frame_size > 0); 
  srcbuf = gst_buffer_copy (outbuf);
  outbuf = gst_buffer_make_writable (outbuf);
 
  if (!gst_buffer_map (srcbuf, &src_info, GST_MAP_READ)) {
    ml_logf ("Failed to map source buffer with tensor_aggregator.\n");
    gst_buffer_unref (srcbuf);
    return FALSE;
  }
  if (!gst_buffer_map (outbuf, &dest_info, GST_MAP_WRITE)) {
    ml_logf ("Failed to map destination buffer with tensor_aggregator.\n");
    gst_buffer_unmap (srcbuf, &src_info);
    gst_buffer_unref (srcbuf);
    return FALSE;
  }
 
  block_size = gst_tensor_get_element_size (info->type);
  for (f = 0; f <= self->frames_dim; f++) {
    block_size *= info->dimension[f];
  }
 
  src_idx = dest_idx = 0;
 
  do {
    for (f = 0; f < self->frames_out; f++) {
      nns_memcpy (dest_info.data + dest_idx,
          src_info.data + src_idx + (frame_size * f), block_size);
      dest_idx += block_size;
    }
 
    src_idx += block_size;
 
    g_assert (src_idx <= frame_size);
    g_assert (dest_idx <= dest_info.size);
  } while (src_idx < frame_size);
 
  gst_buffer_unmap (srcbuf, &src_info);
  gst_buffer_unmap (outbuf, &dest_info);
 
  gst_buffer_unref (srcbuf);
 
  return TRUE;
}
 
static GstFlowReturn
gst_tensor_aggregator_push (GstTensorAggregator * self, GstBuffer * outbuf,
    gsize frame_size)
{
  GstTensorInfo info;
 
  info = self->out_config.info.info[0];
  g_assert (self->frames_dim < NNS_TENSOR_RANK_LIMIT);
  info.dimension[self->frames_dim] /= self->frames_out;
 
  if (frame_size != gst_tensor_info_get_size (&info) || frame_size == 0U) {
    ml_logf
        ("Invalid output capability of tensor_aggregator. Frame size = %"
        G_GSIZE_FORMAT "\n", frame_size);
    return GST_FLOW_ERROR;
  }
 
  if (gst_tensor_aggregator_check_concat_axis (self, &info)) {
    if (!gst_tensor_aggregator_concat (self, outbuf, &info))
      return GST_FLOW_ERROR;
  }
 
  return gst_pad_push (self->srcpad, outbuf);
}
 
static GstFlowReturn
gst_tensor_aggregator_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
{
  GstTensorAggregator *self;
  GstFlowReturn ret = GST_FLOW_OK;
  GstAdapter *adapter;
  gsize avail, buf_size, frame_size, out_size;
  guint frames_in, frames_out, frames_flush;
  GstClockTime duration;
  UNUSED (pad);
 
  self = GST_TENSOR_AGGREGATOR (parent);
  g_assert (self->tensor_configured);
 
  buf_size = gst_buffer_get_size (buf);
  g_return_val_if_fail (buf_size > 0, GST_FLOW_ERROR);
 
  frames_in = self->frames_in;
  frames_out = self->frames_out;
  frames_flush = self->frames_flush;
  frame_size = buf_size / frames_in;
 
  if (frames_in == frames_out) {
    return gst_tensor_aggregator_push (self, buf, frame_size);
  }
 
  adapter = gst_tensor_aggregator_get_adapter (self, buf);
  g_assert (adapter != NULL);
 
  duration = GST_BUFFER_DURATION (buf);
  if (GST_CLOCK_TIME_IS_VALID (duration)) {
    duration = gst_util_uint64_scale_int (duration, frames_out, frames_in);
  }
 
  gst_adapter_push (adapter, buf);
 
  out_size = frame_size * frames_out;
  g_assert (out_size > 0);
 
  while ((avail = gst_adapter_available (adapter)) >= out_size &&
      ret == GST_FLOW_OK) {
    GstBuffer *outbuf;
    GstClockTime pts, dts;
    guint64 pts_dist, dts_dist;
    gsize flush;
 
    pts = gst_adapter_prev_pts (adapter, &pts_dist);
    dts = gst_adapter_prev_dts (adapter, &dts_dist);
 
    if (frames_in > 1) {
      gint fn, fd;
 
      fn = self->in_config.rate_n;
      fd = self->in_config.rate_d;
 
      if (fn > 0 && fd > 0) {
        if (GST_CLOCK_TIME_IS_VALID (pts)) {
          pts +=
              gst_util_uint64_scale_int (pts_dist * fd, GST_SECOND,
              fn * frame_size);
        }
 
        if (GST_CLOCK_TIME_IS_VALID (dts)) {
          dts +=
              gst_util_uint64_scale_int (dts_dist * fd, GST_SECOND,
              fn * frame_size);
        }
      }
    }
 
    outbuf = gst_adapter_get_buffer (adapter, out_size);
    outbuf = gst_buffer_make_writable (outbuf);
 
    GST_BUFFER_PTS (outbuf) = pts;
    GST_BUFFER_DTS (outbuf) = dts;
    GST_BUFFER_DURATION (outbuf) = duration;
 
    ret = gst_tensor_aggregator_push (self, outbuf, frame_size);
 
    if (frames_flush > 0) {
      flush = frame_size * frames_flush;
 
      if (flush > avail) {
        flush = avail;
      }
    } else {
      flush = out_size;
    }
 
    gst_adapter_flush (adapter, flush);
  }
 
  return ret;
}
 
static GstStateChangeReturn
gst_tensor_aggregator_change_state (GstElement * element,
    GstStateChange transition)
{
  GstTensorAggregator *self;
  GstStateChangeReturn ret;
 
  self = GST_TENSOR_AGGREGATOR (element);
 
  switch (transition) {
    case GST_STATE_CHANGE_READY_TO_PAUSED:
      gst_tensor_aggregator_reset (self);
      break;
    default:
      break;
  }
 
  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 
  switch (transition) {
    case GST_STATE_CHANGE_PAUSED_TO_READY:
      gst_tensor_aggregator_reset (self);
      break;
    default:
      break;
  }
 
  return ret;
}
 
static void
gst_tensor_aggregator_reset (GstTensorAggregator * self)
{
  /* remove all buffers from adapter */
  gst_tensor_aggregation_clear_all (self->adapter_table);
}
 
static GstCaps *
gst_tensor_aggregator_query_caps (GstTensorAggregator * self, GstPad * pad,
    GstCaps * filter)
{
  GstCaps *caps;
  GstTensorsConfig *config;
 
  /* tensor config info for given pad */
  if (pad == self->sinkpad) {
    config = &self->in_config;
  } else {
    config = &self->out_config;
  }
 
  /* caps from tensor config info */
  caps = gst_tensor_pad_possible_caps_from_config (pad, config);
 
  silent_debug_caps (self, caps, "caps");
  silent_debug_caps (self, filter, "filter");
 
  if (caps && filter) {
    GstCaps *intersection;
 
    intersection =
        gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
 
    gst_caps_unref (caps);
    caps = intersection;
  }
 
  return caps;
}
 
static gboolean
gst_tensor_aggregator_parse_caps (GstTensorAggregator * self,
    const GstCaps * caps)
{
  GstStructure *structure;
  GstTensorsConfig config;
  GstTensorInfo *_info;
  uint32_t per_frame;
  guint count;
 
  g_return_val_if_fail (caps != NULL, FALSE);
  g_return_val_if_fail (gst_caps_is_fixed (caps), FALSE);
 
  structure = gst_caps_get_structure (caps, 0);
 
  if (!gst_tensors_config_from_structure (&config, structure) ||
      !gst_tensors_config_validate (&config)) {
    GST_ERROR_OBJECT (self, "Cannot configure tensor info");
    return FALSE;
  }
 
  count = (self->frames_out + self->frames_in - 1) / self->frames_in;
  if (self->frames_in * count < self->frames_flush) {
    GST_ERROR_OBJECT (self, "Cannot flush frames");
    return FALSE;
  }
 
  self->in_config = config;
  /* tensor-aggregator now handles single tensor. */
  _info = &config.info.info[0];
 
  if (self->frames_dim >= NNS_TENSOR_RANK_LIMIT ||
      (_info->dimension[self->frames_dim] % self->frames_in) != 0) {
    GST_ERROR_OBJECT (self, "Cannot update dimension in output tensor");
    return FALSE;
  }
  per_frame = _info->dimension[self->frames_dim] / self->frames_in;
 
  _info->dimension[self->frames_dim] = per_frame * self->frames_out;
  self->out_config = config;
  self->tensor_configured = TRUE;
 
  silent_debug_config (self, &self->in_config, "in-tensor");
  silent_debug_config (self, &self->out_config, "out-tensor");
  return TRUE;
}