深度学习实战】行人检测追踪与双向流量计数系统【python源码+Pyqt5界面+数据集+训练代码】YOLOv8、ByteTrack、目标追踪、双向计数、行人检测追踪、过线计数(3)

简介: 深度学习实战】行人检测追踪与双向流量计数系统【python源码+Pyqt5界面+数据集+训练代码】YOLOv8、ByteTrack、目标追踪、双向计数、行人检测追踪、过线计数

深度学习实战】行人检测追踪与双向流量计数系统【python源码+Pyqt5界面+数据集+训练代码】YOLOv8、ByteTrack、目标追踪、双向计数、行人检测追踪、过线计数(2)https://developer.aliyun.com/article/1536938

ByteTrack的实现代码如下:

class ByteTrack:
    """
    Initialize the ByteTrack object.
    Parameters:
        track_thresh (float, optional): Detection confidence threshold
            for track activation.
        track_buffer (int, optional): Number of frames to buffer when a track is lost.
        match_thresh (float, optional): Threshold for matching tracks with detections.
        frame_rate (int, optional): The frame rate of the video.
    """
    def __init__(
        self,
        track_thresh: float = 0.25,
        track_buffer: int = 30,
        match_thresh: float = 0.8,
        frame_rate: int = 30,
    ):
        self.track_thresh = track_thresh
        self.match_thresh = match_thresh
        self.frame_id = 0
        self.det_thresh = self.track_thresh + 0.1
        self.max_time_lost = int(frame_rate / 30.0 * track_buffer)
        self.kalman_filter = KalmanFilter()
        self.tracked_tracks: List[STrack] = []
        self.lost_tracks: List[STrack] = []
        self.removed_tracks: List[STrack] = []
    def update_with_detections(self, detections: Detections) -> Detections:
        """
        Updates the tracker with the provided detections and
            returns the updated detection results.
        Parameters:
            detections: The new detections to update with.
        Returns:
            Detection: The updated detection results that now include tracking IDs.
        """
        tracks = self.update_with_tensors(
            tensors=detections2boxes(detections=detections)
        )
        detections = Detections.empty()
        if len(tracks) > 0:
            detections.xyxy = np.array(
                [track.tlbr for track in tracks], dtype=np.float32
            )
            detections.class_id = np.array(
                [int(t.class_ids) for t in tracks], dtype=int
            )
            detections.tracker_id = np.array(
                [int(t.track_id) for t in tracks], dtype=int
            )
            detections.confidence = np.array(
                [t.score for t in tracks], dtype=np.float32
            )
        else:
            detections.tracker_id = np.array([], dtype=int)
        return detections
    def update_with_tensors(self, tensors: np.ndarray) -> List[STrack]:
        """
        Updates the tracker with the provided tensors and returns the updated tracks.
        Parameters:
            tensors: The new tensors to update with.
        Returns:
            List[STrack]: Updated tracks.
        """
        self.frame_id += 1
        activated_starcks = []
        refind_stracks = []
        lost_stracks = []
        removed_stracks = []
        class_ids = tensors[:, 5]
        scores = tensors[:, 4]
        bboxes = tensors[:, :4]
        remain_inds = scores > self.track_thresh
        inds_low = scores > 0.1
        inds_high = scores < self.track_thresh
        inds_second = np.logical_and(inds_low, inds_high)
        dets_second = bboxes[inds_second]
        dets = bboxes[remain_inds]
        scores_keep = scores[remain_inds]
        scores_second = scores[inds_second]
        class_ids_keep = class_ids[remain_inds]
        class_ids_second = class_ids[inds_second]
        if len(dets) > 0:
            """Detections"""
            detections = [
                STrack(STrack.tlbr_to_tlwh(tlbr), s, c)
                for (tlbr, s, c) in zip(dets, scores_keep, class_ids_keep)
            ]
        else:
            detections = []
        """ Add newly detected tracklets to tracked_stracks"""
        unconfirmed = []
        tracked_stracks = []  # type: list[STrack]
        for track in self.tracked_tracks:
            if not track.is_activated:
                unconfirmed.append(track)
            else:
                tracked_stracks.append(track)
        """ Step 2: First association, with high score detection boxes"""
        strack_pool = joint_tracks(tracked_stracks, self.lost_tracks)
        # Predict the current location with KF
        STrack.multi_predict(strack_pool)
        dists = matching.iou_distance(strack_pool, detections)
        dists = matching.fuse_score(dists, detections)
        matches, u_track, u_detection = matching.linear_assignment(
            dists, thresh=self.match_thresh
        )
        for itracked, idet in matches:
            track = strack_pool[itracked]
            det = detections[idet]
            if track.state == TrackState.Tracked:
                track.update(detections[idet], self.frame_id)
                activated_starcks.append(track)
            else:
                track.re_activate(det, self.frame_id, new_id=False)
                refind_stracks.append(track)
        """ Step 3: Second association, with low score detection boxes"""
        # association the untrack to the low score detections
        if len(dets_second) > 0:
            """Detections"""
            detections_second = [
                STrack(STrack.tlbr_to_tlwh(tlbr), s, c)
                for (tlbr, s, c) in zip(dets_second, scores_second, class_ids_second)
            ]
        else:
            detections_second = []
        r_tracked_stracks = [
            strack_pool[i]
            for i in u_track
            if strack_pool[i].state == TrackState.Tracked
        ]
        dists = matching.iou_distance(r_tracked_stracks, detections_second)
        matches, u_track, u_detection_second = matching.linear_assignment(
            dists, thresh=0.5
        )
        for itracked, idet in matches:
            track = r_tracked_stracks[itracked]
            det = detections_second[idet]
            if track.state == TrackState.Tracked:
                track.update(det, self.frame_id)
                activated_starcks.append(track)
            else:
                track.re_activate(det, self.frame_id, new_id=False)
                refind_stracks.append(track)
        for it in u_track:
            track = r_tracked_stracks[it]
            if not track.state == TrackState.Lost:
                track.mark_lost()
                lost_stracks.append(track)
        """Deal with unconfirmed tracks, usually tracks with only one beginning frame"""
        detections = [detections[i] for i in u_detection]
        dists = matching.iou_distance(unconfirmed, detections)
        dists = matching.fuse_score(dists, detections)
        matches, u_unconfirmed, u_detection = matching.linear_assignment(
            dists, thresh=0.7
        )
        for itracked, idet in matches:
            unconfirmed[itracked].update(detections[idet], self.frame_id)
            activated_starcks.append(unconfirmed[itracked])
        for it in u_unconfirmed:
            track = unconfirmed[it]
            track.mark_removed()
            removed_stracks.append(track)
        """ Step 4: Init new stracks"""
        for inew in u_detection:
            track = detections[inew]
            if track.score < self.det_thresh:
                continue
            track.activate(self.kalman_filter, self.frame_id)
            activated_starcks.append(track)
        """ Step 5: Update state"""
        for track in self.lost_tracks:
            if self.frame_id - track.end_frame > self.max_time_lost:
                track.mark_removed()
                removed_stracks.append(track)
        self.tracked_tracks = [
            t for t in self.tracked_tracks if t.state == TrackState.Tracked
        ]
        self.tracked_tracks = joint_tracks(self.tracked_tracks, activated_starcks)
        self.tracked_tracks = joint_tracks(self.tracked_tracks, refind_stracks)
        self.lost_tracks = sub_tracks(self.lost_tracks, self.tracked_tracks)
        self.lost_tracks.extend(lost_stracks)
        self.lost_tracks = sub_tracks(self.lost_tracks, self.removed_tracks)
        self.removed_tracks.extend(removed_stracks)
        self.tracked_tracks, self.lost_tracks = remove_duplicate_tracks(
            self.tracked_tracks, self.lost_tracks
        )
        output_stracks = [track for track in self.tracked_tracks if track.is_activated]
        return output_stracks

