基于车形检测的划区域车流统计

在本案例中我们使用客流统计的相关知识开发一个“车流统计”技能。

技能开发

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

1）下载模板

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

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

Solutions name:
mask_det_yolo3
...
vehicle_flow_centernet

结果中的vehicle_flow_centernet即为基于车形检测的划区域车流统计应用模板，可使用如下命令下载模板：

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

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

2）创建工程

在ModelBox sdk目录下使用create.bat创建vehicle_flow工程:

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

create.bat工具的参数中，-t 表示创建事务的类别，包括工程(server)、Python功能单元(Python)、推理功能单元(infer)等；-n 代表name，即创建事务的名称；-s 代表solution-name，表示将使用后面参数值代表的模板创建工程，而不是创建空的工程。

workspace目录下将创建出vehicle_flow工程，工程内容如下所示：

vehicle_flow
|--bin
│  |--main.bat：应用执行入口
│  |--mock_task.toml：应用在本地执行时的输入输出配置，此应用默认使用本地视频文件为输入源，最终结果输出到屏幕，可根据需要修改
|--CMake：存放一些自定义CMake函数
|--data：存放应用运行所需要的图片、视频、文本、配置等数据
│  |--vehicle.mp4：车流统计测试用视频文件
|--dependence
│  |--modelbox_requirements.txt：应用运行依赖的外部库在此文件定义，本应用依赖lap、scipy等工具包
|--etc
│  |--flowunit：应用所需的功能单元存放在此目录
│  │  |--cpp：存放C++功能单元编译后的动态链接库，此应用没有C++功能单元
│  │  |--condition：条件功能单元，判断是否检测到车辆
│  │  |--det_post：车辆检测使用的是centernet模型，此处即为后处理功能单元
│  │  |--draw_track_bbox：车流画图功能单元
│  │  |--letter_resize：车辆检测使用的是centernet模型，此处即为预处理功能单元
│  │  |--object_tracker：目标跟踪功能单元
|--flowunit_cpp：存放C++功能单元的源代码，此应用没有C++功能单元
|--graph：存放流程图
│  |--vehicle_flow.toml：默认流程图，使用本地视频文件作为输入源
│  |--modelbox.conf：modelbox相关配置
|--hilens_data_dir：存放应用输出的结果文件、日志、性能统计信息
|--model：推理功能单元目录
│  |--vehicle_det：车辆检测推理功能单元
│  │  |--vehicle_det.toml：车辆检测推理功能单元的配置文件
│  │  |--vehicle_det_320x576.onnx：车辆检测onnx模型
|--build_project.sh：应用构建脚本
|--CMakeLists.txt
|--rpm：打包rpm时生成的目录，将存放rpm包所需数据
|--rpm_copyothers.sh：rpm打包时的辅助脚本

3）查看流程图

vehicle_flow工程graph目录下存放流程图，默认的流程图vehicle_flow.toml与工程同名，其内容为（以Windows版ModelBox为例）：

# 功能单元的扫描路径，包含在[]中，多个路径使用,分隔
# ${HILENS_APP_ROOT} 表示当前应用的实际路径
# ${HILENS_MB_SDK_PATH} 表示ModelBox核心库的实际路径
[driver]
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信息的保存位置

[graph]
format = "graphviz"  # 流程图的格式，当前仅支持graphviz
graphconf = """digraph vehicle_flow {
    node [shape=Mrecord]
    queue_size = 1
    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="rgb"]
    letter_resize[type=flowunit, flowunit=letter_resize, device=cpu]
    color_transpose[type=flowunit, flowunit=packed_planar_transpose, device=cpu, deviceid=0]
    normalize[type=flowunit, flowunit=normalize, device=cpu, deviceid=0, standard_deviation_inverse="0.003921568627451, 0.003921568627451, 0.003921568627451"]
    vehicle_det[type=flowunit, flowunit=vehicle_det, device=cpu, deviceid=0]
    det_post[type=flowunit, flowunit=det_post, device=cpu, deviceid=0]
    object_tracker[type=flowunit, flowunit=object_tracker, device=cpu, deviceid=0]
    condition[type=flowunit, flowunit=condition, device=cpu, deviceid=0]
    draw_track_bbox[type=flowunit, flowunit=draw_track_bbox, device=cpu, deviceid=0]
    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 -> letter_resize:in_image
    letter_resize:resized_image -> color_transpose:in_image
    color_transpose:out_image -> normalize:in_data
    normalize:out_data -> vehicle_det:input
    vehicle_det:output -> det_post:in_feat
    letter_resize:out_image -> det_post:in_image
    det_post:out_feat -> object_tracker:in_feat
    object_tracker:out_track -> condition:in_track
    video_decoder:out_video_frame -> condition:in_image

    condition:out_track -> draw_track_bbox:in_image
    draw_track_bbox:out_image -> video_out:in_video_frame
    condition:out_image -> video_out:in_video_frame
}"""

[flow]
desc = "vehicle_flow run in modelbox-win10-x64"

