环境搭建

安装rk356x提供好的docker镜像,在该镜像中已经安装好了所需要的环境, 我们使用这个环境主要是将模型转换成rknn格式
参考链接：
https://github.com/airockchip/yolov5.git
https://github.com/rockchip-linux/rknpu2
https://github.com/rockchip-linux/rknn-toolkit2

加载docker

  1. 进到环境目录

    1. rknn-toolkit2-1.x.x/docker/docker_full

  2. 加载镜像文件

    1. docker load --input rknn-toolkit2-1.x.x-cp36-docker.tar.gz

  3. 查看是否加载成功

    1. docker images

  4. 创建容器

    1. docker run -it --shm-size 4G -v D:\DATA\ml\rknn-toolkit2-1.5.0\examples\onnx\yolov5:/rknn_yolov5_demo rknn-toolkit2:1.5.0-ml /bin/bash

    其中”-v 主机目录:虚拟机目录” 是将本机目录映射到docker中的目录, 成功进到docker目录之后, 我们就可以来生成模型了

  5. 以后每次启动的时候，启动对应的容器

    1. docker container ls
    2. docker start -ai <container_name>

    可以使用如下命令查看内存使用

    1. df -h

    模型导出

    在docker容器中进入到rknn_yolov5_demo文件夹,也就是上面映射的文件夹

    1. cd /rknn_yolov5_demo

    执行示例代码:

    1. python3 ./test.py

    它会自动将onnx格式的模型转换成rknn格式的模型

    1. yolov5s-640-640.rknn

    yolov5模型训练

    启动训练

    1. python3 train.py --data data/lines/itheima_lines.yaml --weights yolov5s.pt  --img 640

    模型导出为onnx

    1. python export.py --weights yolov5s.pt --include onnx --rknpu rk3566

    再将模型导出为

    鲁班猫板端

    RKNN Toolkit Lite2 主要用于 RKNN 模型在 Rockchip NPU 上的部署。
    在使用 RKNN Toolkit Lite2 之前，用户需要先通过 RKNN Toolkit2 将各深度学习框架导出的模型转成 RKNN 模型。
    RKNN Toolkit2 完整的安装包和使用文档可以从以下链接获取：
    https://github.com/rockchip-linux/rknn-toolkit2

    环境安装

  6. 安装依赖

    1. sudo apt-get install -y python3-opencv
    2. sudo apt-get install -y python3-numpy

  7. 安装rknn_toolkit_lite2板端推理

    1. pip3 install rknn_toolkit_lite2-1.x.0-cp310-cp310-linux_aarch64.whl

  8. 将RKNPU2中的对应的so文件复制到开发板中进行替换

    1. ls -al /usr/lib/librkn*
    2. /usr/lib/librknn_api.so
    3. /usr/lib/librknnrt.so

  9. 使用代码进行推理 ```python import os import urllib import traceback import time import sys import numpy as np import cv2

    from rknn.api import RKNN

    from rknnlite.api import RKNNLite

RKNN_MODEL = ‘yolov8n.rknn’ IMG_PATH = ‘./bus.jpg’ DATASET = ‘./dataset.txt’

QUANTIZE_ON = True

OBJ_THRESH = 0.25 NMS_THRESH = 0.45 IMG_SIZE = 640

CLASSES = (“person”, “bicycle”, “car”, “motorbike “, “aeroplane “, “bus “, “train”, “truck “, “boat”, “traffic light”, “fire hydrant”, “stop sign “, “parking meter”, “bench”, “bird”, “cat”, “dog “, “horse “, “sheep”, “cow”, “elephant”, “bear”, “zebra “, “giraffe”, “backpack”, “umbrella”, “handbag”, “tie”, “suitcase”, “frisbee”, “skis”, “snowboard”, “sports ball”, “kite”, “baseball bat”, “baseball glove”, “skateboard”, “surfboard”, “tennis racket”, “bottle”, “wine glass”, “cup”, “fork”, “knife “, “spoon”, “bowl”, “banana”, “apple”, “sandwich”, “orange”, “broccoli”, “carrot”, “hot dog”, “pizza “, “donut”, “cake”, “chair”, “sofa”, “pottedplant”, “bed”, “diningtable”, “toilet “, “tvmonitor”, “laptop “, “mouse “, “remote “, “keyboard “, “cell phone”, “microwave “, “oven “, “toaster”, “sink”, “refrigerator “, “book”, “clock”, “vase”, “scissors “, “teddy bear “, “hair drier”, “toothbrush “)

def sigmoid(x): return 1 / (1 + np.exp(-x))

def xywh2xyxy(x):

  1. # Convert [x, y, w, h] to [x1, y1, x2, y2]
  2. y = np.copy(x)
  3. y[:, 0] = x[:, 0] - x[:, 2] / 2  # top left x
  4. y[:, 1] = x[:, 1] - x[:, 3] / 2  # top left y
  5. y[:, 2] = x[:, 0] + x[:, 2] / 2  # bottom right x
  6. y[:, 3] = x[:, 1] + x[:, 3] / 2  # bottom right y
  7. return y

def process(input, mask, anchors):

  1. anchors = [anchors[i] for i in mask]
  2. grid_h, grid_w = map(int, input.shape[0:2])
  4. box_confidence = sigmoid(input[..., 4])
  5. box_confidence = np.expand_dims(box_confidence, axis=-1)
  7. box_class_probs = sigmoid(input[..., 5:])
  9. box_xy = sigmoid(input[..., :2])*2 - 0.5
  11. col = np.tile(np.arange(0, grid_w), grid_w).reshape(-1, grid_w)
  12. row = np.tile(np.arange(0, grid_h).reshape(-1, 1), grid_h)
  13. col = col.reshape(grid_h, grid_w, 1, 1).repeat(3, axis=-2)
  14. row = row.reshape(grid_h, grid_w, 1, 1).repeat(3, axis=-2)
  15. grid = np.concatenate((col, row), axis=-1)
  16. box_xy += grid
  17. box_xy *= int(IMG_SIZE/grid_h)
  19. box_wh = pow(sigmoid(input[..., 2:4])*2, 2)
  20. box_wh = box_wh * anchors
  22. box = np.concatenate((box_xy, box_wh), axis=-1)
  24. return box, box_confidence, box_class_probs

def filter_boxes(boxes, box_confidences, box_class_probs): “””Filter boxes with box threshold. It’s a bit different with origin yolov5 post process!

  1. # Arguments
  2.     boxes: ndarray, boxes of objects.
  3.     box_confidences: ndarray, confidences of objects.
  4.     box_class_probs: ndarray, class_probs of objects.
  6. # Returns
  7.     boxes: ndarray, filtered boxes.
  8.     classes: ndarray, classes for boxes.
  9.     scores: ndarray, scores for boxes.
  10. """
  11. boxes = boxes.reshape(-1, 4)
  12. box_confidences = box_confidences.reshape(-1)
  13. box_class_probs = box_class_probs.reshape(-1, box_class_probs.shape[-1])
  15. _box_pos = np.where(box_confidences >= OBJ_THRESH)
  16. boxes = boxes[_box_pos]
  17. box_confidences = box_confidences[_box_pos]
  18. box_class_probs = box_class_probs[_box_pos]
  20. class_max_score = np.max(box_class_probs, axis=-1)
  21. classes = np.argmax(box_class_probs, axis=-1)
  22. _class_pos = np.where(class_max_score >= OBJ_THRESH)
  24. boxes = boxes[_class_pos]
  25. classes = classes[_class_pos]
  26. scores = (class_max_score* box_confidences)[_class_pos]
  28. return boxes, classes, scores

def nms_boxes(boxes, scores): “””Suppress non-maximal boxes.

  1. # Arguments
  2.     boxes: ndarray, boxes of objects.
  3.     scores: ndarray, scores of objects.
  5. # Returns
  6.     keep: ndarray, index of effective boxes.
  7. """
  8. x = boxes[:, 0]
  9. y = boxes[:, 1]
  10. w = boxes[:, 2] - boxes[:, 0]
  11. h = boxes[:, 3] - boxes[:, 1]
  13. areas = w * h
  14. order = scores.argsort()[::-1]
  16. keep = []
  17. while order.size > 0:
  18.     i = order[0]
  19.     keep.append(i)
  21.     xx1 = np.maximum(x[i], x[order[1:]])
  22.     yy1 = np.maximum(y[i], y[order[1:]])
  23.     xx2 = np.minimum(x[i] + w[i], x[order[1:]] + w[order[1:]])
  24.     yy2 = np.minimum(y[i] + h[i], y[order[1:]] + h[order[1:]])
  26.     w1 = np.maximum(0.0, xx2 - xx1 + 0.00001)
  27.     h1 = np.maximum(0.0, yy2 - yy1 + 0.00001)
  28.     inter = w1 * h1
  30.     ovr = inter / (areas[i] + areas[order[1:]] - inter)
  31.     inds = np.where(ovr <= NMS_THRESH)[0]
  32.     order = order[inds + 1]
  33. keep = np.array(keep)
  34. return keep

def yolov5_post_process(input_data): masks = [[0, 1, 2], [3, 4, 5], [6, 7, 8]] anchors = [[10, 13], [16, 30], [33, 23], [30, 61], [62, 45], [59, 119], [116, 90], [156, 198], [373, 326]]

  1. boxes, classes, scores = [], [], []
  2. for input, mask in zip(input_data, masks):
  3.     b, c, s = process(input, mask, anchors)
  4.     b, c, s = filter_boxes(b, c, s)
  5.     boxes.append(b)
  6.     classes.append(c)
  7.     scores.append(s)
  9. boxes = np.concatenate(boxes)
  10. boxes = xywh2xyxy(boxes)
  11. classes = np.concatenate(classes)
  12. scores = np.concatenate(scores)
  14. nboxes, nclasses, nscores = [], [], []
  15. for c in set(classes):
  16.     inds = np.where(classes == c)
  17.     b = boxes[inds]
  18.     c = classes[inds]
  19.     s = scores[inds]
  21.     keep = nms_boxes(b, s)
  23.     nboxes.append(b[keep])
  24.     nclasses.append(c[keep])
  25.     nscores.append(s[keep])
  27. if not nclasses and not nscores:
  28.     return None, None, None
  30. boxes = np.concatenate(nboxes)
  31. classes = np.concatenate(nclasses)
  32. scores = np.concatenate(nscores)
  34. return boxes, classes, scores

def draw(image, boxes, scores, classes): “””Draw the boxes on the image.

  1. # Argument:
  2.     image: original image.
  3.     boxes: ndarray, boxes of objects.
  4.     classes: ndarray, classes of objects.
  5.     scores: ndarray, scores of objects.
  6.     all_classes: all classes name.
  7. """
  8. for box, score, cl in zip(boxes, scores, classes):
  9.     top, left, right, bottom = box
  10.     print('class: {}, score: {}'.format(CLASSES[cl], score))
  11.     print('box coordinate left,top,right,down: [{}, {}, {}, {}]'.format(top, left, right, bottom))
  12.     top = int(top)
  13.     left = int(left)
  14.     right = int(right)
  15.     bottom = int(bottom)
  17.     cv2.rectangle(image, (top, left), (right, bottom), (255, 255, 0), 2)
  18.     cv2.putText(image, '{0} {1:.2f}'.format(CLASSES[cl], score),
  19.                 (top, left - 6),
  20.                 cv2.FONT_HERSHEY_SIMPLEX,
  21.                 0.6, (0, 0, 255), 2)

def letterbox(im, new_shape=(640, 640), color=(0, 0, 0)):

  1. # Resize and pad image while meeting stride-multiple constraints
  2. shape = im.shape[:2]  # current shape [height, width]
  3. if isinstance(new_shape, int):
  4.     new_shape = (new_shape, new_shape)
  6. # Scale ratio (new / old)
  7. r = min(new_shape[0] / shape[0], new_shape[1] / shape[1])
  9. # Compute padding
  10. ratio = r, r  # width, height ratios
  11. new_unpad = int(round(shape[1] * r)), int(round(shape[0] * r))
  12. dw, dh = new_shape[1] - new_unpad[0], new_shape[0] - new_unpad[1]  # wh padding
  14. dw /= 2  # divide padding into 2 sides
  15. dh /= 2
  17. if shape[::-1] != new_unpad:  # resize
  18.     im = cv2.resize(im, new_unpad, interpolation=cv2.INTER_LINEAR)
  19. top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1))
  20. left, right = int(round(dw - 0.1)), int(round(dw + 0.1))
  21. im = cv2.copyMakeBorder(im, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color)  # add border
  22. return im, ratio, (dw, dh)

if name == ‘main‘:

  1. # Create RKNN object
  2. rknn = RKNNLite(verbose=True)
  4. print('--> Load RKNN model')
  5. ret = rknn.load_rknn("yolov5s.rknn")
  6. # Init runtime environment
  7. print('--> Init runtime environment')
  8. ret = rknn.init_runtime()
  9. # ret = rknn.init_runtime('rk3566')
  10. if ret != 0:
  11.     print('Init runtime environment failed!')
  12.     exit(ret)
  13. print('done')
  15. # Set inputs
  16. img = cv2.imread(IMG_PATH)
  17. # img, ratio, (dw, dh) = letterbox(img, new_shape=(IMG_SIZE, IMG_SIZE))
  18. img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
  19. img = cv2.resize(img, (IMG_SIZE, IMG_SIZE))
  21. # Inference
  22. print('--> Running model')
  23. outputs = rknn.inference(inputs=[img])
  24. np.save('./onnx_yolov5_0.npy', outputs[0])
  25. np.save('./onnx_yolov5_1.npy', outputs[1])
  26. np.save('./onnx_yolov5_2.npy', outputs[2])
  27. print('done')
  29. # post process
  30. input0_data = outputs[0]
  31. input1_data = outputs[1]
  32. input2_data = outputs[2]
  34. input0_data = input0_data.reshape([3, -1]+list(input0_data.shape[-2:]))
  35. input1_data = input1_data.reshape([3, -1]+list(input1_data.shape[-2:]))
  36. input2_data = input2_data.reshape([3, -1]+list(input2_data.shape[-2:]))
  38. input_data = list()
  39. input_data.append(np.transpose(input0_data, (2, 3, 0, 1)))
  40. input_data.append(np.transpose(input1_data, (2, 3, 0, 1)))
  41. input_data.append(np.transpose(input2_data, (2, 3, 0, 1)))
  43. boxes, classes, scores = yolov5_post_process(input_data)
  45. img_1 = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
  46. if boxes is not None:
  47.     draw(img_1, boxes, scores, classes)
  48.     cv2.imwrite('result.jpg', img_1)
  50. rknn.release()
  2. <a name="ehmjK"></a>
  3. ### 多线程读取
  4. ```python
  5. import os
  6. import urllib
  7. import traceback
  8. import time
  9. import sys
  10. import numpy as np
  11. import cv2
  12. # from rknn.api import RKNN
  13. from rknnlite.api import RKNNLite
  15. import threading
  16. import queue
  17. # 自定义无缓存读视频类
  18. class VideoCapture:
  19.     """Customized VideoCapture, always read latest frame"""
  21.     def __init__(self, name):
  22.         self.cap = cv2.VideoCapture(name)
  23.         self.q = queue.Queue(maxsize=3)
  24.         self.stop_threads = False    # to gracefully close sub-thread
  25.         th = threading.Thread(target=self._reader)
  26.         th.daemon = True             # 设置工作线程为后台运行
  27.         th.start()
  29.     # 实时读帧，只保存最后一帧
  30.     def _reader(self):
  31.         while not self.stop_threads:
  32.             ret, frame = self.cap.read()
  33.             if not ret:
  34.                 break
  35.             if not self.q.empty():
  36.                 try:
  37.                     self.q.get_nowait() 
  38.                 except queue.Empty:
  39.                     pass
  40.             self.q.put(frame)
  42.     def read(self):
  43.         return self.q.get()
  45.     def terminate(self):
  46.         self.stop_threads = True
  47.         self.cap.release()
  49. if __name__ == '__main__':
  50.     capture = VideoCapture("/dev/video9")
  51.     img = capture.read()
  53.     while img is not None:
  55.         cv2.imshow('src',img_1)
  57.         img = capture.read()
  58.         if ord("q") == cv2.waitKey(10):
  59.             break;
  60.     capture.terminate()

