ModelBox
新手入门
【2022 ModelBox实战营】Stream功能单元

发布于30个月以前

  • 1
  • 0
  • 696

发布于30个月以前

ModelBox AI应用开发——Stream功能单元

前面教程介绍了通用Python功能单元、推理功能单元的开发,本文将介绍一类特殊的功能单元——Stream功能单元,我们将在手部检测应用基础上,使用ModelBox开发一个手部检测+跟踪的AI应用:读取摄像头,解码出视频帧,在画面中检测手部并进行跟踪,不同的手将标记不同的跟踪框,再输出画面到本地屏幕。读者将了解到ModelBox中Stream功能单元的基本结构和开发流程。

1. 传统开发模式

对于这样的应用,除了模型推理和前后处理操作外,我们还需要添加跟踪算法,伪代码如下:

import cv2
import onnx_model
import object_tracker

# 初始化摄像头,参数0表示设备的第一个摄像头
cap = cv2.VideoCapture(0)

# 判断初始化是否成功
if not cap.isOpened():
    print('failed to open camera 0')
    exit()

# 初始化手部检测模型
hand_det = onnx_model('./hand_det.onnx')

# 初始化目标跟踪算法
tracker = object_tracker()

while True:
    # 读取一帧视频图像,ret表示读取是否成功
    ret, frame = cap.read()

    # 图像预处理
    img_pre = hand_det.preprocess(frame)

    # 手部检测推理
    output = hand_det.infer(img_pre)

    # 检测结果后处理
    hand_bboxes = hand_det.postprocess(output)

    # 对检测框进行跟踪并画出跟踪框
    tracker.update(hand_bboxes)
    tracker.draw_tracking_object(frame, hand_bboxes)

    # 打开一个名为hand_tracking的窗口,显示图像
    cv2.imshow('hand_tracking', frame)

    # 阻塞等待键盘响应1ms,如果按下q键则跳出循环
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

# 释放摄像头资源
cap.release()

# 关闭所有窗口
cv2.destroyAllWindows()

在这个应用中,开发者需要选择合适的跟踪算法,并封装出对应的接口供主程序调用。

2. ModelBox Pipeline模式

这个应用对应的ModelBox版本已经做成模板放在华为云OBS中,可以用sdk中的solution.bat工具下载,接下来我们给出该应用在ModelBox中的完整开发过程:

1)下载模板

执行.\solution.bat -l可看到当前公开的技能模板:

PS ███> .\solution.bat -l
...

Solutions name:
mask_det_yolo3
...
hand_tracking_yolox
single_hand_pose_yolox_mbv2
multi_hand_pose_yolox_mbv2

结果中的hand_tracking_yolox即为手部检测+跟踪应用模板,可使用如下命令下载模板:

PS ███> .\solution.bat -s hand_tracking_yolox
...

solution.bat工具的参数中,-l 代表list,即列出当前已有的模板名称;-s 代表solution-name,即下载对应名称的模板。下载下来的模板资源,将存放在ModelBox核心库的solution目录下。

2)创建工程

ModelBox sdk目录下使用create.bat创建hand_tracking工程

PS ███> .\create.bat -t server -n hand_tracking -s hand_tracking_yolox
sdk version is modelbox-xxx
success: create hand_tracking in ███\modelbox\workspace

与之前创建工程的命令相比,末尾多了-s参数,表示将使用后面参数值代表的模板创建工程,而不是创建空的工程。
workspace目录下将创建出hand_tracking工程,工程内容如下所示:

hand_tracking
|--bin
│  |--main.bat:应用执行入口
│  |--mock_task.toml:应用在本地执行时的输入输出配置,此应用默认使用本地视频文件为输入源,最终结果输出到另一本地视频文件,可根据需要修改
|--CMake:存放一些自定义CMake函数
|--data:存放应用运行所需要的图片、视频、文本、配置等数据
│  |--hand_test_video.mp4:手部检测+跟踪测试用视频文件
|--dependence
│  |--modelbox_requirements.txt:应用运行依赖的外部库在此文件定义
|--etc
│  |--flowunit:应用所需的功能单元存放在此目录
│  │  |--cpp:存放C++功能单元编译后的动态链接库,此应用没有C++功能单元
│  │  |--object_tracker:目标跟踪功能单元,这是一个Stream功能单元
│  │  |--yolox_post:手部检测使用的是YOLOX模型,此处即为后处理功能单元
|--flowunit_cpp:存放C++功能单元的源代码,此应用没有C++功能单元
|--graph:存放流程图
│  |--hand_tracking.toml:默认流程图,使用本地视频文件作为输入源
│  |--hand_tracking_camera.toml:摄像头输入对应的流程图
│  |--modelbox.conf:modelbox相关配置
|--hilens_data_dir:存放应用输出的结果文件、日志、性能统计信息
|--model:推理功能单元目录
│  |--detect_hand:手部检测推理功能单元
│  │  |--detect_hand.toml:手部检测推理功能单元的配置文件
│  │  |--yolox_hand.onnx:手部检测onnx模型
|--build_project.sh:应用构建脚本
|--CMakeLists.txt
|--rpm:打包rpm时生成的目录,将存放rpm包所需数据
|--rpm_copyothers.sh:rpm打包时的辅助脚本

