python 基于HyperLPR3和OpenCV实时RTSP视频流中文车牌识别（实现ONNX、OpenVINO、NCNN车牌框检测）

1、HyperLPR3简介

HyperLPR3是一款高性能开源中文车牌识别。源码：https://github.com/szad670401/HyperLPR

2、快速部署与简单使用

参考文章：HyperLPR3车牌识别-五分钟搞定: 中文车牌识别光速部署与使用

3、利用OpenCV对RTSP视频流实时读取

我是参考yolo5的streamload使用python迭代器实现视频流读取，视频流读取部分可以按照自己想法继续优化，例如视频流读取分离。

dataset.py

  1. import math
  2. import time
  3. import threading
  4. import platform
  5. import cv2
  6. import numpy as np
  7.  
  8. from utils.my_logger import logger
  9.  
  10. class LoadStreamsByRTSP:
  11.     
  12.     def __init__(self,camera_list):
  13.  
  14.         self.camera_list = camera_list
  15.         
  16.         n = len(camera_list)
  17.         # 原尺寸图像 初始化图片 fps 总帧数(无穷大) 线程数 相机ip数组 
  18.         self.orgImgs,self.imgs, self.fps, self.frames, self.threads,self.camIps,self.ids =[None] * n, [None] * n, [0] * n, [0] * n, [None] * n,[None]*n,[None]*n
  19.         self.sources = []  # clean source names for later
  20.         self.countFrm = {} #统计帧数，索引，帧数
  21.         for i,item in enumerate(camera_list):
  22.             # Start thread to read frames from video stream
  23.             st = f'{i + 1}/{n}: {item[0]}... '
  24.             rtspUrl = item[2]
  25.             cap = cv2.VideoCapture(rtspUrl)
  26.             if cap.isOpened():
  27.                 print(f'{rtspUrl} open success !')
  28.                 self.sources.append(rtspUrl) # rtsp地址
  29.             # 如果当前Rtsp流打开失败，记录错误日志，并循环下一个流
  30.             else :
  31.                 logger.error(f'{st}Failed to open {item[0]}')
  32.                 continue
  33.  
  34.             w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) # 获取视频宽度
  35.             h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) # 获取视频高度
  36.             fps = cap.get(cv2.CAP_PROP_FPS)  # warning: may return 0 or nan
  37.             self.frames[i] = max(int(cap.get(cv2.CAP_PROP_FRAME_COUNT)), 0) or float('inf')  # infinite stream fallback
  38.             self.fps[i] = max((fps if math.isfinite(fps) else 0) % 100, 0) or 30  # 30 FPS fallback
  39.             _, self.imgs[i] = cap.read()  # guarantee first frame
  40.             self.orgImgs[i] = self.imgs[i].copy()
  41.             self.countFrm[i] = 0
  42.             self.camIps[i] = item[0] # 记录当前相机IP，用于业务判断
  43.             self.ids[i] = item[1] # 记录当前相机编号
  44.             self.threads[i] = threading.Thread(target=self.update, args=([i, cap, rtspUrl]), daemon=True)
  45.             logger.info(f"{st} Success ({self.frames[i]} frames {w}x{h} at {self.fps[i]:.2f} FPS)")
  46.             print(f"{st} Success ({self.frames[i]} frames {w}x{h} at {self.fps[i]:.2f} FPS)")
  47.             self.threads[i].start()
  48.         print(' ')  # newline
  49.  
  50.     def update(self, i, cap, stream):
  51.         if cap.isOpened():
  52.             while True:
  53.                 success, im = cap.read()
  54.                 if success:
  55.                     self.orgImgs[i] = im #原始图像
  56.                 else:
  57.                     logger.warning(f'WARNING: Video stream[{stream}] unresponsive, please check your IP camera connection.15s restart stream...')
  58.                     self.orgImgs[i] = np.zeros_like(self.orgImgs[i])
  59.                     time.sleep(15)
  60.                     cap = cv2.VideoCapture(stream)
  61.                     if cap.isOpened():
  62.                         logger.info(f'{stream} restart success !')
  63.             #time.sleep(1 / self.fps[i])  # wait time if videofile
  64.  
  65.     def __iter__(self):
  66.         #self.count = -1
  67.         return self
  68.     
  69.     def initImgs(self):
  70.         # 需保证imgs里没有None的图片
  71.         while self.imgs[0] is None:
  72.             time.sleep(1)
  73.         
  74.         for i,im in enumerate(self.imgs):
  75.             if self.imgs[i] is None:
  76.                 self.imgs[i] = np.zeros_like(self.imgs[0])
  77.         while self.orgImgs[0] is None:
  78.             time.sleep(1)
  79.         
  80.         for i,im in enumerate(self.orgImgs):
  81.             if self.orgImgs[i] is None:
  82.                 self.orgImgs[i] = np.zeros_like(self.orgImgs[0])
  83.     
  84.     def __next__(self):
  85.         beginTime = time.time()
  86.  
  87.         orgImgs = self.orgImgs.copy()
  88.  
  89.         return  orgImgs, beginTime,self.camIps,self.ids
  90.  
  91.     def __len__(self):
  92.         return len(self.sources)

