PS:要转载请注明出处,本人版权所有。

PS: 这个只是基于《我自己》的理解,

如果和你的原则及想法相冲突,请谅解,勿喷。

前置说明

  本文作为本人csdn blog的主站的备份。(BlogID=104)

环境说明
  • Ubuntu 18.04
  • gcc version 7.5.0 (Ubuntu 7.5.0-3ubuntu1~18.04)
  • RK3399PRO 板卡

前言


  本文有一篇前置文章为《Mediapipe 在RK3399PRO上的初探(一)(编译、运行CPU和GPU Demo, RK OpenglES 填坑,编译bazel)》 https://blog.csdn.net/u011728480/article/details/115838306 ,本文是在前置文章上的一点点探索与发现。

  在前置文章中,我们介绍了一些编译和使用Mediapipe的注意事项,也初步跑起来了一点点demo,算是对这个框架有了一个初步的认知。但是前置文章也说了,了解这个框架的意义是替换我们小组现有的框架,而且能够支撑我们的板卡产品。于是我们还需要一个最最最最最最重要的功能就是“Custormer Calculator”,就是自定义计算节点,因为这个框架的核心就是计算节点。下文我们将会讲到这个框架的一些基本概念,这些概念都是来至于官方文档的机翻+我自己的理解。同时,我会给出一个我定义的计算节点的实例,算是给大家一个感性认知。





Mediapipe的一些概念(本小节基本来至于官方文档的机翻+我自己的理解,不感兴趣,请直接看下一小节)


  本小节内容,主要参考 https://github.com/google/mediapipe/tree/master/docs/framework_concepts 的几个介绍概念的md文件。
  MediaPipe 感觉中文直译为“媒体管道”,为啥会有这个名字呢?因为它把数据+处理组合成一个计算节点,然后把一个一个节点连接起来,用数据来驱动整个核心逻辑。如果大家对Caffe、ncnn类似的框架源码有一点点了解的话,就会觉得他们非常像,但是又不像。像是说,都是通过配置文件定义计算节点逻辑图,然后通过运算,得到我们想得到的逻辑图中节点的数据。不像的话,就是说的是Mediapipe的调度机制了,极大的增加了节点计算的并行功能,而那些框架是按照图节点的上下顺序进行执行的。
  上面这段话可能有点抽象,我想表达的就是 Mediapipe 就是把任何一个“操作”都可以变为一个Calculator,因为我们的每一个项目的逻辑抽象出来都是 Calculator0+Data0->Calculator1+Data1->Calculator2+Data2->... ...,然后,Mediapipe 做的是基于这种calculator的调度和执行。这里我举个栗子:

  • 人脸图+检测算法=人脸检测结果,这是一个Calculator
  • RTSP流+解码模块=解码之后的图片,这是一个Calculator

