Add vp8 support for analytics (#1031)

qwu16 · web-flow · commit ad89b182a0d0 · 2021-07-12T10:38:01.000+08:00
diff --git a/source/agent/analytics/analytics-agent.js b/source/agent/analytics/analytics-agent.js
@@ -83,12 +83,6 @@ createInternalConnection(connectionId, direction, internalOpt) {
       this.onStatus(options.controller, connectionId, 'out', status);
 
       const newStreamId = algo + options.media.video.from;
-      const streamInfo = {
-          type: 'analytics',
-          media: {video: Object.assign({}, videoFormat, videoParameters)},
-          analyticsId: connectionId,
-          locality: {agent:this.agentId, node:this.rpcId},
-        };
 
       const pluginName = this.algorithms[algo].name;
       let codec = videoFormat.codec;
@@ -102,14 +96,11 @@ createInternalConnection(connectionId, direction, internalOpt) {
       log.debug('resolution:',resolution,'framerate:',framerate,'keyFrameInterval:',
                keyFrameInterval, 'bitrate:',bitrate);
 
-      this.engine.setOutputParam(codec,resolution,framerate,bitrate,keyFrameInterval,algo,pluginName);
+      this.engine.setInputParam(codec,resolution,framerate,bitrate,keyFrameInterval,algo,pluginName);
       if (this.engine.createPipeline() < 0) {
         return Promise.reject('Create pipeline failed');
       }
 
-      streamInfo.media.video.bitrate = bitrate;
-      this.onStreamGenerated(options.controller, newStreamId, streamInfo);
-
       if (this.engine.addElementMany() < 0) {
         return Promise.reject('Link element failed');
       }
@@ -125,6 +116,43 @@ createInternalConnection(connectionId, direction, internalOpt) {
           notifyStatus(options.controller, connectionId, 'out', {type: 'failed', reason: 'Analytics error: ' + error});
       });
 
+      var generateStream = this.onStreamGenerated;
+      const streamInfo = {
+        type: 'analytics',
+        media: {video: Object.assign({}, videoFormat, videoParameters)},
+        analyticsId: connectionId,
+        locality: {agent:this.agentId, node:this.rpcId},
+      };
+      this.engine.addEventListener('streamadded', function (data) {
+          log.debug('GStreamer pipeline stream generated:', data);
+          var params;
+            try {
+            params = JSON.parse(data)
+            streamInfo.media.video.bitrate = bitrate;
+            if (params.codec) {
+              streamInfo.media.video.codec = params.codec;
+            }
+
+            if (params.profile) {
+              if (params.profile === "constrained-baseline") {
+                streamInfo.media.video.profile = 'CB';
+              } else if (params.profile === "baseline") {
+                streamInfo.media.video.profile = 'B';
+              } else if (params.profile === "main") {
+                streamInfo.media.video.profile = 'M';
+              } else if (params.profile === "high") {
+                streamInfo.media.video.profile = 'H';
+              } else {
+                log.error("Not supported profile:", params.profile);
+                return;
+              }
+            }
+            generateStream(options.controller, newStreamId, streamInfo);
+          } catch (e) {
+            log.error("Parse stream added data with error:", e);
+          }
+      });
+
       return Promise.resolve();
     }
     return Promise.reject('Connection already exist: ' + connectionId);
diff --git a/source/agent/analytics/videoGstPipeline/GstInternalIn.cpp b/source/agent/analytics/videoGstPipeline/GstInternalIn.cpp
@@ -4,7 +4,6 @@
 
 #include "GstInternalIn.h"
 #include <gst/gst.h>
-#include <stdio.h>
 
 static void dump(void* index, uint8_t* buf, int len)
 {
@@ -18,6 +17,18 @@ static void dump(void* index, uint8_t* buf, int len)
     }
 }
 
