手把手教你实现—基于OpenCV的车流量统计和车速检测代码

如果找到匹配的车辆，更新该车辆的追踪器对象，位置信息和ID。如果没有找到匹配的车辆，创建一个新的追踪器对象，将其添加到carTracker中，并更新位置信息和ID。在每一帧中，函数更新已跟踪车辆的位置和质量，并计算车辆的速度。本章将实现了一个简单的车辆速度估计和车流量统计的GUI应用，它使用了Haar级联检测器和相关跟踪器来检测和跟踪视频中的车辆，并通过图像处理和数学计算来估计车辆的速度。综上所述

就是求关注

4409人浏览 · 2023-06-08 19:33:33

就是求关注 · 2023-06-08 19:33:33 发布

本章将实现了一个简单的车辆速度估计和车流量统计的GUI应用，它使用了Haar级联检测器和相关跟踪器来检测和跟踪视频中的车辆，并通过图像处理和数学计算来估计车辆的速度。

1.首先，该代码需要cv2：用于图像处理和计算机视觉任务；dlib：用于对象跟踪和检测；time：用于计算帧率和时间间隔；threading：用于处理多线程操作；math：用于数学计算等。

2.然后，定义了全局变量video_path和highest_speed，用于存储选择的视频文件路径和最高时速度。

3.接下来，通过carCascade = cv2.CascadeClassifier('myhaar.xml')导入车辆检测器模型。然后定义了全局变量video和一些常量，如视频的宽度和高度。

carCascade = cv2.CascadeClassifier('myhaar.xml')
video = None
WIDTH = 1280
HEIGHT = 720

3.实现estimateSpeed(location1, location2, fps)函数用于估计车辆速度，根据两个位置点的像素距离、像素每米比例和帧率来计算车辆的速度。

def estimateSpeed(location1, location2, fps):
    d_pixels = math.sqrt(math.pow(location2[0] - location1[0], 2) + math.pow(location2[1] - location1[1], 2))
    ppm = 8.8
    d_meters = d_pixels / ppm
    if fps == 0.0:
        fps = 18
    speed = d_meters * fps * 3.6
    return speed

代码中：location1和location2是表示车辆位置的参数。它们是包含两个元素的元组或列表，分别表示车辆在图像中的坐标（x，y）。fps是帧率（Frames Per Second），表示视频播放的帧速率。d_pixels是计算两个位置之间像素距离的变量。通过使用两个位置的x和y坐标差值计算欧氏距离。ppm是每像素米数（Pixels Per Meter）的值。它表示图像中的像素距离与实际距离之间的比例关系。在这里，假设每个像素代表8.8厘米。d_meters是将像素距离转换为米的值。通过将像素距离除以ppm得到。如果帧率fps为0.0（未提供或无效值），将其设置为默认值18。speed是估计的速度，以千米/小时为单位。通过将距离（以米为单位）乘以帧率（以秒为单位）再乘以3.6（将小时转换为秒）计算得到。返回估计的速度。
综上所述，该函数根据两个位置之间的像素距离、帧率和像素到米的比例关系，估计车辆的速度（以千米/小时为单位）。

4.实现track_multiple_objects()函数是实现多目标跟踪的核心部分。它使用Haar级联检测器来检测图像中的车辆，并使用dlib的相关跟踪器来跟踪每个车辆。函数使用一个无限循环，从视频中读取帧，然后进行车辆检测和跟踪。在每一帧中，函数更新已跟踪车辆的位置和质量，并计算车辆的速度。还更新了最高时速度和实时车流量的标签，并显示跟踪结果的图像。

