Counting objects from different zones of several RTSP streams.

[epigraphe]

Hello.
I was able to implement a project on in-zone detection, counting and further processing. This works great with a single rtsp stream. Unfortunately, I encountered the fact that if I try to use two threads, everything freezes.
I read that yolo8 can be used for multiple threads

# Multiple streams with batched inference (i.e. batch-size 8 for 8 streams)
source = 'path/to/list.streams'  # *.streams text file with one streaming address per row

Can I involve supervision in this?
How can i make that all streams have their own zones and have their own counting?

Ultimately, I need to do the following: this is a video wall with 4-8 cameras, each of which defines its own zones and counts objects in these zones.

I chose Django to display the final image in realtime. I can show the code of what works for me with one thread if that helps.

This is where I am now. Now I'm stuck.
I can't figure out how to extend the detection, counting and tracking logic to all streams from a file

import numpy as np
import cv2
from ultralytics import YOLO
import supervision as sv
import queue
import threading
import time

def format_time(seconds):
    hours = seconds // 3600
    minutes = (seconds % 3600) // 60
    seconds = seconds % 60
    return f"{int(hours):02}:{int(minutes):02}:{int(seconds):02}"

class CountObject():
    def __init__(self, rtsp_url, frame_resolution_wh=(2560, 1440)):
        self.model = YOLO("yolov8s.pt")
        self.model.predict("sources.list", show=False, stream=True, classes=0, conf=0.9)
        self.tracker = sv.ByteTrack()
        self.colors = sv.ColorPalette.default()
        self.box_annotator = sv.BoundingBoxAnnotator()
        self.rtsp_url = rtsp_url
        self.label_annotator = sv.LabelAnnotator(text_scale=2.0, text_thickness=4)

#        self.polygons = np.array([
#            [[0, 0], [0.4, 0], [0.4, 1], [0, 1]],
#            [[0.6, 0], [1, 0], [1, 1], [0.6, 1]]
#        ])

        self.polygons = np.array([
            [[0.6, 0.39],[0.75, 0.39],[0.76, 0.57],[0.6, 0.58]],
            [[0.02, 0.62],[0.26, 0.61],[0.26, 0.97],[0.02, 0.97]],
            [[0.41, 0.37],[0.41, 0.68],[0.54, 0.67],[0.53, 0.38]],
            [[0.77, 0.66],[0.77, 0.99],[0.99, 0.98],[0.98, 0.64]]
        ])


        self.polygons = (self.polygons * np.array(frame_resolution_wh)).astype(int)
        self.zones = [sv.PolygonZone(polygon=polygon, frame_resolution_wh=frame_resolution_wh) for polygon in self.polygons]
        self.zone_annotators = [sv.PolygonZoneAnnotator(zone=zone, color=self.colors.by_idx(index), thickness=6, text_thickness=8, text_scale=4) for index, zone in enumerate(self.zones)]
        self.box_annotators = [sv.BoxAnnotator(color=self.colors.by_idx(index), thickness=4, text_thickness=4, text_scale=2) for index in range(len(self.polygons))]

        # Dictionary to store entry times for objects in each zone
        self.enter_times = {index: {} for index in range(len(self.zones))}

        # Initialize the last write time for each zone
        self.last_write_times = [time.time() for _ in range(len(self.zones))]

    def process_frame(self, frame: np.ndarray) -> np.ndarray:
        results = self.model(frame)[0]
        detections = sv.Detections.from_ultralytics(results)
        detections = self.tracker.update_with_detections(detections)

        # Annotated Frame
        annotated_frame = frame.copy()

        for index, zone in enumerate(self.zones):
            zone_value = zone.current_count

            # Check if 10 seconds have passed since the last write
            current_time = time.time()
            if current_time - self.last_write_times[index] >= 10:
                # Write the zone_value to the file
                write_zone_value_to_file(index, zone_value)
                # Update the last write time
                self.last_write_times[index] = current_time

            # Filter Object in Zone
            mask = zone.trigger(detections=detections)
            detections_filtered = detections[mask]

            # Annotate
            annotated_frame = self.zone_annotators[index].annotate(annotated_frame)

            if len(detections_filtered) > 0:
                # If object in zone
                labels = []
                for class_id, tracker_id in zip(detections_filtered.class_id, detections_filtered.tracker_id):
                    # Check object in zone
                    if tracker_id not in self.enter_times[index]:
                        self.enter_times[index][tracker_id] = time.time()

                    # Get object time
                    entry_time = self.enter_times[index][tracker_id]
                    current_time = time.time()
                    time_difference = current_time - entry_time
                    formatted_time = format_time(time_difference)

                    # Annotate Time
                    labels.append(f"#{tracker_id} - {formatted_time}")

                annotated_frame = self.box_annotators[index].annotate(annotated_frame, detections=detections_filtered, skip_label=True)
                annotated_frame = self.label_annotator.annotate(annotated_frame, detections=detections_filtered, labels=labels)

        return annotated_frame

def write_zone_value_to_file(zone_number, zone_value):
    filename = f"zone_{zone_number}.tmp"
    current_time = time.strftime("%Y-%m-%d %H:%M:%S")
    with open(filename, "a") as file:
        file.write(f"{current_time} - Zone Value: {zone_value}\n")

I also have a function for one stream, but I think it is not needed if using sources.list.