整个应用逻辑比较简单，视频解码后做图像预处理，接着是车辆检测，模型后处理得到车形框与128维车辆reid特征，送入跟踪算法进行实时跟踪，经过条件功能单元判断，检测到车辆的图送入画图功能单元进行区域内外判断与跟踪信息绘制，最终结果输出到屏幕。

4）核心逻辑

本应用核心逻辑跟踪与区域判断参照客流统计应用设计，跟踪逻辑在object_tracker功能单元中，使用的是JDE（Towards Real-Time Multi-object Tracking）算法，算法介绍可参考论文，本应用在客流统计案例基础上加入了车辆reid特征值进行跟踪匹配。

区域判断在object_tracker功能单元中，为了更贴合实际使用场景，本应用在采用任务参数的形式配置感兴趣区域。在mock_task.toml中配置划区域任务参数：

[common]
content = "{\"task\":\"area\", \"area\":\"787,678,1572,652,1800,795,977,853\"}"

同时，本应用也支持配置过线参数：

[common]
content = "{\"task\":\"line\", \"line\":\"720\"}"

在object_tracker功能单元的data_pre函数中获取配置的任务参数：

def data_pre(self, data_context):
    iva_params = json.loads(data_context.get_session_config().get_string("iva_task_common"))
    self.task = None
    if iva_params:
        self.task = iva_params.get('task')
        if self.task:
            if self.task == "area":
                self.area = iva_params.get(self.task)
                self.area = np.array(list(map(int, self.area.split(",")))).reshape(1, -1, 2)
            elif self.task == "line":
                self.area = float(iva_params.get(self.task))
            else:
                modelbox.info("Incorrect task type")
                self.task = None

为跟踪对象增设isPassline属性，默认为False:

def get_tracking_objects(self, online_targets_dict):
    tracking_objects = {}
    for cls_id in range(self.num_classes):
        online_targets = online_targets_dict[cls_id]
        for t in online_targets:
            obj = {}
            tlwh = t.tlwh
            if tlwh[2] * tlwh[3] < self.min_box_area:
                continue
            tid = t.track_id
            obj["bbox"] = [max(0, tlwh[0]), max(0, tlwh[1]), tlwh[0] + tlwh[2], tlwh[1] + tlwh[3]]
            obj["isPassline"] = False
            obj["bbox_score"] = t.score
            tracking_objects[tid] = obj
    return tracking_objects

在check_passline函数中进行过线判断：

def check_passline(self, tracking_objects):
    for idx, obj in tracking_objects.items():
        box = list(map(int, obj["bbox"]))
        if self.task == "area" and np.any(self.area):
            c_x = int((box[0] + box[2]) / 2)
            c_y = int((box[1] + box[3]) / 2)
            flag = cv2.pointPolygonTest(self.area, (c_x, c_y), False)
            if flag > 0:
                obj["isPassline"] = True
        elif self.task == "line" and np.any(self.area):
            if box[-1] > self.area:
                obj["isPassline"] = True

可以看到，我们的划区域车流统计共有两种任务类型，针对过线任务我们采用车辆检测框下边沿作为过线判断；针对划区域任务我们使用OpenCV的 pointPolygonTest 函数判断车辆检测框中心点与区域的位置关系作为驶入判断。

在draw_track_bbox功能单元draw_bbox函数中，我们根据不同的过线状态绘制不同的车辆检测框颜色，过线/进入划定区域的车绘制为灰色，其余车根据id赋色：

def draw_bbox(self, img_data, tracking_objects):
    for idx, track in tracking_objects.items():
        box = list(map(int, track["bbox"]))
        score = round(track["bbox_score"], 2)
        color = self.gray if track.get("isPassline") else self.get_color(float(idx))
        cv2.rectangle(img_data, (box[0], box[1]), (box[2], box[3]), color, self.thickness)
        cv2.putText(img_data, f"id:{idx}, score:{score}", (box[0], box[1]), cv2.FONT_HERSHEY_SIMPLEX, 0.8, color=color, thickness=self.thickness)

后续如需对过线/进入划定区域的车进行其他处理，也可在此处进行操作。

5）三方依赖库

本应用依赖scipy等工具包，ModelBox应用不需要手动安装三方依赖库，只需要配置在dependence\modelbox_requirements.txt，应用在编译时会自动安装。

6）查看输入输出配置

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

[input]
type = "url"
url = "${HILENS_APP_ROOT}/data/vehicle.mp4"

[output]
type = "local"
url = "0"

即，使用本地视频文件data/vehicle.mp4作为输入，结果输出到屏幕。

7）用启动脚本执行应用

在项目目录下执行.\bin\main.bat运行应用：

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

可以看到屏幕出现车流统计画面：

白线即车流统计的区域，区域外/未过线车辆根据id赋色，区域内/已过线车辆的使用灰色框，可在输入输出配置中修改划区域任务类型与坐标点。

小结

本案例开发了车流统计应用，可用于违停检测等应用。