使用方法

1.创建ByteTrack跟踪器

# 创建跟踪器
byte_tracker = sv.ByteTrack(track_thresh=0.25, track_buffer=30, match_thresh=0.8, frame_rate=30)

2.对YOLOv8的目标检测结果进行追踪

model = YOLO(path)
results = model(frame)[0]
detections = sv.Detections.from_ultralytics(results)
detections = byte_tracker.update_with_detections(detections)

3.显示追踪结果ID、检测框及标签信息

labels = [
            f"id{tracker_id} {model.model.names[class_id]}"
            for _, _, confidence, class_id, tracker_id
            in detections
        ]
annotated_frame = frame.copy()
annotated_frame = box_annotator.annotate(
            scene=annotated_frame,
            detections=detections,
            labels=labels)

最终检测效果如下:

四、过线计数判断方式

定义过线线段

定义用于统计过线的线段,代码如下:

cap = cv2.VideoCapture(video_path)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
point_A = [10, int(height/5*4)]
point_B = [width-10, int(height/5)]
# 定义过线使用的线段点
LINE_START = sv.Point(point_A[0], point_A[1])
LINE_END = sv.Point(point_B[0], point_B[1])
line_zone = MyLineZone(start=LINE_START, end=LINE_END)

判断过线方法

使用目标中心点判断是否过线,核心代码如下:

for i, (xyxy, _, confidence, class_id, tracker_id) in enumerate(detections):
            if tracker_id is None:
                continue
            # 使用中心点判断是否过线
            x1, y1, x2, y2 = xyxy
            center_x = int((x1 + x2) / 2)
            center_y = int((y1 + y2) / 2)
            center_point = Point(x=center_x, y=center_y)
            triggers = [self.vector.is_in(point=center_point)]

上述通过目标框坐标计算出目标中心点坐标center_x ,center_y ,然后通过is_in函数判断过线状态,其中is_in函数定义如下:

def is_in(self, point: Point) -> bool:
        v1 = Vector(self.start, self.end)
        v2 = Vector(self.start, point)
        cross_product = (v1.end.x - v1.start.x) * (v2.end.y - v2.start.y) - (
            v1.end.y - v1.start.y
        ) * (v2.end.x - v2.start.x)
        return cross_product < 0

函数首先根据线段的起点和终点构造两个向量v1和v2,分别表示线段和待判断的点与线段起点的向量。然后计算两个向量的叉积,并判断叉积的正负来确定点的位置关系。若叉积小于0,则点在线段的左侧;若叉积大于0,则点在线段的右侧;若叉积等于0,则点在线段上。根据题设,函数返回的是点在线段不同侧的状态,即当叉积小于0时返回True,否则返回False

判断是否通过线段

上述判断方式只能用于判断目标是否通过线段所在直线,并不是在线段内通过。如果想判断在线段内通过,需要另外加上过线时的判断条件,核心代码如下:

def point_in_line(self, center_point):
        # 判断点是否在线段之间通过
        # 计算向量 AP 与向量 AB 的点积(也称为“标量积”)
        # 点积的绝对值应在 0(包括端点)与向量 AB 的模长平方之间,且方向应与 AB 相同(即点积为正)
        point_A, point_B = self.get_line_points(self.vector)
        xA, yA = point_A
        xB, yB = point_B
        xP, yP = center_point
        AB = (xB - xA, yB - yA)
        AP = (xP - xA, yP - yA)
        # 计算向量 AP 与向量 AB 的点积
        dot_product = AB[0] * AP[0] + AB[1] * AP[1]
        # 计算向量 AB 模长的平方
        AB_length_squared = AB[0] ** 2 + AB[1] ** 2
        # 判断标准:点积的绝对值应在 0(包括端点)与向量 AB 的模长平方之间,且方向应与 AB 相同(即点积为正)
        if 0 <= dot_product <= AB_length_squared and dot_product >= 0:
            within_segment = True
        else:
            within_segment = False
        return within_segment