ROS订阅摄像头

  1. 创建工作空间

    1. mkdir -p car_ws/src

  2. 创建包

    1. ros2 pkg create --build-type ament_python yolo_detect
  • 导入代码 ```python import rclpy from rclpy.node import Node from std_msgs.msg import String from sensor_msgs.msg import Image from cv_bridge import CvBridge import cv2 as cv

class SubNode(Node): def init(self): super(SubNode, self).init(‘py_sub’) self.create_subscription(Image,’/image_raw’,self.call_back,10) self.cv_bridge = CvBridge() 

  1. def call_back(self,msg):
  2.     # self.get_logger().info(msg.data)
  3.     image = self.cv_bridge.imgmsg_to_cv2(msg, 'bgr8')
  4.     cv.imshow("image",image)
  5.     cv.waitKey(10)

def main(): rclpy.init() node = SubNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown()

if name == ‘main‘: main()

  2. - 修改配置setup.py
  3. ```python
  4. entry_points={
  5.         'console_scripts': [
  6.             'yolo_detect = yolo_detect.det:main',
  7.         ],
  8.     },

编译包

  1. colcon build --packages-select yolo_detect

启动节点

  1. ros2 run yolo_detect yolo_detect

导入usb_cam包

  1. colcon build
  2. source install/setup.bash

编译完成之后

  1. ros2 launch usb_cam demo_launch.py

常用指令

  1. ros2 topic list
  2. ros2 topic type /image_raw
  3. ros2 interface show sensor_msgs/msg/Image