除了手部检测应用hand_det包含的推理功能单元、检测后处理功能单元外,hand_tracking增加了一个Stream功能单元object_tracker

3)查看Stream功能单元

Stream功能单元的开发流程与通用功能单元是相同的:

a. 创建功能单元

本应用已经包含了Stream功能单元,如果需要创建新的Stream功能单元,创建操作与通用功能单元相同,功能单元目录结构也完全一样(以Python版本的Stream功能单元为例):

[flowunit_name]
|--[flowunit_name].toml    # 功能单元属性配置文件
|--[flowunit_name].py      # 功能单元接口实现文件

b. 设置功能单元属性

打开Stream功能单元的配置文件object_tracker.toml,看到其内容为:

# 基础配置
[base]
name = "object_tracker"                         # 功能单元名称
device = "cpu"                                  # 功能单元运行的设备类型,Python功能单元仅支持cpu类型
version = "1.0.0"                               # 功能单元版本号
type = "python"                                 # 功能单元类型,Python功能单元此处为固定值python
group_type = "generic"                          # 功能单元分组信息, 包括input/output/image/video/inference/generic等,默认为generic
description = "object tracker"                  # 功能单元的功能描述信息
entry = "object_tracker@object_trackerFlowUnit" # Python功能单元入口

# 工作模式,以下配置项默认全为false,表示通用功能单元;且配置项之前互斥,即最多只能设置其中一个为true
stream = true     # 是否是Stream类型功能单元,此处设置为true,即表示object_tracker是一个stream功能单元
condition = false # 是否是条件功能单元
collapse = false  # 是否是合并功能单元
expand = false    # 是否是展开功能单元

# 自定义的配置项:跟踪算法需要的参数
[config]
labels = ['0']
init_hits = 2
max_age = 2
min_iou = 0.4

# 输入端口描述
[input]
[input.input1] # 输入数据1:uint8格式的原图
name = "in_image"
type = "uint8"

[input.input2] # 输入数据2:json字符串格式的检测框
name = "in_bbox"
type = "string"

# 输出端口描述
[output]
[output.output1] # 输出数据:画上跟踪框的uint8格式的图片
name = "out_image"
type = "uint8"

可以看到,与通用功能单元唯一不同的是,Stream功能单元的base.stream属性设置成了true。

  • 通用功能单元
    通用功能单元在处理数据时,只针对当前数据,并不关心数据之间的关联,也不记录任何状态。对于同一个数据流内的数据,它在一次process调用中并不保证这些数据处理的先后(当然数据处理完毕后会由框架重新收集排序),甚至一次process调用中可能包含来自多个流的数据。例如,图像缩放功能单元(resize)在处理当前图像时,处理过程和输出结果只跟当前输入图像相关,并不需要关心其他图像的状态,因此它是一个通用功能单元。下图展示了通用功能单元的数据流图:

    不同颜色代表了不同数据流,图中两个数据流进入通用功能单元节点进行处理,假设功能单元的batch设置为2,可以看到数据被划分为最多两个一组,多组数据同时开始处理,生成的结果数据与输入一致,并且整理成独立的输出流输出。

  • Stream功能单元
    Stream功能单元在处理数据时,一次process调用中处理的都是来自同一个数据流的数据,且这些数据会顺序的进入process,开发者无需关心数据之前是否是有序的,在process此处,已经由框架保证顺序。例如,object_tracker功能单元在处理当前帧时,需要依赖来自同一数据流的历史跟踪目标,且当前帧一定要在上一帧处理完之后才能处理,因此,object_tracker需要被设置成一个Stream功能单元。下图展示了Stream功能单元的数据流图:

    不同颜色代表了不同数据流,图中两个数据流进入Stream功能单元节点进行处理,假设功能单元的batch设置为2,可以看到数据被划分为最多两个一组,不同的数据流同时处理,同一个数据流的多组batch串行处理,生成的结果数据与输入数量可以不一致。