判断点是否在线段之间通过,通过计算向量 AP向量 AB点积(也称为“标量积”)来进行判断。其中P表示目标中心点,AB表示目标需要通过的线段。

判断标准:点积的绝对值应在 0(包括端点)与向量 AB 的模长平方之间,且方向应与 AB 相同(即点积为正),则表示在线段内通过。

过线效果展示

过线效果展示如下:

以上便是关于此款行人检测追踪与双向流量计数系统的原理与代码介绍。基于以上内容,博主用pythonPyqt5开发了一个带界面的软件系统,即文中第二部分的演示内容,能够很好的通过视频及摄像头进行检测追踪,以及自定义过线计数

相关文章
|
3天前
|
并行计算 算法 Python
Dantzig-Wolfe分解算法解释与Python代码示例
Dantzig-Wolfe分解算法解释与Python代码示例
|
3天前
|
存储 Python
离散事件模拟(Discrete Event Simulation)详解与Python代码示例
离散事件模拟(Discrete Event Simulation)详解与Python代码示例
|
3天前
|
供应链 Python
供需匹配(Demand-Supply Matching)的详细解释与Python代码示例
供需匹配(Demand-Supply Matching)的详细解释与Python代码示例
|
3天前
|
供应链 Python
Demand Forecasting模型解释与Python代码示例
Demand Forecasting模型解释与Python代码示例
|
3天前
|
存储 Python
数据包络分析(Data Envelopment Analysis, DEA)详解与Python代码示例
数据包络分析(Data Envelopment Analysis, DEA)详解与Python代码示例
|
23小时前
|
存储 Python
深度剖析:Python里字典树Trie的构建与查询,让你的代码更优雅!
【7月更文挑战第20天】Trie树(前缀树)是高效处理字符串搜索的 数据结构**。通过Python实现,每个节点含指向子节点的链接(字典)和结束标识。`TrieNode`和`Trie`类分别表示节点和树,支持插入、搜索和前缀检查。空间效率高,共享公共前缀,时间复杂度O(m)。适用于字符串集合的快速检索和灵活扩展,如自动补全。学习和应用Trie能提升代码效率和质量。
6 0
|
2天前
|
机器学习/深度学习 人工智能 算法
深度学习在医疗影像诊断中的应用与挑战
随着人工智能技术的飞速发展,深度学习在医疗影像诊断领域中展现出前所未有的潜力。本文将深入探讨深度学习技术如何革新传统的医疗影像分析流程,提高诊断的准确性和效率,并讨论实施过程中遇到的主要挑战。通过对比分析深度学习模型与传统方法在处理医学影像数据时的性能差异,揭示深度学习在图像识别、特征提取和模式分类方面的优势。同时,指出数据隐私、模型解释性及跨域泛化能力等关键问题,为未来研究提供方向。
|
1天前
|
机器学习/深度学习 监控 自动驾驶
探索深度学习在图像识别中的应用
【7月更文挑战第20天】本文深入探讨了深度学习技术在图像识别领域的应用,并分析了其背后的关键技术和挑战。通过实例分析,文章揭示了深度学习如何提高图像处理的准确度和效率,并预测了未来技术的发展趋势。
|
1天前
|
机器学习/深度学习 人工智能 计算机视觉
探索深度学习在图像识别中的应用
【7月更文挑战第20天】随着人工智能技术的飞速发展,深度学习已成为推动图像识别技术革新的核心动力。本文旨在探讨深度学习如何优化图像识别过程,并分析其在多个领域的应用前景。通过案例研究,我们将揭示深度学习模型在处理复杂图像数据时的强大能力及其潜在的挑战。