4、根据自定义修改一些默认设置（使用默认onnx推理也可跳过该步骤）

4.1、下载hyperlpr3

从Github下载Prj-Python源码，并将其中的hyperlpr3文件夹放入自己项目中。

4.2、修改模型文件夹默认位置

修改其中config文件夹下的setting.py，将模型文件夹路径设置为当前项目下，这一步也可以不做，默认是在家目录下面。

  1. import os
  2. import sys
  3.  
  4. _MODEL_VERSION_ = "20230229"
  5.  
  6. if 'win32' in sys.platform:
  7.     #_DEFAULT_FOLDER_ = os.path.join(os.environ['HOMEPATH'], ".hyperlpr3")
  8.     _DEFAULT_FOLDER_ = './'
  9. else:
  10.     #_DEFAULT_FOLDER_ = os.path.join(os.environ['HOME'], ".hyperlpr3")
  11.     _DEFAULT_FOLDER_ = './'
  12.  
  13. _ONLINE_URL_ = "http://hyperlpr.tunm.top/raw/"
  14.  
  15. onnx_runtime_config = dict(
  16.     det_model_path_320x=os.path.join(_MODEL_VERSION_, "onnx", "y5fu_320x_sim.onnx"),
  17.     det_model_path_640x=os.path.join(_MODEL_VERSION_, "onnx", "y5fu_640x_sim.onnx"),
  18.     rec_model_path=os.path.join(_MODEL_VERSION_, "onnx", "rpv3_mdict_160_r3.onnx"),
  19.     cls_model_path=os.path.join(_MODEL_VERSION_, "onnx", "litemodel_cls_96x_r1.onnx"),
  20. )
  21.  
  22. onnx_model_maps = ["det_model_path_320x", "det_model_path_640x", "rec_model_path", "cls_model_path"]
  23.  
  24. _REMOTE_URL_ = "https://github.com/szad670401/HyperLPR/blob/master/resource/models/onnx/"

之后将模型文件夹复制到当前项目根目录下

5、对RTSP视频流进行实时车牌识别