好了,其他的就不过分的解读了,下面就使用MediaPipe的helloworld example( https://github.com/google/mediapipe/blob/master/mediapipe/examples/desktop/hello_world/hello_world.cc )为例,简单的说说以下几个概念。



Packet

  Packet,是mediapipe中的数据单元,它可以接收任意类型的数据。也是mediapipe中的数据流动单元。就是在mediapipe中,我们设计的Graph中,所有的逻辑流动都是通过packet流动来实现的。

  实例代码片段:

//MakePacket<std::string>("Hello World!") 创建一个packet,顺带说一句,我不喜欢这里的宏,不利于维护
MP_RETURN_IF_ERROR(graph.AddPacketToInputStream("in", MakePacket<std::string>("Hello World!").At(Timestamp(i))));

//从packet中获取数据
mediapipe::Packet packet;
packet.Get<std::string>();


Graph

  Graph是由各个Calculator组成的,可以直接把Calculator理解为数据结构中图的节点。而Graph直接把他当做图就行了。Graph是我们定义的逻辑流程的具体载体,也就是说我们的业务逻辑是什么样子的,那么Graph里面就会有相应的逻辑流程。可以具备多输入输出。

  实例代码片段:

CalculatorGraph graph;


Caculator(Node)

  上面不是介绍了Graph和Packet嘛,这里的Calculator就是Graph里面的节点,也是处理Packet的具体单元。可以具备多输入输出。

  实例代码片段:

//比如mediapipe/calculators/core/pass_through_calculator.h里面的定义,这个Calculator被Helloworld这个例子使用,作用就是把输入的数据直接传递到输出,不做任何处理,类似NOP
class PassThroughCalculator : public CalculatorBase


Stream

  Stream 就是 Caculator 之间的连接起来后,形成的一个数据流动路径。



Side packets

  Side packets 可以直接理解为一些静态的数据packet,在graph创建之后就不会改变的数据。

... ...





自定义实现Calculator


  Talk is cheap, show me the code.

/*
 * @Description:
 * @Author: Sky
 * @Date:
 * @LastEditors: Sky
 * @LastEditTime:
 * @Github:
 */
#include <cstdio>
#include "mediapipe/framework/calculator_graph.h"
#include "mediapipe/framework/port/logging.h"
#include "mediapipe/framework/port/parse_text_proto.h"
#include "mediapipe/framework/port/status.h"

//customer calculator
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/port/canonical_errors.h"


class CustomerDataType{

  public:
  CustomerDataType(int i, float f, bool b, const std::string & str):
  val_i(i),
  val_f(f),
  val_b(b),
  s_str(str)
  {}
  int val_i = 1;
  float val_f = 11.f;
  bool val_b = true;
  std::string s_str = "customer str.";
};



namespace mediapipe {

  class MyStringProcessCalculator : public CalculatorBase {
  public:
    /*
    Calculator authors can specify the expected types of inputs and outputs of a calculator in GetContract().
    When a graph is initialized, the framework calls a static method to verify if the packet types of the connected inputs and outputs match the information in this specification.
    */
    static absl::Status GetContract(CalculatorContract* cc) {

      /*
      class InputStreamShard;
      typedef internal::Collection<InputStreamShard> InputStreamShardSet;
      class OutputStreamShard;
      typedef internal::Collection<OutputStreamShard> OutputStreamShardSet;
      */
      //cc->Inputs().NumEntries() returns the number of input streams
      // if (!cc->Inputs().TagMap()->SameAs(*cc->Outputs().TagMap())) {

      //   return absl::InvalidArgumentError("Input and output streams's TagMap can't be same.");
      // }

      //set stream
      // for (CollectionItemId id = cc->Inputs().BeginId(); id < cc->Inputs().EndId(); ++id) {

      //   cc->Inputs().Get(id).SetAny();
      //   cc->Outputs().Get(id).SetSameAs(&cc->Inputs().Get(id));
      // }
      cc->Inputs().Index(0).SetAny();
      cc->Inputs().Index(1).Set<CustomerDataType>();

      cc->Outputs().Index(0).SetSameAs(&cc->Inputs().Index(0));


      //set stream package
      // for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

      //   cc->InputSidePackets().Get(id).SetAny();
      // }
      // cc->InputSidePackets().Index(0).SetAny();
      // cc->InputSidePackets().Index(1).Set<CustomerDataType>();//set customer data-type


      if (cc->OutputSidePackets().NumEntries() != 0) {

        // if (!cc->InputSidePackets().TagMap()->SameAs(*cc->OutputSidePackets().TagMap())) {

        //   return absl::InvalidArgumentError("Input and output side packets's TagMap can't be same.");
        // }

        // for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

        //   cc->OutputSidePackets().Get(id).SetSameAs(&cc->InputSidePackets().Get(id));
        // }

        cc->OutputSidePackets().Index(0).SetSameAs(&cc->InputSidePackets().Index(0));
      }

      return absl::OkStatus();
    }

    absl::Status Open(CalculatorContext* cc) final {

      for (CollectionItemId id = cc->Inputs().BeginId();id < cc->Inputs().EndId(); ++id) {

        if (!cc->Inputs().Get(id).Header().IsEmpty()) {

          cc->Outputs().Get(id).SetHeader(cc->Inputs().Get(id).Header());
        }
      }

      if (cc->OutputSidePackets().NumEntries() != 0) {

        for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

          cc->OutputSidePackets().Get(id).Set(cc->InputSidePackets().Get(id));
        }
      }

      // Sets this packet timestamp offset for Packets going to all outputs.
      // If you only want to set the offset for a single output stream then
      // use OutputStream::SetOffset() directly.
      cc->SetOffset(TimestampDiff(0));

      return absl::OkStatus();
    }

    absl::Status Process(CalculatorContext* cc) final {

      if (cc->Inputs().NumEntries() == 0) {
        return tool::StatusStop();
      }

      //get node input data
      mediapipe::Packet  _data0 = cc->Inputs().Index(0).Value();
      mediapipe::Packet  _data1 = cc->Inputs().Index(1).Value();

      //not safety.
      char _tmp_buf[1024];

      ::memset(_tmp_buf, 0, 1024);

      snprintf(_tmp_buf, 1024, _data0.Get<std::string>().c_str(), _data1.Get<CustomerDataType>().val_i, _data1.Get<CustomerDataType>().val_f, _data1.Get<CustomerDataType>().val_b, _data1.Get<CustomerDataType>().s_str.c_str());

      std::string _out_data = _tmp_buf;
      cc->Outputs().Index(0).AddPacket(MakePacket<std::string>(_out_data).At(cc->InputTimestamp()));

      return absl::OkStatus();
    }

    absl::Status Close(CalculatorContext* cc) final {

      return absl::OkStatus();
    }
  };

  REGISTER_CALCULATOR(MyStringProcessCalculator);
}




