Skip to content

Reference for ultralytics/solutions/queue_management.py

Note

This file is available at https://github.com/ultralytics/ultralytics/blob/main/ultralytics/solutions/queue_management.py. If you spot a problem please help fix it by contributing a Pull Request 🛠️. Thank you 🙏!


ultralytics.solutions.queue_management.QueueManager 

QueueManager(
    names, reg_pts=None, line_thickness=2, view_img=False, draw_tracks=False
)

A class to manage the queue in a real-time video stream based on object tracks.

Parameters:

Name Type Description Default
names dict

A dictionary mapping class IDs to class names.

 required
reg_pts list of tuples

Points defining the counting region polygon. Defaults to a predefined rectangle.

 None
line_thickness int

Thickness of the annotation lines. Defaults to 2.

 2
view_img bool

Whether to display the image frames. Defaults to False.

 False
draw_tracks bool

Whether to draw tracks of the objects. Defaults to False.

 False
Source code in ultralytics/solutions/queue_management.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    names,
    reg_pts=None,
    line_thickness=2,
    view_img=False,
    draw_tracks=False,
):
    """
    Initializes the QueueManager with specified parameters for tracking and counting objects.

    Args:
        names (dict): A dictionary mapping class IDs to class names.
        reg_pts (list of tuples, optional): Points defining the counting region polygon. Defaults to a predefined
            rectangle.
        line_thickness (int, optional): Thickness of the annotation lines. Defaults to 2.
        view_img (bool, optional): Whether to display the image frames. Defaults to False.
        draw_tracks (bool, optional): Whether to draw tracks of the objects. Defaults to False.
    """
    # Region & Line Information
    self.reg_pts = reg_pts if reg_pts is not None else [(20, 60), (20, 680), (1120, 680), (1120, 60)]
    self.counting_region = (
        Polygon(self.reg_pts) if len(self.reg_pts) >= 3 else Polygon([(20, 60), (20, 680), (1120, 680), (1120, 60)])
    )

    # annotation Information
    self.tf = line_thickness
    self.view_img = view_img

    self.names = names  # Class names

    # Object counting Information
    self.counts = 0

    # Tracks info
    self.track_history = defaultdict(list)
    self.draw_tracks = draw_tracks

    # Check if environment supports imshow
    self.env_check = check_imshow(warn=True)

extract_and_process_tracks 

extract_and_process_tracks(tracks, im0)

Extracts and processes tracks for queue management in a video stream.

Source code in ultralytics/solutions/queue_management.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def extract_and_process_tracks(self, tracks, im0):
    """Extracts and processes tracks for queue management in a video stream."""
    # Initialize annotator and draw the queue region
    annotator = Annotator(im0, self.tf, self.names)
    self.counts = 0  # Reset counts every frame
    if tracks[0].boxes.id is not None:
        boxes = tracks[0].boxes.xyxy.cpu()
        clss = tracks[0].boxes.cls.cpu().tolist()
        track_ids = tracks[0].boxes.id.int().cpu().tolist()

        # Extract tracks
        for box, track_id, cls in zip(boxes, track_ids, clss):
            # Draw bounding box
            annotator.box_label(box, label=self.names[cls], color=colors(int(track_id), True))

            # Update track history
            track_line = self.track_history[track_id]
            track_line.append((float((box[0] + box[2]) / 2), float((box[1] + box[3]) / 2)))
            if len(track_line) > 30:
                track_line.pop(0)

            # Draw track trails if enabled
            if self.draw_tracks:
                annotator.draw_centroid_and_tracks(
                    track_line,
                    color=colors(int(track_id), True),
                    track_thickness=self.line_thickness,
                )

            prev_position = self.track_history[track_id][-2] if len(self.track_history[track_id]) > 1 else None

            # Check if the object is inside the counting region
            if len(self.reg_pts) >= 3:
                is_inside = self.counting_region.contains(Point(track_line[-1]))
                if prev_position is not None and is_inside:
                    self.counts += 1

    # Display queue counts
    label = f"Queue Counts : {str(self.counts)}"
    if label is not None:
        annotator.queue_counts_display(
            label,
            points=self.reg_pts,
            region_color=(255, 0, 255),
            txt_color=(104, 31, 17),
        )

    if self.env_check and self.view_img:
        annotator.draw_region(reg_pts=self.reg_pts, thickness=self.tf * 2, color=(255, 0, 255))
        cv2.imshow("Ultralytics YOLOv8 Queue Manager", im0)
        # Close window on 'q' key press
        if cv2.waitKey(1) & 0xFF == ord("q"):
            return

process_queue 

process_queue(im0, tracks)

Main function to start the queue management process.

Parameters:

Name Type Description Default
im0 ndarray

Current frame from the video stream.

 required
tracks list

List of tracks obtained from the object tracking process.

 required
Source code in ultralytics/solutions/queue_management.py 
113
114
115
116
117
118
119
120
121
122
def process_queue(self, im0, tracks):
    """
    Main function to start the queue management process.

    Args:
        im0 (ndarray): Current frame from the video stream.
        tracks (list): List of tracks obtained from the object tracking process.
    """
    self.extract_and_process_tracks(tracks, im0)  # Extract and process tracks
    return im0




📅 Created 5 months ago ✏️ Updated 6 days ago
jk4e glenn-jocher Burhan-Q