plateRecognizer.py

  1. # 导入cv相关库
  2. import cv2
  3. import os
  4. import json
  5. import time
  6. import traceback
  7.  
  8. # 导入依赖包
  9. import hyperlpr3 as lpr3
  10. from hyperlpr3.common.typedef import *
  11.  
  12. from utils.my_logger import logger
  13. from utils.dataset import LoadStreamsByRTSP
  14. import utils.plateUtil as plateUtil
  15.  
  16. #将10位时间戳或者13位转换为时间字符串，默认为2017-10-01 13:37:04格式
  17. def timestamp_to_date(time_stamp, format_string="%Y%m%d%H%M%S"):
  18.     time_array = time.localtime(time_stamp)
  19.     other_style_time = time.strftime(format_string, time_array)
  20.     
  21.     return other_style_time
  22.  
  23.  
  24. if __name__ == '__main__':
  25.     try:
  26.         camera_list = []
  27.         if not os.path.exists('images'):
  28.             os.mkdir('images')
  29.         camPlateInfoRecord = {} #记录当前相机识别到的车牌号
  30.         camPlateInfoRecordTime = {}
  31.         #catcher = {} # 识别对象
  32.         with open("config.json", encoding='utf-8') as f:
  33.             jsonData = json.load(f)
  34.         for ipkey in jsonData:
  35.             camera_list.append([ipkey,jsonData[ipkey]['id'],jsonData[ipkey]['rtspUrl']])
  36.         
  37.         for item in camera_list:
  38.             print(item)
  39.             camPlateInfoRecord[item[0]] = [] # 初始化
  40.             camPlateInfoRecordTime[item[0]] = None
  41.             # catcher[item[0]] = lpr3.LicensePlateCatcher()
  42.             
  43.             
  44.             if not os.path.exists('./images/'+item[0]):
  45.                 os.mkdir('./images/'+item[0])
  46.                 
  47.             
  48.         # 实例化识别对象
  49.         catcher = lpr3.LicensePlateCatcher() # detect_level=lpr3.DETECT_LEVEL_HIGH
  50.         dataset = LoadStreamsByRTSP(camera_list)
  51.         dataset.initImgs()
  52.         frameSkip = 5 # 跳帧
  53.         for orgImgs, beginTime,camIps,ids in dataset:
  54.             recordTime = time.time()
  55.             for i,orgImg in enumerate(orgImgs) :
  56.                 results = catcher(orgImg)
  57.                 print(results)
  58.                 # print(f'catcher Done. ({time.time() - beginTime:.3f}s)') # 打印时间
  59.                 if len(results)>0:
  60.                     plateInfo = sorted(results,key=lambda x : x[1], reverse=True)[0]# 根据置信度进行降序排序，并取置信度最高的一项
  61.                     plateNo = plateUtil.get_plate_color(plateInfo[2]) + plateInfo[0] # 车牌号
  62.                     plateConf = plateInfo[1] # 置信度
  63.                     camPlateInfoRecordTime[camIps[i]] = recordTime
  64.                     if plateConf > 0.97:
  65.                         # TODO 可直接判定为最佳车牌
  66.                         camPlateInfoRecord[camIps[i]].append((plateNo,plateConf,orgImg))
  67.                         break
  68.                     elif plateConf > 0.8:
  69.                         camPlateInfoRecord[camIps[i]].append((plateNo,plateConf,orgImg))
  70.                         pass
  71.                 else:
  72.                     # TODO 未识别到车牌
  73.                     pass
  74.                 #cv2.imshow(camIps[i], cv2.resize(orgImg,(1280,720))) # cv2.resize(im0,(800,600))
  75.                 #cv2.waitKey(1)  # 1 millisecond
  76.                 if camPlateInfoRecordTime[camIps[i]] is not None:
  77.                     currentTime = time.time()
  78.                     # 判断时间是否大于2秒
  79.                     if recordTime - camPlateInfoRecordTime[camIps[i]] > 2 and len(camPlateInfoRecord[camIps[i]]) > 0:
  80.                         # 选取置信度最大的车牌信息保存
  81.                         savePlateInfo = sorted(camPlateInfoRecord[camIps[i]], key=lambda e: e[1], reverse=True)[0]
  82.                         save_path = './images/'+camIps[i]+'/'+timestamp_to_date(int(currentTime),'%Y%m%d%H%M%S')+'_'+savePlateInfo[0]+'_'+ids[i]+'.jpg'
  83.                         cv2.imencode('.jpg', savePlateInfo[2])[1].tofile(save_path)
  84.                         logger.info(f"{camIps[i]} 识别到车牌：{savePlateInfo[0]},置信度:{savePlateInfo[1]},保存路径:{save_path}")
  85.                         camPlateInfoRecordTime[camIps[i]] = None # 清空时间记录
  86.                         camPlateInfoRecord[camIps[i]] = [] # 清空车牌记录
  87.             # 显示FPS
  88.             fps_time = time.time()-beginTime
  89.             if(fps_time > 0):
  90.                 video_fps  = int(1/fps_time)
  91.                 print(f'{fps_time:.3f}s'+" fps= %.2f"%(video_fps))
  92.             #进行跳帧休眠处理
  93.             loopTime = int(round(fps_time * 1000))
  94.             if loopTime < 40*frameSkip:
  95.                 time.sleep(0.04*frameSkip-loopTime/1000)
  96.     except BaseException:
  97.         logger.error("处理时出错！"+traceback.format_exc())
  98.         time.sleep(1)