+static void mem_put_le16(uint8_t *mem, unsigned int val) {
+    mem[0] = val;
+    mem[1] = val >> 8;
+}
+ 
+static void mem_put_le32(uint8_t *mem, unsigned int val) {
+    mem[0] = val;
+    mem[1] = val >> 8;
+    mem[2] = val >> 16;
+    mem[3] = val >> 24;
+}
+
 DEFINE_LOGGER(GstInternalIn, "GstInternalIn");
 GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int maxPort, std::string ticket)
 {
@@ -33,6 +44,8 @@ GstInternalIn::GstInternalIn(GstAppSrc *data, unsigned int minPort, unsigned int
     m_needKeyFrame = true;
     m_start = false;
     m_dumpIn = false;
+    num_frames = 0;
+    m_framerate = 30;
     char* pIn = std::getenv("DUMP_ANALYTICS_IN");
     if(pIn != NULL) {
         ELOG_INFO("Dump analytics in stream");
@@ -52,6 +65,13 @@ unsigned int GstInternalIn::getListeningPort()
 
 void GstInternalIn::setPushData(bool status){
     m_start = status;
+    if (!m_start) {
+        m_needKeyFrame = true;
+    }
+}
+
+void GstInternalIn::setFramerate(int framerate) {
+    m_framerate = framerate;
 }
 
 void GstInternalIn::onFeedback(const owt_base::FeedbackMsg& msg)
@@ -97,16 +117,56 @@ void GstInternalIn::onTransportData(char* buf, int len)
                 }
             }
 
+
+            size_t allocLength = payloadLength + headerLength;
+            uint8_t ivf_header[32] = {0};
+            uint8_t ivf_frame_header[12] = {0};
+            size_t ivf_header_length = 0;
+            size_t ivf_frame_header_length = 0;
+
+            if (frame->format == owt_base::FRAME_FORMAT_VP8) {
+                if (num_frames == 0) {
+                    ivf_header[0] = 'D';
+                    ivf_header[1] = 'K';
+                    ivf_header[2] = 'I';
+                    ivf_header[3] = 'F';
+
+                    mem_put_le16(ivf_header+4,  0);
+                    mem_put_le16(ivf_header+6,  32);
+                    ivf_header[8] = 'V';
+                    ivf_header[9] = 'P';
+                    ivf_header[10] = '8';
+                    ivf_header[11] = '0';
+                    mem_put_le16(ivf_header+12, frame->additionalInfo.video.width);
+                    mem_put_le16(ivf_header+14, frame->additionalInfo.video.height);
+                    mem_put_le32(ivf_header+16, m_framerate);
+                    mem_put_le32(ivf_header+20, 1);
+                    mem_put_le32(ivf_header+24, num_frames);
+                    mem_put_le32(ivf_header+28, 0);
+                    ivf_header_length = 32;
+                }
+                mem_put_le32(ivf_frame_header, payloadLength);
+                mem_put_le32(ivf_frame_header+4, num_frames);
+                ivf_frame_header_length = 12;
+            }
+
             /* Create a new empty buffer */
-            buffer = gst_buffer_new_and_alloc (payloadLength + headerLength);
+            buffer = gst_buffer_new_and_alloc (payloadLength + ivf_header_length + ivf_frame_header_length);
             gst_buffer_map(buffer, &map, GST_MAP_WRITE);
-            memcpy(map.data, frame, headerLength);
-            memcpy(map.data + headerLength, frame->payload, payloadLength);
+            if (frame->format == owt_base::FRAME_FORMAT_VP8) {
+                if (num_frames == 0) {
+                    memcpy(map.data, ivf_header, ivf_header_length);
+                }
+                memcpy(map.data + ivf_header_length, ivf_frame_header, ivf_frame_header_length);
+            }
+
+            memcpy(map.data + ivf_header_length + ivf_frame_header_length, frame->payload, payloadLength);
 
             if(m_dumpIn) {
                 dump(this, map.data, map.size);
             }
 
+
             gst_buffer_unmap(buffer, &map);
             g_signal_emit_by_name(appsrc, "push-buffer", buffer, &ret);
 
@@ -116,6 +176,7 @@ void GstInternalIn::onTransportData(char* buf, int len)
                 ELOG_DEBUG("Push buffer to appsrc got error\n");
                 m_start=false;
             }
+            num_frames++;
 
             break;
         }
diff --git a/source/agent/analytics/videoGstPipeline/GstInternalIn.h b/source/agent/analytics/videoGstPipeline/GstInternalIn.h
@@ -29,11 +29,14 @@ class GstInternalIn : public owt_base::RawTransportListener{
     void onTransportError() { }
     void onTransportConnected() { }
     void setPushData(bool status);
+    void setFramerate(int framerate);
 
 private:
     bool m_start;
     bool m_needKeyFrame;
     bool m_dumpIn;
+    size_t num_frames;
+    int m_framerate;
     GstAppSrc *appsrc;
     boost::shared_ptr<owt_base::RawTransportInterface> m_transport;
 };
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.cc b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.cc
@@ -79,6 +79,7 @@ static void dump(void* index, uint8_t* buf, int len)
     }
 }
 