c. 实现功能单元接口

ModelBox功能单元定义了如下接口,开发者需要根据功能单元类型按需实现:

接口 功能说明 是否必须 使用说明
FlowUnit::open 功能单元初始化 实现功能单元的初始化,资源申请,配置参数获取等
FlowUnit::close 功能单元关闭 实现资源的释放
FlowUnit::process 功能单元数据处理 实现核心的数据处理逻辑
FlowUnit::data_pre 功能单元Stream流开始 实现Stream流开始时的处理逻辑,功能单元属性 base.stream = true 时生效,即Stream功能单元才有此接口
FlowUnit::data_post 功能单元Stream流结束 实现Stream流结束时的处理逻辑,功能单元数据处理类型是base.stream = true 时生效,即Stream功能单元才有此接口

对于Stream功能单元,除了通用功能单元的FlowUnit::openFlowUnit::closeFlowUnit::process三个接口外,Stream功能单元还需要实现FlowUnit::data_preFlowUnit::data_post接口。

  • FlowUnit::data_pre接口
    data_pre接口可用于一个数据流中状态数据的初始化,因此,object_tracker功能单元中,跟踪器的初始化就放在data_pre函数中:
    def data_pre(self, data_context):
        self.tracker = Sort(self.init_hits, self.max_age, self.min_iou)
        return modelbox.Status()

此外,还可以在data_pre阶段设置Stream流级别信息存储在DataContext中,在process或者data_post使用或者更新:

    def data_pre(self, data_context):
        # 保存Stream流级别信息。
        data_context.set_private_string("key", value)
        ...
        return modelbox.Status.StatusCode.STATUS_SUCCESS

    def process(self, data_context):
        ...
        # 获取Stream流级别信息
        value = data_context.get_private_string("key")
        ...
        # 更新数据。
        value = data_context.set_private_string("key")
        ...
        return modelbox.Status.StatusCode.STATUS_SUCCESS

    def data_post(self, data_context):
        # 获取Stream流级别信息
        value = data_context.get_private_string("key")
        ...
        return modelbox.Status.StatusCode.STATUS_SUCCESS

注意:本应用的Stream功能单元object_tracker依赖filterpy库中的卡尔曼滤波器,因此可以看到在工程目录的“dependence/modelbox_requirements.txt”文件中添加了filterpy依赖描述,运行本应用时ModelBox将自动下载安装filterpy.

4)查看流程图

hand_tracking工程graph目录下包含两个流程图,其中与工程同名的hand_tracking.toml流程图,其内容为(以Windows版ModelBox为例):

[driver]
# 功能单元的扫描路径,包含在[]中,多个路径使用,分隔
# ${HILENS_APP_ROOT} 表示当前应用的实际路径
# ${HILENS_MB_SDK_PATH} 表示ModelBox核心库的实际路径
dir = [
    "${HILENS_APP_ROOT}/etc/flowunit",
    "${HILENS_APP_ROOT}/etc/flowunit/cpp",
    "${HILENS_APP_ROOT}/model",
    "${HILENS_MB_SDK_PATH}/flowunit",
]
skip-default = true

[profile]
# 通过配置profile和trace开关启用应用的性能统计
profile = false                       # 是否记录profile信息,每隔60s记录一次统计信息
trace = false                         # 是否记录trace信息,在任务执行过程中和结束时,输出统计信息
dir = "${HILENS_DATA_DIR}/mb_profile" # profile/trace信息的保存位置

[flow]
desc = "hand detection and tracking example using yolox for local video or rtsp video stream" # 应用的简单描述