代码解释：

1）、frameSkip = 5 ，我这里使用了跳帧方式去识别所读取到的视频流，主要是避免过多过快的车牌识别检测导致的CPU过高或者其他系统性能问题，一般情况下每5帧识别一次已经足够，可根据自身需求进行调整。

2）、车牌真实性判断逻辑，我这里是当车牌的置信度大于0.8则保存该帧的车牌信息及图像，当该视频流2s内不再出现大于0.8置信度的车牌时，保存图像及车牌信息至文件夹中。该逻辑可根据自身需求进行修改代码，例如通过有效帧次数来保存识别最佳的车牌信息等。

其他文件：

config.json

  1. {
  2.     "192.168.**.**": {"id":"XXX001","rtspUrl":"相机的RTSP地址"}
  3. }

plateUtil.py

  1. from hyperlpr3.common.typedef import *
  2.  
  3.  
  4. def get_plate_color(plate_type):
  5.     plateColor = "蓝"
  6.     if plate_type == UNKNOWN:
  7.         plateColor = '未知'
  8.     elif plate_type == BLUE:
  9.         plateColor = "蓝"
  10.     elif plate_type == YELLOW_SINGLE:
  11.         plateColor = "黄"
  12.     elif plate_type == WHILE_SINGLE:
  13.         plateColor = "白"
  14.     elif plate_type == GREEN:
  15.         plateColor = "绿"
  16.     return plateColor

6、使用OpenVINO检测车牌框

pip install openvino

6.1 转换为OpenVINO格式的模型文件

onnx2openvino.py

  1. # coding:UTF-8
  2. import os
  3.  
  4. import openvino as ov
  5.  
  6. # 定义 ONNX 模型路径
  7. onnx_model_path = "y5fu_320x_sim.onnx"
  8.  
  9. # 定义输出目录路径
  10. output_dir = "./openvino"
  11.  
  12. # 检查输出目录是否存在，如果不存在则创建
  13. if not os.path.exists(output_dir):
  14.     os.makedirs(output_dir)
  15.     print(f"Created output directory at: {output_dir}")
  16. else:
  17.     print(f"Output directory already exists at: {output_dir}")
  18.  
  19. # 设置模型转换参数
  20. try:
  21.     # 转换 ONNX 模型为 OpenVINO 格式
  22.     ov_model = ov.convert_model(onnx_model_path)
  23.  
  24.     # 保存 OpenVINO 模型
  25.     ir_path = './openvino/y5fu_320x_sim.xml'
  26.     ov.save_model(ov_model, ir_path)
  27.  
  28.     
  29.     print(f"Model successfully converted and saved to: {output_dir}")
  30. except Exception as e:
  31.     print(f"Error during model conversion: {e}")

执行后即可得到OpenVINO推理文件

6.2、添加OpenVINO推理类

找到 hyperlpr3\inference\multitask_detect.py

