1
#摄像头类创建
2
class pi_Camera():
3
    def __init__(self):
4
        # 图像初始化配置
5
        self.Video = cv2.VideoCapture(8, cv2.CAP_V4L2) # 使能摄像头8的驱动
6

7
        # 检查摄像头是否打开
8
        ret = self.Video.isOpened()
9
        if ret:
10
            print("The video is opened.")
11
        else:
12
            print("No video.")
13

14
        codec = cv2.VideoWriter_fourcc('M', 'J', 'P', 'G')
15
        self.Video.set(cv2.CAP_PROP_FOURCC, codec)
16
        self.Video.set(cv2.CAP_PROP_FPS, 60)  # 帧数
17
        self.Video.set(cv2.CAP_PROP_FRAME_WIDTH, 640)  # 列 宽度
18
        self.Video.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)  # 行 高度
19

20
    def GuideLine(self, c1, c2):
21
        ret, image = self.Video.read()#注意：read返回一个bull值和图像数据list！，需要用两个变量获取
22
        if ret:
23
            cv2.line(image, (0, 360), (640, 360), color=(0, 0, 255), thickness=3)  # 红色的线
24
            cv2.line(image, (0, 240), (640, 240), color=(0, 0, 255), thickness=3)  # 红色的线
25
            cv2.line(image, (int(c1), 360), (int(c2), 240), color=(0, 255, 0), thickness=2)  # 绘出倾角线
26
            cv2.imshow("GuideLine", image)

1
# Get HDMI display object
2
disp = srcampy.Display()
3
# For the meaning of parameters, please refer to the relevant documents of HDMI display
4
disp_w, disp_h = get_display_res()
5
disp.display(0, disp_w, disp_h) # 显示模块初始化，并配置显示参数

1
# 创建 FcosPostProcessInfo_t 对象
2
fcos_postprocess_info = FcosPostProcessInfo_t()
3

4
# 设置目标图像的高度和宽度
5
fcos_postprocess_info.height = 512
6
fcos_postprocess_info.width = 512
7

8
# 设置原始图像的高度和宽度
9
fcos_postprocess_info.ori_height = disp_h
10
fcos_postprocess_info.ori_width = disp_w
11

12
# 设置分数阈值，用于过滤低置信度的检测框
13
fcos_postprocess_info.score_threshold = 0.5
14

15
# 设置非极大值抑制（NMS）的阈值，用于合并重叠的检测框
16
fcos_postprocess_info.nms_threshold = 0.6
17

18
# 设置 NMS 阶段保留的检测框数量上限
19
fcos_postprocess_info.nms_top_k = 5
20

21
# 指定是否在图像调整大小时使用填充
22
fcos_postprocess_info.is_pad_resize = 0

1
for i in range(len(models[0].outputs)):
2
    # 设置每个输出张量的布局属性
3
    output_tensors[i].properties.tensorLayout = get_TensorLayout(models[0].outputs[i].properties.layout)
4

5
    # 检查输出张量的 scale_data 是否为空
6
    if len(models[0].outputs[i].properties.scale_data) == 0:
7
        # 如果 scale_data 为空，设置量化类型为 0
8
        output_tensors[i].properties.quantiType = 0
9
    else:
10
        # 如果 scale_data 不为空，设置量化类型为 2
11
        output_tensors[i].properties.quantiType = 2
12
        # 将 scale_data 重新调整形状
13
        scale_data_tmp = models[0].outputs[i].properties.scale_data.reshape(1, 1, 1, models[0].outputs[i].properties.shape[3])
14
        # 将 scale_data 转换为指向浮点数的指针
15
        output_tensors[i].properties.scale.scaleData = scale_data_tmp.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
16

17
    # 设置每个输出张量的有效形状和对齐形状
18
    for j in range(len(models[0].outputs[i].properties.shape)):
19
        output_tensors[i].properties.validShape.dimensionSize[j] = models[0].outputs[i].properties.shape[j]
20
        output_tensors[i].properties.alignedShape.dimensionSize[j] = models[0].outputs[i].properties.shape[j]

1
while True:
2
        _ ,frame = cap.read()
3

4
        # print(frame.shape)
5

6
        if frame is None:
7
            print("Failed to get image from usb camera")
8
        # 把图片缩放到模型的输入尺寸
9
        # 获取算法模型的输入tensor 的尺寸
10
        h, w = models[0].inputs[0].properties.shape[2], models[0].inputs[0].properties.shape[3]
11
        des_dim = (w, h)
12
        resized_data = cv2.resize(frame, des_dim, interpolation=cv2.INTER_AREA)
13

14
        nv12_data = bgr2nv12_opencv(resized_data)
15

16
        t0 = time()
17
        # Forward
18
        outputs = models[0].forward(nv12_data)
19
        t1 = time()
20
        # print("forward time is :", (t1 - t0))
21

22
        # Do post process
23
        strides = [8, 16, 32, 64, 128]
24
        for i in range(len(strides)):
25
            if (output_tensors[i].properties.quantiType == 0):
26
                output_tensors[i].sysMem[0].virAddr = ctypes.cast(outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_float)), ctypes.c_void_p)
27
                output_tensors[i + 5].sysMem[0].virAddr = ctypes.cast(outputs[i + 5].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_float)), ctypes.c_void_p)
28
                output_tensors[i + 10].sysMem[0].virAddr = ctypes.cast(outputs[i + 10].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_float)), ctypes.c_void_p)
29
            else:
30
                output_tensors[i].sysMem[0].virAddr = ctypes.cast(outputs[i].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), ctypes.c_void_p)
31
                output_tensors[i + 5].sysMem[0].virAddr = ctypes.cast(outputs[i + 5].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), ctypes.c_void_p)
32
                output_tensors[i + 10].sysMem[0].virAddr = ctypes.cast(outputs[i + 10].buffer.ctypes.data_as(ctypes.POINTER(ctypes.c_int32)), ctypes.c_void_p)
33

34
            libpostprocess.FcosdoProcess(output_tensors[i], output_tensors[i + 5], output_tensors[i + 10], ctypes.pointer(fcos_postprocess_info), i)
35

36
        result_str = get_Postprocess_result(ctypes.pointer(fcos_postprocess_info))
37
        result_str = result_str.decode('utf-8')
38
        t2 = time()
39
        # print("FcosdoProcess time is :", (t2 - t1))
40
        # print(result_str)
41

42
        # draw result
43
        # 解析JSON字符串
44
        data = json.loads(result_str[14:])
45

46
        if frame.shape[0]!=disp_h or frame.shape[1]!=disp_w:
47
            frame = cv2.resize(frame, (disp_w,disp_h), interpolation=cv2.INTER_AREA)
48

49
        # Draw bboxs
50
        box_bgr = draw_bboxs(frame, data)
51

52
        # cv2.imwrite("imf.jpg", box_bgr)
53

54
        # Convert to nv12 for HDMI display
55
        box_nv12 = bgr2nv12_opencv(box_bgr)
56
        disp.set_img(box_nv12.tobytes())
57

58
        finish_time = time()
59
        image_counter += 1
60
        if finish_time - start_time >  10:
61
            print(start_time, finish_time, image_counter)
62
            print("FPS: {:.2f}".format(image_counter / (finish_time - start_time)))
63
            start_time = finish_time
64
            image_counter = 0