+
 VideoGstAnalyzer::VideoGstAnalyzer(EventRegistry *handle) : m_asyncHandle(handle)
 {
     ELOG_INFO("Init");
@@ -142,7 +143,6 @@ gboolean VideoGstAnalyzer::StreamEventCallBack(GstBus *bus, GstMessage *message,
         g_main_loop_quit(pStreamObj->loop);
         break;
     case GST_MESSAGE_TAG:{
-        /* end-of-stream */
         GstTagList *tags = NULL;
         gst_message_parse_tag (message, &tags);
 
@@ -151,14 +151,58 @@ gboolean VideoGstAnalyzer::StreamEventCallBack(GstBus *bus, GstMessage *message,
         break;
     }
     case GST_MESSAGE_QOS:{
-        /* end-of-stream */
         ELOG_DEBUG("Got QOS message from %s \n",message->src->name);
         break;
     }
     case GST_MESSAGE_STATE_CHANGED:{
         GstState old_state, new_state, pending_state;
         gst_message_parse_state_changed(message, &old_state, &new_state, &pending_state);
-        ELOG_DEBUG("State change from %d to %d, play:%d \n",old_state, new_state, GST_STATE_PAUSED);
+        ELOG_DEBUG("State change from %d to %d, from:%d \n",old_state, new_state, message->src->name);
+        if (strcmp(message->src->name,"appsink") == 0 && new_state == GST_STATE_PLAYING) {
+            GstPad *pad = NULL;
+            GstCaps *caps = NULL;
+            pad = gst_element_get_static_pad (pStreamObj->sink, "sink");
+            if (!pad) {
+                ELOG_ERROR("Could not retrieve sink pad of sink element\n");
+            } else {
+                caps = gst_pad_get_current_caps (pad);
+                if (!caps)
+                caps = gst_pad_query_caps (pad, NULL);
+
+                /* Print and free */
+                const gchar *type;
+                int width, height;
+                std::string str;
+                GstStructure *structure;
+                structure = gst_caps_get_structure (caps, 0);
+                ELOG_DEBUG("Caps for the sink pad type is:%s %s\n",gst_structure_get_name(structure), gst_caps_to_string(caps));
+                type = gst_structure_get_name(structure);
+                if (strstr(type, "h264") != NULL) {
+                    str.append("{\"codec\":\"h264\",\"profile\":\"");
+                    str.append(gst_structure_get_string(structure, "profile"));
+                    str.append("\",");
+                    pStreamObj->outputcodec = "h264";
+                } else if (strstr(type, "vp8") != NULL) {
+                    str.append("{\"codec\":\"vp8\",");
+                    pStreamObj->outputcodec = "vp8";
+                }
+
+                gst_structure_get_int (structure, "width", &width);
+                gst_structure_get_int (structure, "height", &height);
+
+                str.append("\"width\":");
+                str.append(std::to_string(width));
+                str.append(",\"height\":");
+                str.append(std::to_string(height));
+                str.append("}");
+                pStreamObj->notifyAsyncEvent("streamadded", str.c_str());
+
+                gst_caps_unref (caps);
+                gst_object_unref (pad);
+            }
+
+        }
+
         break;
     }
     default:
@@ -219,7 +263,7 @@ int VideoGstAnalyzer::createPipeline()
         {"inputwidth", std::to_string(width)},
         {"inputheight", std::to_string(height)},
         {"inputframerate", std::to_string(framerate)},
-        {"inputcodec", codec},
+        {"inputcodec", inputcodec},
         {"pipelinename", algo} };
     pipeline_->PipelineConfig(plugin_config_map);
 
@@ -257,7 +301,6 @@ void VideoGstAnalyzer::stop_feed (GstElement * source, gpointer data)
 
 void VideoGstAnalyzer::new_sample_from_sink (GstElement * source, gpointer data)
 {
-    ELOG_DEBUG("Got new sample from sink\n");
     VideoGstAnalyzer* pStreamObj = static_cast<VideoGstAnalyzer*>(data);
     GstSample *sample;
     GstBuffer *buffer;
@@ -272,12 +315,23 @@ void VideoGstAnalyzer::new_sample_from_sink (GstElement * source, gpointer data)
     owt_base::Frame outFrame;
     memset(&outFrame, 0, sizeof(outFrame));
 
-    outFrame.format = owt_base::FRAME_FORMAT_H264;
+    if (pStreamObj->outputcodec.compare("vp8") == 0) {
+        outFrame.format = owt_base::FRAME_FORMAT_VP8;
+        outFrame.additionalInfo.video.isKeyFrame = isVp8KeyFrame(map.data, map.size);
+    } else if (pStreamObj->outputcodec.find("h264") != std::string::npos) {
+        outFrame.format = owt_base::FRAME_FORMAT_H264;
+        outFrame.additionalInfo.video.isKeyFrame = isH264KeyFrame(map.data, map.size);
+    } else {
+        printf("Not support codec:%s\n", pStreamObj->outputcodec.c_str());
+        gst_buffer_unmap(buffer, &map);
+        gst_sample_unref(sample);
+        return;
+    }
+
+    outFrame.timeStamp = (pStreamObj->m_frameCount++) * 1000 / pStreamObj->framerate * 90;
     outFrame.length = map.size;
     outFrame.additionalInfo.video.width = pStreamObj->width;
     outFrame.additionalInfo.video.height = pStreamObj->height;
-    outFrame.additionalInfo.video.isKeyFrame = isH264KeyFrame(map.data, map.size);
-    outFrame.timeStamp = (pStreamObj->m_frameCount++) * 1000 / 30 * 90;
 
     outFrame.payload = map.data;
 
@@ -353,8 +407,8 @@ int VideoGstAnalyzer::setPlaying()
 }
 
 void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort, std::string ticket) {
-    ELOG_DEBUG("Listening\n");
     m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort, ticket));