添加：

  1. class MultiTaskDetectorOpenVINO(HamburgerABC):
  2.  
  3.     def __init__(self, openvino_path, box_threshold: float = 0.5, nms_threshold: float = 0.6, *args, **kwargs):
  4.         super().__init__(*args, **kwargs)
  5.         import openvino as ov
  6.         self.box_threshold = box_threshold
  7.         self.nms_threshold = nms_threshold
  8.         # 👇Create OpenVINO Core
  9.         core = ov.Core()
  10.         # 👇读取模型
  11.         self.model = core.read_model(model=openvino_path, weights=openvino_path.replace(".xml", ".bin"))
  12.         # 👇加载模型，如果用intel的显卡就把CPU改成GPU，但是要确保你的显卡驱动安装好
  13.         self.compiled_model = core.compile_model(model=self.model, device_name="CPU")
  14.         self.inputs_option = self.model.input(0)
  15.         self.outputs_option = self.model.output(0)
  16.         
  17.         #input_option = self.inputs_option[0]
  18.         input_size_ = tuple(self.inputs_option.shape[2:])
  19.         self.input_size = tuple(self.input_size)
  20.         if not self.input_size:
  21.             self.input_size = input_size_
  22.         assert self.input_size == input_size_, 'The dimensions of the input do not match the model expectations.'
  23.         assert self.input_size[0] == self.input_size[1]
  24.         self.input_name = self.inputs_option.names
  25.  
  26.     def _run_session(self, data):
  27.         result = self.compiled_model(data)[0]
  28.         return result
  29.  
  30.     def _postprocess(self, data):
  31.         r, left, top = self.tmp_pack
  32.         return post_precessing(data, r, left, top)
  33.  
  34.     def _preprocess(self, image):
  35.         img, r, left, top = detect_pre_precessing(image, self.input_size)
  36.         self.tmp_pack = r, left, top
  37.  
  38.         return img

6.3、添加OpenVINO推理方法

找到 hyperlpr3\hyperlpr3.py

  1. from .config.settings import onnx_runtime_config as ort_cfg
  2. from .inference.pipeline import LPRMultiTaskPipeline
  3. from .common.typedef import *
  4. from os.path import join
  5. from .config.settings import _DEFAULT_FOLDER_
  6. from .config.configuration import initialization
  7.  
  8.  
  9. initialization()
  10.  
  11. class LicensePlateCatcher(object):
  12.  
  13.     def __init__(self,
  14.                  detect_inference: int = INFER_ONNX_RUNTIME,
  15.                  inference: int = INFER_ONNX_RUNTIME,
  16.                  folder: str = _DEFAULT_FOLDER_,
  17.                  detect_level: int = DETECT_LEVEL_LOW,
  18.                  logger_level: int = 3,
  19.                  full_result: bool = False):
  20.         if inference == INFER_ONNX_RUNTIME:
  21.             
  22.             from hyperlpr3.inference.recognition import PPRCNNRecognitionORT
  23.             from hyperlpr3.inference.classification import ClassificationORT
  24.             import onnxruntime as ort
  25.             ort.set_default_logger_severity(logger_level)
  26.             if detect_level == DETECT_LEVEL_LOW:
  27.                 if detect_inference == INFER_ONNX_RUNTIME:
  28.                     from hyperlpr3.inference.multitask_detect import MultiTaskDetectorORT
  29.                     det = MultiTaskDetectorORT(join(folder, ort_cfg['det_model_path_320x']), input_size=(320, 320))
  30.                 elif detect_inference == INFER_OPENVINO:
  31.                     from hyperlpr3.inference.multitask_detect import MultiTaskDetectorOpenVINO
  32.                     det = MultiTaskDetectorOpenVINO('./openvino/y5fu_320x_sim.xml', input_size=(320, 320))
  33.                 elif detect_inference == INFER_NCNN:
  34.                     from hyperlpr3.inference.multitask_detect import MultiTaskDetectorNCNN
  35.                     det = MultiTaskDetectorNCNN('./ncnn/y5fu_320x_sim.ncnn.bin', input_size=(320, 320))
  36.             elif detect_level == DETECT_LEVEL_HIGH:
  37.                 det = MultiTaskDetectorORT(join(folder, ort_cfg['det_model_path_640x']), input_size=(640, 640))
  38.             else:
  39.                 raise NotImplemented
  40.             rec = PPRCNNRecognitionORT(join(folder, ort_cfg['rec_model_path']), input_size=(48, 160))
  41.             cls = ClassificationORT(join(folder, ort_cfg['cls_model_path']), input_size=(96, 96))
  42.             self.pipeline = LPRMultiTaskPipeline(detector=det, recognizer=rec, classifier=cls, full_result=full_result)
  43.         else:
  44.             raise NotImplemented
  45.  
  46.     def __call__(self, image: np.ndarray, *args, **kwargs):
  47.         return self.pipeline(image)

6.4、设置为OpenVINO推理

plateRecognizer.py 中

