【2022ModelBox客流分析活动打卡】区域内客流统计作业2

题目描述

2、综合课程1/3的内容，将课程3改成区域内客流统计（300分）

完成要求：需要贴出改进后的核心代码和程序运行效果截图，效果如下：

核心代码

draw_passenger_bbox.toml 文件
workspace\passenger_flow\etc\flowunit\draw_passenger_bbox\draw_passenger_bbox.toml

# Copyright (c) Huawei Technologies Co., Ltd. 2022. All rights reserved.

                

                # Basic config
[base]

                name = "draw_passenger_bbox"                              # 功能单元名称

                device = "cpu"                                            # 功能单元运行的设备类型，Python功能单元仅支持cpu类型

                version = "1.0.0"                                         # 功能单元版本号

                type = "python"                                           # 功能单元类型，Python功能单元此处为固定值python

                group_type = "generic"                                    # 功能单元分组信息, 包括input/output/image/video/inference/generic等，默认为generic

                description = "draw passenger bounding boxes for frame"   # 功能单元的功能描述信息

                entry = "draw_passenger_bbox@draw_passenger_bboxFlowUnit" # Python功能单元入口

                

                # 工作模式，以下配置项默认全为false，表示通用功能单元；且配置项之间互斥，即最多只能设置其中一个为true

                stream = false    # 是否是Stream类型功能单元

                condition = false # 是否是条件功能单元

                collapse = false  # 是否是合并功能单元

                expand = false    # 是否是展开功能单元

                

                # 自定义的配置项：行人检测框的类别编号
[config]

                resolution = ["1920", "1080"]  # 视频分辨率

                #line = ["0.2", "0.4", "0.7", "0.4"] # 过线统计的线段，归一化到[0,1]的端点坐标，限制为水平线段

                area = ["0.25", "0.27", "0.63", "0.27", "0.83", "0.74", "0.05", "0.74"]

                # 客流统计的划定区域端点坐标 [每两个为一个坐标点，左上，右上，右下，左下] ["500", "300", "1200", "300", "1600", "800", "100", "800"]

                font_path = "${HILENS_APP_ROOT}/data/simsun.ttc"

                

                # 输入端口描述
[input]
[input.input1] # 输入数据1：uint8格式的原图

                name = "in_image"

                type = "uint8"

                
[input.input2] # 输入数据2：json字符串格式的跟踪结果

                name = "in_track"

                type = "string"

                

                # 输出端口描述
[output]
[output.output1] # 输出数据：画上检测框的uint8格式的图片

                name = "out_image"

                type = "uint8"

draw_passenger_bbox.py 文件
workspace\passenger_flow\etc\flowunit\draw_passenger_bbox\draw_passenger_bbox.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

                
# Copyright (C) 2022 Huawei Technologies Co., Ltd. All rights reserved.

                
import _flowunit as modelbox
import cv2
import json
import numpy as np
from PIL import Image, ImageDraw, ImageFont

                

                
class draw_passenger_bboxFlowUnit(modelbox.FlowUnit):
def __init__(self):
super().__init__()

                
def open(self, config):
# 获取功能单元的配置参数
        
                resolution = config.get_int_list('resolution', [])
#line = config.get_float_list('line', [])
#self.line = [int(line[ix] * resolution[ix % 2]) for ix in range(4)]

                
        
                area = config.get_float_list('area', [])
        
                self.area = [int(area[ix] * resolution[ix % 2]) for ix in range(8)]
        
                modelbox.info(f'self.area = {self.area}')
        
                self.area = np.array(self.area).reshape(-1, 1, 2).astype(np.int32)
        
                modelbox.info(f'self.area = {self.area}')

                
        
                self.font_path = config.get_string("font_path")
return modelbox.Status.StatusCode.STATUS_SUCCESS

                
def process(self, data_context):
# 从DataContext中获取输入输出BufferList对象
        
                in_image = data_context.input("in_image")
        
                in_track = data_context.input("in_track")
        
                out_image = data_context.output("out_image")

                
# 循环处理每一个输入Buffer数据
for buffer_img, buffer_track in zip(in_image, in_track):
# 输入图像Buffer获取宽、高、通道数等属性信息
            
                width = buffer_img.get('width')
            
                height = buffer_img.get('height')
            
                channel = buffer_img.get('channel')

                
# 将输入Buffer转换为Python对象
            
                img_data = np.array(buffer_img.as_object(), copy=False)
            
                img_data = img_data.reshape((height, width, channel))

                