[graph]
format = "graphviz" # 流程图的格式,当前仅支持graphviz
graphconf = """digraph hand_tracking {
    node [shape=Mrecord]
    queue_size = 4
    batch_size = 1

    # 定义节点,即功能单元及其属性
    input1[type=input, flowunit=input, device=cpu, deviceid=0]
    data_source_parser[type=flowunit, flowunit=data_source_parser, device=cpu, deviceid=0]
    video_demuxer[type=flowunit, flowunit=video_demuxer, device=cpu, deviceid=0]
    video_decoder[type=flowunit, flowunit=video_decoder, device=cpu, deviceid=0, pix_fmt=bgr]
    color_transpose[type=flowunit flowunit=packed_planar_transpose device=cpu deviceid=0]
    normalize[type=flowunit flowunit=normalize device=cpu deviceid=0 standard_deviation_inverse="1., 1., 1."]
    image_resize[type=flowunit, flowunit=resize, device=cpu, deviceid=0, image_width=320, image_height=320]
    detect_hand[type=flowunit, flowunit=detect_hand, device=cpu, deviceid=0]
    yolox_post[type=flowunit, flowunit=yolox_post, device=cpu, deviceid=0]
    object_tracker[type=flowunit, flowunit=object_tracker, device=cpu]
    video_out[type=flowunit, flowunit=video_out, device=cpu, deviceid=0]

    # 定义边,即功能间的数据传递关系
    input1:input -> data_source_parser:in_data
    data_source_parser:out_video_url -> video_demuxer:in_video_url
    video_demuxer:out_video_packet -> video_decoder:in_video_packet
    video_decoder:out_video_frame -> image_resize:in_image
    image_resize:out_image -> color_transpose:in_image
    color_transpose:out_image -> normalize:in_data
    normalize:out_data -> detect_hand:input
    detect_hand:output -> yolox_post:in_feat
    yolox_post:out_data -> object_tracker: in_bbox
    video_decoder:out_video_frame -> object_tracker:in_image
    object_tracker:out_image -> video_out:in_video_frame
}"""

5)查看输入输出配置

查看任务配置文件bin/mock_task.toml,可以看到其中的任务输入和任务输出配置为如下内容::

[input]
type = "url"
url = "${HILENS_APP_ROOT}/data/hand_test_video.mp4"  # 表示输入源为本地视频文件

[output]
type = "local"
url = "${HILENS_APP_ROOT}/hilens_data_dir/hand_test_result.mp4"  # 表示输出为本地视频文件

该流程图在本地运行时的逻辑过程是:data_source_parser解析bin/mock_task.toml文件中输入配置的data/hand_test_video.mp4文件,video_demuxervideo_decoder对该文件进行解码,resizepacked_planar_transposenormalize对原始图像进行缩放、转码、归一化等预处理,然后detect_hand在预处理后的图像上进行手部检测,yolox_post从推理结果中解码出检测框,object_tracker对检测框做跟踪并将跟踪结果画到原始图像上,最后video_out将图像输出到bin/mock_task.toml文件中输出配置的hilens_data_dir/hand_test_result.mp4文件中。

6)用启动脚本执行应用

启动应用前执行.\build_project.sh进行工程构建,该脚本将编译自定义的C++功能单元(本应用不涉及)、将应用运行时会用到的配置文件转码为Unix格式(防止执行过程中的格式错误):

PS ███> .\build_project.sh
...
PS ███>

然后执行.\bin\main.bat运行应用:

PS ███> .\bin\main.bat
...

运行结束后在hilens_data_dir目录下生成了hand_test_result.mp4文件,可以打开查看:

7)查看camera流程图

除了测试视频文件外,我们还可以使用PC自带或者外接的USB摄像头进行手部实时检测+跟踪,对应的流程图为hand_tracking_camera.toml,其内容如下:

[driver]
# 功能单元的扫描路径,包含在[]中,多个路径使用,分隔
# ${HILENS_APP_ROOT} 表示当前应用的实际路径
# ${HILENS_MB_SDK_PATH} 表示ModelBox核心库的实际路径
dir = [
    "${HILENS_APP_ROOT}/etc/flowunit",
    "${HILENS_APP_ROOT}/etc/flowunit/cpp",
    "${HILENS_APP_ROOT}/model",
    "${HILENS_MB_SDK_PATH}/flowunit",
]
skip-default = true

[profile]
# 通过配置profile和trace开关启用应用的性能统计
profile = false                       # 是否记录profile信息,每隔60s记录一次统计信息
trace = false                         # 是否记录trace信息,在任务执行过程中和结束时,输出统计信息
dir = "${HILENS_DATA_DIR}/mb_profile" # profile/trace信息的保存位置

[flow]
desc = "hand detection and tracking example using yolox for local usb camera" # 应用的简单描述