catcher = lpr3.LicensePlateCatcher(detect_inference=INFER_OPENVINO) # detect_level=lpr3.DETECT_LEVEL_HIGH

7、使用NCNN检测车牌框

7.1、转换为NCNN格式的模型文件

注意：网上大部分都是说要下载protobuf和ncnn源码进行编译，再使用onnx2ncnn.exe进行转换，实际验证该方法对onnx转换为ncnn格式时，会报错误Unsupported slice axes !。

建议使用pnnx方式直接转换。

pip install pnnx
pnnx y5fu_320x_sim.onnx inputshape=[1,3,320,320]

执行后会生成一堆文件，选择ncnn的两个文件即可

在项目根目录新建一个ncnn文件夹，并上述两个模型文件放入。

7.2、添加NCNN推理类

找到 hyperlpr3\inference\multitask_detect.py

添加：

  1. import ncnn
  2. class MultiTaskDetectorNCNN(HamburgerABC):
  3.  
  4.     def __init__(self, ncnn_path, box_threshold: float = 0.5, nms_threshold: float = 0.6, *args, **kwargs):
  5.         super().__init__(*args, **kwargs)
  6.         self.box_threshold = box_threshold
  7.         self.nms_threshold = nms_threshold
  8.         # 加载ncnn模型
  9.         self.net = ncnn.Net()
  10.         self.net.opt.use_vulkan_compute = False
  11.         self.net.load_param(ncnn_path.replace(".bin", ".param"))
  12.         self.net.load_model(ncnn_path)
  13.         self.input_size = (320,320)
  14.  
  15.     def _run_session(self, data):
  16.         input_data = ncnn.Mat(data[0])
  17.         # 创建提取器
  18.         extractor = self.net.create_extractor()
  19.         extractor.input("in0", input_data)
  20.         # 执行推理
  21.         ret1, mat_out1 = extractor.extract("out0")
  22.         # 处理输出
  23.         result = np.expand_dims(np.array(mat_out1), axis=0)
  24.         return result
  25.  
  26.     def _postprocess(self, data):
  27.         r, left, top = self.tmp_pack
  28.         return post_precessing(data, r, left, top)
  29.  
  30.     def _preprocess(self, image):
  31.         img, r, left, top = detect_pre_precessing(image, self.input_size)
  32.         self.tmp_pack = r, left, top
  33.  
  34.         return img

7.3、添加NCNN推理方法

同6.3

7.4、设置为NCNN推理

plateRecognizer.py 中

catcher = lpr3.LicensePlateCatcher(detect_inference=INFER_NCNN) # detect_level=lpr3.DETECT_LEVEL_HIGH

8、ONNX、OpenVINO、NCNN 推理速度记录（包含预处理和后处理）

本人电脑硬件如下：

内存：16G

CPU：AMD Ryzen 5 4600U with Radeon Graphics  2.10 GHz

无独立显卡

输入视频流大小：1920*1080

8.1、在ONNX下的单帧车牌框检测速度

8.2、在OpenVINO下的单帧车牌框检测速度

不太稳定。。。。

8.3、在NCNN下的单帧车牌框检测速度

9、打包成exe可执行文件进行部署

9.1、安装auto-py-to-exe

  1. pip install auto-py-to-exe
  2.  
  3. auto-py-to-exe

9.2、部署ONNX推理方式

注意：不包含ncnn部分的代码可以直接打包

9.3、部署OpenVINO推理方式

其余一致，需在--collect-data添加openvino

9.4、部署NCNN推理方式

与9.2一样直接打包可能执行会出现错误：

Library not found: could no resolve或者显示ncnn的DLL load的问题

需要打开你的python环境所在目录找到 Lib\site-packages\ncnn.libs，将里面的DLL复制到Lib同一层目录下

再次打包即可通过并正常运行

10、结语

本人实际测试中，将上述车牌识别代码在Win732位下也部署了一套，其中OpenVINO因为不支持win7无法部署，实测发现ncnn的推理速度大概是onnx的1.5倍左右，并且ncnn推理占用CPU资源也比较低。

注：本文转载自blog.csdn.net的向往光的地方的文章"https://blog.csdn.net/Ding_Jie_/article/details/143425882"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！