# 获取json字符串格式的跟踪结果
            
                track_json = buffer_track.as_object()

                
# 业务处理：从前一个功能单元的json数据中获取检测框，画在图像上
            
                flow_count, tracking_objects = self.decode_track_info(track_json, (height, width))
            
                img_out = self.draw_tracking_object(img_data, flow_count, tracking_objects)

                
# 将画框后的图像转换为Buffer
            
                out_buffer = modelbox.Buffer(self.get_bind_device(), img_out)

                
# 设置输出Buffer的Meta信息，此处直接拷贝输入Buffer的Meta信息
            
                out_buffer.copy_meta(buffer_img)

                
# 将输出Buffer放入输出BufferList中
            
                out_image.push_back(out_buffer)

                
# 返回成功标志，ModelBox框架会将数据发送到后续的功能单元
return modelbox.Status.StatusCode.STATUS_SUCCESS

                
def decode_track_info(self, track_json, input_shape):
'''从json数据中解码出检测框'''
try:
            
                track_result = json.loads(track_json)
            
                flow_count = track_result['passenger_flow']

                
            
                tracking_objects = []
            
                tracking_list = json.loads(track_result['tracking_objects'])
for track_dict in tracking_list:
                
                tracking_obj = {}
                
                tracking_obj["id"] = track_dict["id"]
                
                x1 = int(track_dict["bbox"][0] * input_shape[1])
                
                y1 = int(track_dict["bbox"][1] * input_shape[0])
                
                x2 = int(track_dict["bbox"][2] * input_shape[1])
                
                y2 = int(track_dict["bbox"][3] * input_shape[0])
                
                tracking_obj["bbox"] = [x1, y1, x2, y2]
                
                tracking_obj["passline"] = track_dict["passline"]
                
                tracking_objects.append(tracking_obj)
except Exception as ex:
            
                modelbox.error(str(ex))
return None, None
else:
return flow_count, tracking_objects

                
def draw_tracking_object(self, img_data, flow_count, tracking_objects):
'''在图中画出跟踪对象的检测框和过线的行人数据'''

                
        
                GRAY = (117, 117, 117)
        
                BLUE = (0, 0, 255)
        
                YELLO = (255, 255, 0)
        
                RED = (255, 0, 0)
        
                GREEN = (0, 255, 0)
        
                thickness = 2
        
                flow_count = 0
for track in tracking_objects:

                
#人形框的中心点
            
                c_x = int((track["bbox"][0] + track["bbox"][2])/2)
            
                c_y = int((track["bbox"][1]+ track["bbox"][3])/2)#·判断人形框的中心点是否在区域内
            
                flag = cv2.pointPolygonTest(self.area,(c_x,c_y),False)
if flag > 0:
# 区域内人形框使用绿色
                
                flow_count += 1
                
                cv2.rectangle(img_data,(track["bbox"][0],track["bbox"][1]),(track["bbox"][2],track["bbox"][3]),GREEN,2)
else:
#·区域外人形框用灰色
                
                cv2.rectangle(img_data,(track["bbox"][0],track["bbox"][1]),(track["bbox"][2],track["bbox"][3]),GRAY,thickness)

                
# color = GRAY
# if track["passline"]:
#     color = BLUE
# cv2.rectangle(img_data, (track["bbox"][0], track["bbox"][1]),
#               (track["bbox"][2], track["bbox"][3]), color, thickness)

                
#cv2.line(img_data, (self.line[0], self.line[1]),
#        (self.line[2], self.line[3]), YELLO, 5)

                
# 画出区域边界线
        
                cv2.polylines(img_data, [self.area], True, YELLO, 3)

                
        
                img_data = self.put_chi_text(
            
                img_data, 'JaneConanの作业2 \t \t \t \t \t \t \t \t 圈定区域客流计数：%d 人' % flow_count, (50, 20), RED, 50) 

                
return img_data

                
def put_chi_text(self, img, text, location, color=(0, 255, 0), size=50):
'''在图片中写汉字'''
if (isinstance(img, np.ndarray)):  # 判断是否OpenCV图片类型
            
                img = Image.fromarray(img)

                
        
                draw = ImageDraw.Draw(img)  # 创建一个可以在给定图像上绘图的对象
        
                font_style = ImageFont.truetype(
            
                self.font_path, size, encoding="utf-8")  # 字体的格式
        
                draw.text(location, text, color, font=font_style)  # 绘制文本
return np.asarray(img)  # 转换回OpenCV格式

                
def close(self):
return modelbox.Status()