namespace mediapipe {

  absl::Status  RunMyGraph() {

    // Configures a simple graph, which concatenates 2 PassThroughCalculators.
    CalculatorGraphConfig config = ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
      input_stream: "in"
      input_stream: "customer_in"
      output_stream: "out"
      node {
        calculator: "PassThroughCalculator"
        input_stream: "in"
        output_stream: "out1"
      }
      node {
        calculator: "MyStringProcessCalculator"
        input_stream: "out1"
        input_stream: "customer_in"
        output_stream: "out2"
      }
      node {
        calculator: "PassThroughCalculator"
        input_stream: "out2"
        output_stream: "out"
      }
    )");
    LOG(INFO)<<"parse graph cfg-str done ... ...";

    CalculatorGraph graph;
    MP_RETURN_IF_ERROR(graph.Initialize(config));
    LOG(INFO)<<"init graph done ... ...";

    ASSIGN_OR_RETURN(OutputStreamPoller poller,
                    graph.AddOutputStreamPoller("out"));
    LOG(INFO)<<"add out-node to output-streampoller done ... ...";


    MP_RETURN_IF_ERROR(graph.StartRun({}));
    LOG(INFO)<<"start run graph done ... ...";


    // Give 10 input packets that contains the same std::string "Hello World!".
    for (int i = 0; i < 10; ++i) {

      MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
          "in", MakePacket<std::string>("CustomerCalculator: val_i %d, val_f %f, val_b %d, val_str %s").At(Timestamp(i))));

      MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
          "customer_in", MakePacket<CustomerDataType>(i, i + 1.f, i%2==0, "s" + std::to_string(i)).At(Timestamp(i))));
    }
    // Close the input stream "in".
    MP_RETURN_IF_ERROR(graph.CloseInputStream("in"));
    MP_RETURN_IF_ERROR(graph.CloseInputStream("customer_in"));

    mediapipe::Packet packet;
    // Get the output packets std::string.
    while (poller.Next(&packet)) {
      LOG(INFO) << packet.Get<std::string>();
    }

    LOG(INFO)<<"RunGraph Done";
    return graph.WaitUntilDone();
  }

}  // namespace mediapipe



int main(int argc, char** argv) {

  gflags::ParseCommandLineFlags(&argc, &argv, true);

  FLAGS_minloglevel = 0;
  FLAGS_stderrthreshold = 0;
  FLAGS_alsologtostderr = 1;

  google::InitGoogleLogging(argv[0]);

  LOG(INFO) << "glog init success ... ...";

  absl::Status run_status = mediapipe::RunMyGraph();
  if (!run_status.ok())
    LOG(ERROR) << "Failed to run the graph: " << run_status.message();

  google::ShutdownGoogleLogging();
  return 0;
}

下面简单介绍这段代码。



自定义Calculator:MyStringProcessCalculator

   这里自定义了一个Calculator,主要作用就是传入snprintf的fmt字符串和fmt字符串所需要的数据。所以可以看到有两个输入,一个是string,一个是我自定义的data-type。输出是一个格式化之后的字符串,所以输出是string。

   自定义Calculator主要还是实现4个接口,分别是GetContract,Open,Process,Close。其中GetContract是Graph初始化的时候,检查Calculator用的。Open接口是在Graph开始后,对Calculator做一些初始化工作,例如设定一些Calculator初始状态等。Process是实际的Calculator功能。

namespace mediapipe {