def track_multiple_objects():
    rectangleColor = (0, 255, 0)
    frameCounter = 0
    currentCarID = 0
    fps = 0

    carTracker = {}
    carLocation1 = {}
    carLocation2 = {}
    speed = [None] * 1000

    def update_highest_speed(speed):
        global highest_speed
        highest_speed = max(highest_speed, speed)
        speed_label.config(text="最高时速度：{} km/hr".format(highest_speed))

    def update_video_frame(image):
        image = Image.fromarray(image)
        image = image.resize((640, 480))
        photo = ImageTk.PhotoImage(image)
        video_label.config(image=photo)
        video_label.image = photo

        # 其他代码...

    total_cars = 0  # 车辆总数
    prev_total_cars = 0  # 上一帧的车辆总数

    while True:
        start_time = time.time()
        rc, image = video.read()
        if not rc:
            break

        image = cv2.resize(image, (WIDTH, HEIGHT))
        resultImage = image.copy()

        frameCounter = frameCounter + 1

        carIDtoDelete = []

        # 更新已跟踪车辆的位置和质量
        for carID in carTracker.keys():
            trackingQuality = carTracker[carID].update(image)

            if trackingQuality < 7:
                carIDtoDelete.append(carID)

        for carID in carIDtoDelete:
            carTracker.pop(carID, None)
            carLocation1.pop(carID, None)
            carLocation2.pop(carID, None)

        if not (frameCounter % 10):
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
            cars = carCascade.detectMultiScale(gray, 1.1, 13, 18, (24, 24))

            for (_x, _y, _w, _h) in cars:
                x = int(_x)
                y = int(_y)
                w = int(_w)
                h = int(_h)

                x_bar = x + 0.5 * w
                y_bar = y + 0.5 * h

                matchCarID = None

                for carID in carTracker.keys():
                    trackedPosition = carTracker[carID].get_position()

                    t_x = int(trackedPosition.left())
                    t_y = int(trackedPosition.top())
                    t_w = int(trackedPosition.width())
                    t_h = int(trackedPosition.height())

                    t_x_bar = t_x + 0.5 * t_w
                    t_y_bar = t_y + 0.5 * t_h

                    if (
                            (t_x <= x_bar <= (t_x + t_w))
                            and (t_y <= y_bar <= (t_y + t_h))
                            and (x <= t_x_bar <= (x + w))
                            and (y <= t_y_bar <= (y + h))
                    ):
                        matchCarID = carID

                if matchCarID is None:
                    tracker = dlib.correlation_tracker()
                    tracker.start_track(image, dlib.rectangle(x, y, x + w, y + h))

                    carTracker[currentCarID] = tracker
                    carLocation1[currentCarID] = [x, y, w, h]

                    currentCarID = currentCarID + 1

        for carID in carTracker.keys():
            trackedPosition = carTracker[carID].get_position()

            t_x = int(trackedPosition.left())
            t_y = int(trackedPosition.top())
            t_w = int(trackedPosition.width())
            t_h = int(trackedPosition.height())

            cv2.rectangle(resultImage, (t_x, t_y), (t_x + t_w, t_y + t_h), rectangleColor, 4)

            carLocation2[carID] = [t_x, t_y, t_w, t_h]

        end_time = time.time()

        if not (end_time == start_time):
            fps = 1.0 / (end_time - start_time)

        for i in carLocation1.keys():
            if frameCounter % 1 == 0:
                [x1, y1, w1, h1] = carLocation1[i]
                [x2, y2, w2, h2] = carLocation2[i]

                carLocation1[i] = [x2, y2, w2, h2]

                if [x1, y1, w1, h1] != [x2, y2, w2, h2]:
                    if speed[i] is None or speed[i] == 0:
                        speed[i] = estimateSpeed([x1, y1, w1, h1], [x2, y2, w2, h2], fps)

                    if speed[i] != 0 and speed[i] < 150:
                        update_highest_speed(speed[i])

                    cv2.putText(resultImage,str(int(speed[i])) + " km/hr", (int(x1 + w1 / 2), int(y1 - 10)),cv2.FONT_HERSHEY_SIMPLEX, 0.75,(0, 255, 0),2,)

        # 更新车流量
        prev_total_cars = total_cars
        total_cars = len(carTracker)

        # 更新实时车流量标签
        traffic_label.config(text="实时车流量：{}".format(total_cars))

        update_video_frame(resultImage)

    video.release()

该代码中首先，定义了一些变量和数据结构用于跟踪车辆。rectangleColor是绘制车辆边界框的颜色，frameCounter用于计算帧数，currentCarID表示当前车辆的ID，fps表示帧率。carTracker是一个字典，用于存储每辆车的追踪器对象。carLocation1和carLocation2分别是字典，用于存储每辆车的当前和上一帧的位置信息。speed是一个列表，用于存储每辆车的速度。update_highest_speed()是一个内部函数，用于更新最高速度的显示。update_video_frame()是一个内部函数，用于更新视频帧的显示。total_cars和prev_total_cars是用于计算车流量的变量。进入主循环，首先读取视频的一帧图像。如果无法读取到图像，就退出循环。对读取到的图像进行预处理，包括调整图像大小和创建一个副本。增加帧计数器的值。创建一个空列表carIDtoDelete，用于存储需要删除的车辆的ID。遍历已跟踪的车辆，更新其位置信息和质量。如果某辆车的跟踪质量低于阈值（7），则将其ID添加到carIDtoDelete列表中。遍历carIDtoDelete列表，从carTracker、carLocation1和carLocation2中删除对应的车辆。每隔10帧，检测图像中的车辆。使用级联分类器（carCascade）对灰度图像进行目标检测，并返回检测到的车辆的边界框。遍历检测到的车辆边界框，与已跟踪的车辆进行匹配。如果找到匹配的车辆，更新该车辆的追踪器对象，位置信息和ID。如果没有找到匹配的车辆，创建一个新的追踪器对象，将其添加到carTracker中，并更新位置信息和ID。遍历carTracker中的每辆车，获取其位置信息，并在结果图像上绘制车辆边界框。根据每辆车的位置信息和帧率，估计车辆的速度，并将其存储在speed列表中。如果速度不为0且小于150，更新最高速度。在结果图像上显示每辆车的速

5.实现open_file()函数用于打开文件对话框并选择视频文件。选择视频文件后，将启动一个新线程来处理视频。

def open_file():
    global video_path, video
    file_types = [("视频文件", "*.mp4;*.avi;*.mkv")]  # 允许上传的视频文件类型
    video_path = filedialog.askopenfilename(title="选择视频文件", filetypes=file_types)
    if video_path:
        video = cv2.VideoCapture(video_path)
        t = threading.Thread(target=process_video)
        t.start()

6. 实现process_video()函数在新线程中执行，它首先通过cv2.VideoCapture打开选择的视频文件，然后调用track_multiple_objects()函数对视频进行处理。

7.实现update_highest_speed(speed)函数用于更新最高时速度的全局变量，并更新显示最高时速度的标签。

8.实现update_video_frame(image)函数用于更新显示视频帧的标签，它将图像转换为Tkinter可用的格式并进行适当的调整。

9.最后，代码创建了一个Tkinter窗口，并在窗口中添加了按钮、车流量标签、最高时速度标签和视频显示标签。通过window.mainloop()来启动GUI窗口的事件循环，使应用保持运行状态。实现的界面如下所示：