+    m_internalin->setFramerate(this->framerate);
 }
 
 void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* out)
@@ -397,10 +451,10 @@ int VideoGstAnalyzer::getListeningPort()
     return listeningPort; 
 }
 
-void VideoGstAnalyzer::setOutputParam(std::string codec, int width, int height, 
+void VideoGstAnalyzer::setInputParam(std::string codec, int width, int height,
     int framerate, int bitrate, int kfi, std::string algo, std::string libraryName)
 {
-    this->codec = codec;
+    this->inputcodec = codec;
     this->width = width;
     this->height = height;
     this->framerate = framerate;
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.h b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.h
@@ -35,7 +35,7 @@ class VideoGstAnalyzer : public EventRegistry {
 
     int addElementMany();
 
-    void setOutputParam(std::string codec, int width, int height, 
+    void setInputParam(std::string codec, int width, int height,
     int framerate, int bitrate, int kfi, std::string algo, std::string pluginName);
 
     void stopLoop();
@@ -86,7 +86,8 @@ class VideoGstAnalyzer : public EventRegistry {
     int m_frameCount;
 
     //param
-    std::string codec;
+    std::string inputcodec;
+    std::string outputcodec;
     std::string algo,libraryName;
     std::string resolution;
     int width,height;
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.cc b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.cc
@@ -30,7 +30,7 @@ void VideoGstAnalyzerWrap::Init(Handle<Object> exports, Handle<Object> module) {
   NODE_SET_PROTOTYPE_METHOD(tpl, "emitListenTo", emitListenTo);
   NODE_SET_PROTOTYPE_METHOD(tpl, "addElementMany", addElementMany);
   NODE_SET_PROTOTYPE_METHOD(tpl, "setPlaying", setPlaying);
-  NODE_SET_PROTOTYPE_METHOD(tpl, "setOutputParam", setOutputParam);
+  NODE_SET_PROTOTYPE_METHOD(tpl, "setInputParam", setInputParam);
   NODE_SET_PROTOTYPE_METHOD(tpl, "disconnect", disconnect);
   NODE_SET_PROTOTYPE_METHOD(tpl, "addOutput", addOutput);
   NODE_SET_PROTOTYPE_METHOD(tpl, "addEventListener", addEventListener);
@@ -144,7 +144,7 @@ void VideoGstAnalyzerWrap::disconnect(const FunctionCallbackInfo<Value>& args){
   me->disconnect(out);
 }
 
-void VideoGstAnalyzerWrap::setOutputParam(const FunctionCallbackInfo<Value>& args){
+void VideoGstAnalyzerWrap::setInputParam(const FunctionCallbackInfo<Value>& args){
   Isolate* isolate = Isolate::GetCurrent();
   HandleScope scope(isolate);
   VideoGstAnalyzerWrap* obj = ObjectWrap::Unwrap<VideoGstAnalyzerWrap>(args.Holder());
@@ -170,7 +170,7 @@ void VideoGstAnalyzerWrap::setOutputParam(const FunctionCallbackInfo<Value>& arg
   String::Utf8Value param7(args[6]->ToString());
   std::string pluginName = std::string(*param7);
 
-  me->setOutputParam(codec,width,height,framerateFPS,bitrateKbps,
+  me->setInputParam(codec,width,height,framerateFPS,bitrateKbps,
                        keyFrameIntervalSeconds,algorithm,pluginName);
 }
 
diff --git a/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.h b/source/agent/analytics/videoGstPipeline/VideoGstAnalyzerWrap.h
diff --git a/source/agent/plugins/samples/cpu_pipeline/mypipeline.cc b/source/agent/plugins/samples/cpu_pipeline/mypipeline.cc
diff --git a/source/agent/plugins/samples/cpu_pipeline/mypipeline.h b/source/agent/plugins/samples/cpu_pipeline/mypipeline.h