[graph]
format = "graphviz" # 流程图的格式,当前仅支持graphviz
graphconf = """digraph hand_tracking {
    node [shape=Mrecord]
    queue_size = 4
    batch_size = 1

    # 定义节点,即功能单元及其属性
    input1[type=input, flowunit=input, device=cpu, deviceid=0]
    data_source_parser[type=flowunit, flowunit=data_source_parser, device=cpu, deviceid=0]
    local_camera[type=flowunit, flowunit=local_camera, device=cpu, deviceid=0, pix_fmt=bgr, cam_width=1280, cam_height=720]
    color_transpose[type=flowunit flowunit=packed_planar_transpose device=cpu deviceid=0]
    normalize[type=flowunit flowunit=normalize device=cpu deviceid=0 standard_deviation_inverse="1., 1., 1."]
    image_resize[type=flowunit, flowunit=resize, device=cpu, deviceid=0, image_width=320, image_height=320]
    detect_hand[type=flowunit, flowunit=detect_hand, device=cpu, deviceid=0]
    yolox_post[type=flowunit, flowunit=yolox_post, device=cpu, deviceid=0]
    object_tracker[type=flowunit, flowunit=object_tracker, device=cpu]
    video_out[type=flowunit, flowunit=video_out, device=cpu, deviceid=0]

    # 定义边,即功能间的数据传递关系
    input1:input -> data_source_parser:in_data
    data_source_parser:out_video_url -> local_camera:in_camera_packet
    local_camera:out_camera_frame -> image_resize:in_image
    image_resize:out_image -> color_transpose:in_image
    color_transpose:out_image -> normalize:in_data
    normalize:out_data -> detect_hand:input
    detect_hand:output -> yolox_post:in_feat
    yolox_post:out_data -> object_tracker: in_bbox
    local_camera:out_camera_frame -> object_tracker:in_image
    object_tracker:out_image -> video_out:in_video_frame
}"""

hand_tracking.toml相比,这个流程图使用local_camera替换了video_demuxervideo_decoder功能单元,其他部分是一致的。

8)修改输入和输出配置

打开工程目录下bin/mock_task.toml文件,修改其中的任务输入和任务输出配置为如下内容:

[input]
type = "url"
url = "0"  # 表示0号摄像头,即PC自带摄像头,若PC无摄像头需外接USB摄像头

[output]
type = "local"
url = "0:hand_tracking"  # 表示名为```hand_tracking```的本地窗口

即使用编号为0的摄像头(默认为PC自带的摄像头),输出画面显示到名为hand_tracking的本地屏幕窗口中。

9)运行应用

执行.\bin\main.bat camera运行应用,将会自动弹出实时的手部检测+跟踪画面,可以看到不同的手会标记不同编号的跟踪框,同一只手只要别移动太快,跟踪框会一直保持相同编号:

3. 小结

通过本教程,我们学习了ModelBoxStream功能单元的开发,视频AI应用中经常会有类似目标跟踪这类依赖上下帧顺序的功能模块,传统开发模式中,如果程序中没有使用多线程/多进程等并发方式,程序本身就是顺序执行,视频前后帧也是依次处理的,并不需要特别考虑这个问题;但是ModelBox为了应用运行得更加高效,每个功能单元支持并行处理,功能单元间也是异步并发执行,因此需要Stream功能单元处理此类依赖上下帧顺序的问题。

LLM初学者

作者相关内容

【2022 ModelBox实战营】第一个应用
发布于30个月以前
【2022 ModelBox实战营】通用Python功能单元
发布于30个月以前
【ModelBox客流分析实战营】ModelBox端云协同AI开发套件(Windows)SDK安装篇
发布于28个月以前
【2022 ModelBox实战营】推理功能单元
发布于30个月以前
ModelBox开发案例 - 使用YOLO v3做口罩检测
发布于34个月以前

暂无数据

热门内容推荐

ModelArts JupyterLab常见问题解决办法
ModelArts开发者 发布于45个月以前
为医生打造专属数字分身!华为云联合万木健康打造医疗医学科普和患者教育数字人引擎
HWCloudAI 发布于20个月以前
图数据库 | 聊聊超级快的图上多跳过滤查询
弓役是也 发布于23个月以前
ModelArts准备工作_简易版
ModelArts开发者 发布于46个月以前
”智蔗见智·向新而生”广西第二届人工智能大赛baseline使用教程
追乐小王子 发布于31个月以前

暂无数据