Mediapipe 在RK3399PRO上的初探（二）(自定义Calculator)

2024-06-16 06:58
文章标签 自定义 初探 mediapipe calculator rk3399pro

本文主要是介绍Mediapipe 在RK3399PRO上的初探（二）(自定义Calculator)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

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

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

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

环境说明
  • 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-typeif (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 datamediapipe::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 mediapipeint 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-typeif (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的核心，就是调用snprintfabsl::Status Process(CalculatorContext* cc) final {if (cc->Inputs().NumEntries() == 0) {return tool::StatusStop();}//get node input datamediapipe::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

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

result




后记

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



打赏、订阅、收藏、丢香蕉、硬币，请关注公众号（攻城狮的搬砖之路）
qrc_img

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

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

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

这篇关于Mediapipe 在RK3399PRO上的初探（二）(自定义Calculator)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1065757。

相关文章

Vite 打包目录结构自定义配置小结

Vite 打包目录结构自定义配置小结

《Vite打包目录结构自定义配置小结》在Vite工程开发中,默认打包后的dist目录资源常集中在asset目录下,不利于资源管理,本文基于Rollup配置原理,本文就来介绍一下通过Vite配置自定义... 目录一、实现原理二、具体配置步骤1. 基础配置文件2. 配置说明（1）js 资源分离（2）非 JS 资
阅读更多...
聊聊springboot中如何自定义消息转换器

聊聊springboot中如何自定义消息转换器

《聊聊springboot中如何自定义消息转换器》SpringBoot通过HttpMessageConverter处理HTTP数据转换,支持多种媒体类型,接下来通过本文给大家介绍springboot中... 目录核心接口springboot默认提供的转换器如何自定义消息转换器Spring Boot 中的消息
阅读更多...
Python自定义异常的全面指南(入门到实践)

Python自定义异常的全面指南(入门到实践)

《Python自定义异常的全面指南(入门到实践)》想象你正在开发一个银行系统,用户转账时余额不足,如果直接抛出ValueError,调用方很难区分是金额格式错误还是余额不足,这正是Python自定义异... 目录引言：为什么需要自定义异常一、异常基础：先搞懂python的异常体系1.1 异常是什么？1.2
阅读更多...
Linux中的自定义协议+序列反序列化用法

Linux中的自定义协议+序列反序列化用法

《Linux中的自定义协议+序列反序列化用法》文章探讨网络程序在应用层的实现,涉及TCP协议的数据传输机制、结构化数据的序列化与反序列化方法,以及通过JSON和自定义协议构建网络计算器的思路,强调分层... 目录一，再次理解协议二，序列化和反序列化三，实现网络计算器3.1 日志文件3.2Socket.hpp
阅读更多...
C语言自定义类型之联合和枚举解读

C语言自定义类型之联合和枚举解读

《C语言自定义类型之联合和枚举解读》联合体共享内存,大小由最大成员决定,遵循对齐规则；枚举类型列举可能值,提升可读性和类型安全性,两者在C语言中用于优化内存和程序效率... 目录一、联合体1.1 联合体类型的声明1.2 联合体的特点1.2.1 特点11.2.2 特点21.2.3 特点31.3 联合体的大小1
阅读更多...
springboot自定义注解RateLimiter限流注解技术文档详解

springboot自定义注解RateLimiter限流注解技术文档详解

《springboot自定义注解RateLimiter限流注解技术文档详解》文章介绍了限流技术的概念、作用及实现方式,通过SpringAOP拦截方法、缓存存储计数器,结合注解、枚举、异常类等核心组件,... 目录什么是限流系统架构核心组件详解1. 限流注解 (@RateLimiter)2. 限流类型枚举 (
阅读更多...
SpringBoot 异常处理/自定义格式校验的问题实例详解

SpringBoot 异常处理/自定义格式校验的问题实例详解

《SpringBoot异常处理/自定义格式校验的问题实例详解》文章探讨SpringBoot中自定义注解校验问题,区分参数级与类级约束触发的异常类型,建议通过@RestControllerAdvice... 目录1. 问题简要描述2. 异常触发1) 参数级别约束2) 类级别约束3. 异常处理1) 字段级别约束
阅读更多...
SpringBoot+EasyExcel实现自定义复杂样式导入导出

SpringBoot+EasyExcel实现自定义复杂样式导入导出

《SpringBoot+EasyExcel实现自定义复杂样式导入导出》这篇文章主要为大家详细介绍了SpringBoot如何结果EasyExcel实现自定义复杂样式导入导出功能,文中的示例代码讲解详细,... 目录安装处理自定义导出复杂场景1、列不固定，动态列2、动态下拉3、自定义锁定行/列，添加密码4、合并
阅读更多...
Java实现自定义table宽高的示例代码

Java实现自定义table宽高的示例代码

《Java实现自定义table宽高的示例代码》在桌面应用、管理系统乃至报表工具中,表格（JTable）作为最常用的数据展示组件,不仅承载对数据的增删改查,还需要配合布局与视觉需求,而JavaSwing... 目录一、项目背景详细介绍二、项目需求详细介绍三、相关技术详细介绍四、实现思路详细介绍五、完整实现代码
阅读更多...
一文详解Java Stream的sorted自定义排序

一文详解Java Stream的sorted自定义排序

《一文详解JavaStream的sorted自定义排序》Javastream中的sorted方法是用于对流中的元素进行排序的方法,它可以接受一个comparator参数,用于指定排序规则,sorte... 目录一、sorted 操作的基础原理二、自定义排序的实现方式1. Comparator 接口的 Lam
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2