  class MyStringProcessCalculator : public CalculatorBase {
  public:
    /*
    Calculator authors can specify the expected types of inputs and outputs of a calculator in GetContract().
    When a graph is initialized, the framework calls a static method to verify if the packet types of the connected inputs and outputs match the information in this specification.
    */
    static absl::Status GetContract(CalculatorContract* cc) {

      /*
      class InputStreamShard;
      typedef internal::Collection<InputStreamShard> InputStreamShardSet;
      class OutputStreamShard;
      typedef internal::Collection<OutputStreamShard> OutputStreamShardSet;
      */
      //cc->Inputs().NumEntries() returns the number of input streams
      // if (!cc->Inputs().TagMap()->SameAs(*cc->Outputs().TagMap())) {

      //   return absl::InvalidArgumentError("Input and output streams's TagMap can't be same.");
      // }

      //set stream
      // for (CollectionItemId id = cc->Inputs().BeginId(); id < cc->Inputs().EndId(); ++id) {

      //   cc->Inputs().Get(id).SetAny();
      //   cc->Outputs().Get(id).SetSameAs(&cc->Inputs().Get(id));
      // }
      cc->Inputs().Index(0).SetAny();
      cc->Inputs().Index(1).Set<CustomerDataType>();

      cc->Outputs().Index(0).SetSameAs(&cc->Inputs().Index(0));


      //set stream package
      // for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

      //   cc->InputSidePackets().Get(id).SetAny();
      // }
      // cc->InputSidePackets().Index(0).SetAny();
      // cc->InputSidePackets().Index(1).Set<CustomerDataType>();//set customer data-type


      if (cc->OutputSidePackets().NumEntries() != 0) {

        // if (!cc->InputSidePackets().TagMap()->SameAs(*cc->OutputSidePackets().TagMap())) {

        //   return absl::InvalidArgumentError("Input and output side packets's TagMap can't be same.");
        // }

        // for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

        //   cc->OutputSidePackets().Get(id).SetSameAs(&cc->InputSidePackets().Get(id));
        // }

        cc->OutputSidePackets().Index(0).SetSameAs(&cc->InputSidePackets().Index(0));
      }

      return absl::OkStatus();
    }

    absl::Status Open(CalculatorContext* cc) final {

      for (CollectionItemId id = cc->Inputs().BeginId();id < cc->Inputs().EndId(); ++id) {

        if (!cc->Inputs().Get(id).Header().IsEmpty()) {

          cc->Outputs().Get(id).SetHeader(cc->Inputs().Get(id).Header());
        }
      }

      if (cc->OutputSidePackets().NumEntries() != 0) {

        for (CollectionItemId id = cc->InputSidePackets().BeginId(); id < cc->InputSidePackets().EndId(); ++id) {

          cc->OutputSidePackets().Get(id).Set(cc->InputSidePackets().Get(id));
        }
      }

      // Sets this packet timestamp offset for Packets going to all outputs.
      // If you only want to set the offset for a single output stream then
      // use OutputStream::SetOffset() directly.
      cc->SetOffset(TimestampDiff(0));

      return absl::OkStatus();
    }
    //这里是整个Calculator的核心,就是调用snprintf
    absl::Status Process(CalculatorContext* cc) final {

      if (cc->Inputs().NumEntries() == 0) {
        return tool::StatusStop();
      }

      //get node input data
      mediapipe::Packet  _data0 = cc->Inputs().Index(0).Value();
      mediapipe::Packet  _data1 = cc->Inputs().Index(1).Value();

      //not safety.
      char _tmp_buf[1024];

      ::memset(_tmp_buf, 0, 1024);

      snprintf(_tmp_buf, 1024, _data0.Get<std::string>().c_str(), _data1.Get<CustomerDataType>().val_i, _data1.Get<CustomerDataType>().val_f, _data1.Get<CustomerDataType>().val_b, _data1.Get<CustomerDataType>().s_str.c_str());

      std::string _out_data = _tmp_buf;
      cc->Outputs().Index(0).AddPacket(MakePacket<std::string>(_out_data).At(cc->InputTimestamp()));

      return absl::OkStatus();
    }

    absl::Status Close(CalculatorContext* cc) final {

      return absl::OkStatus();
    }
  };

  REGISTER_CALCULATOR(MyStringProcessCalculator);
}


然后开始编译运行得到结果

   编译。


# 注意,这里的--check_visibility=false 为了关闭bazel关于target之间的可见性检查,因为我的Calculator自定义放在我自己的目录的,有一个target对这个目录不可见,编译会报错。

bazel build -c dbg --define MEDIAPIPE_DISABLE_GPU=1 --copt -DMESA_EGL_NO_X11_HEADERS --copt -DEGL_NO_X11 my_target --check_visibility=false --verbose_failures  --local_cpu_resources=1

   然后运行。得到如下图的结果。





后记


  好了,一个超级简单的自定义calculator已经实现了,相信你已经明白了吧。本系列也就此终结吧,以后随缘更新。




PS: 请尊重原创,不喜勿喷。

PS: 要转载请注明出处,本人版权所有。

PS: 有问题请留言,看到后我会第一时间回复。

04-25 02:28