class VideoCapture:
    def __init__(self, rtsp_url):
        self.cap = cv2.VideoCapture(rtsp_url)
        self.q = queue.Queue()
        t = threading.Thread(target=self._reader)
        t.daemon = True
        t.start()

    def _reader(self):
        while True:
            ret, frame = self.cap.read()
            if not ret:
                break
            if not self.q.empty():
                try:
                    self.q.get_nowait()
                except queue.Empty:
                    pass
            self.q.put(frame)

    def read(self):
        return self.q.get()

[epigraphe]

If I use the clean function, everything works at excellent speed:
How can I implement zones, detection, tracking and counting into it?

cv2.cuda.printCudaDeviceInfo(0)
model = YOLO("yolov8n.pt")
results = model("list.streams", stream=True, classes=0, conf=0.5)
video_placeholders = [st.empty() for _ in range(3)]
for i, result in enumerate(results):
    image = result.orig_img
    resized_image = cv2.resize(image, (1024, 768))
    video_placeholders[i % 3].image(resized_image, channels="BGR", use_column_width=True, caption=f"Video {i+1}")

I came up with this option, it displays what I need. But the video display is slow.

I realized he's not slowing down. It shows every third frame. I have three streams in the file and it turns out that it processes not all frames in the stream, but one by one in each. I'm even more confused(

I still really need help with this:

import streamlit as st
import time
import numpy as np
import cv2
from ultralytics import YOLO
import supervision as sv

def format_time(seconds):

def write_zone_value_to_file(zone_number, zone_value):

class CountObject:
    def __init__(self, frame_resolution_wh=(2560, 1440)):
        self.model = YOLO("yolov8s.pt")
        self.tracker = sv.ByteTrack()
        self.colors = sv.ColorPalette.default()
        self.box_annotator = sv.BoundingBoxAnnotator()
        self.label_annotator = sv.LabelAnnotator(text_scale=2.0, text_thickness=4)

        self.polygons = np.array([
            [[0.6, 0.39],[0.75, 0.39],[0.76, 0.57],[0.6, 0.58]],
            [[0.02, 0.62],[0.26, 0.61],[0.26, 0.97],[0.02, 0.97]],
            [[0.41, 0.37],[0.41, 0.68],[0.54, 0.67],[0.53, 0.38]],
            [[0.77, 0.66],[0.77, 0.99],[0.99, 0.98],[0.98, 0.64]]
        ])

        self.polygons = (self.polygons * np.array(frame_resolution_wh)).astype(int)
        self.zones = [sv.PolygonZone(polygon=polygon, frame_resolution_wh=frame_resolution_wh) for polygon in self.polygons]
        self.zone_annotators = [sv.PolygonZoneAnnotator(zone=zone, color=self.colors.by_idx(index), thickness=6, text_thickness=8, text_scale=4) for index, zone in enumerate(self.zones)]
        self.box_annotators = [sv.BoxAnnotator(color=self.colors.by_idx(index), thickness=4, text_thickness=4, text_scale=2) for index in range(len(self.polygons))]

        self.enter_times = {index: {} for index in range(len(self.zones))}
        self.last_write_times = [time.time() for _ in range(len(self.zones))]

    def process_frame(self, frame: np.ndarray) -> np.ndarray:
        results = self.model(frame)[0]
        detections = sv.Detections.from_ultralytics(results)
        detections = self.tracker.update_with_detections(detections)
        annotated_frame = frame.copy()

        for index, zone in enumerate(self.zones):
            zone_value = zone.current_count

            current_time = time.time()
            if current_time - self.last_write_times[index] >= 10:
                write_zone_value_to_file(index, zone_value)
                self.last_write_times[index] = current_time

            mask = zone.trigger(detections=detections)
            detections_filtered = detections[mask]
            annotated_frame = self.zone_annotators[index].annotate(annotated_frame)

            if len(detections_filtered) > 0:
                labels = []
                for class_id, tracker_id in zip(detections_filtered.class_id, detections_filtered.tracker_id):
                    if tracker_id not in self.enter_times[index]:
                        self.enter_times[index][tracker_id] = time.time()

                    entry_time = self.enter_times[index][tracker_id]
                    current_time = time.time()
                    time_difference = current_time - entry_time
                    formatted_time = format_time(time_difference)
                    labels.append(f"#{tracker_id} - {formatted_time}")

                annotated_frame = self.box_annotators[index].annotate(annotated_frame, detections=detections_filtered, skip_label=True)
                annotated_frame = self.label_annotator.annotate(annotated_frame, detections=detections_filtered, labels=labels)

        return annotated_frame

#  YOLO Init
model = YOLO("yolov8n.pt")

# Result for many streams
results = model("list.streams", stream=True, classes=0, conf=0.5)

# StreamlitPlaceholders
video_placeholders = [st.empty() for _ in range(3)]

# Initialize the class
count_object = CountObject()

# 
for i, result in enumerate(results):
    # 
    image = result.orig_img

    # Processing CountObject
    annotated_image = count_object.process_frame(image)

    # Annotated Video
    video_placeholders[i % 3].image(annotated_image, channels="BGR", use_column_width=True, caption=f"Video {i+1}")

[epigraphe]

Solved

[SkalskiP]

Awesome! Great to see you managed to solve it on your own. 🎉