People Counting System Development

System Development подсчёта посетителей (People Counting)

Системы подсчёта посетителей используются в ритейле (footfall analytics), музеях, аэропортах, транспортных узлах для понимания потоков людей, оценки эффективности пространства, управления очередями, compliance (ограничение наполняемости). Точность коммерческих систем — 95–99% при хорошем освещении.

Подход: верхний обзор + виртуальная линия

Оптимальное расположение камеры — сверху (top-view), перпендикулярно полу. Минимизирует перекрытия, упрощает детекцию. Виртуальная линия в кадре — граница для подсчёта входящих/выходящих.

from ultralytics import YOLO
import numpy as np
import cv2

class PeopleCounter:
    def __init__(self, model_path: str,
                 count_line: tuple,        # ((x1,y1), (x2,y2))
                 direction: str = 'both'):  # 'in', 'out', 'both'
        self.model = YOLO(model_path)
        self.count_line = count_line
        self.direction = direction

        # ByteTrack встроен в Ultralytics
        self.tracker_config = 'bytetrack.yaml'

        self.track_history = {}
        self.count_in = 0
        self.count_out = 0
        self.counted_ids = set()

    def process(self, frame: np.ndarray) -> dict:
        # Детекция людей с трекингом
        results = self.model.track(
            frame,
            persist=True,
            conf=0.4,
            classes=[0],             # только люди
            tracker=self.tracker_config
        )

        if results[0].boxes.id is None:
            return self._get_counts()

        for box, track_id in zip(results[0].boxes.xyxy,
                                   results[0].boxes.id):
            tid = int(track_id)
            x1, y1, x2, y2 = map(int, box)
            cx, cy = (x1 + x2) // 2, (y1 + y2) // 2

            if tid not in self.track_history:
                self.track_history[tid] = []
            self.track_history[tid].append((cx, cy))

            # Проверяем пересечение линии
            if len(self.track_history[tid]) >= 2 and tid not in self.counted_ids:
                prev_pos = self.track_history[tid][-2]
                curr_pos = self.track_history[tid][-1]

                crossing = self._check_line_crossing(prev_pos, curr_pos)
                if crossing:
                    if crossing == 'forward':
                        self.count_in += 1
                    else:
                        self.count_out += 1
                    self.counted_ids.add(tid)

        return self._get_counts()

    def _check_line_crossing(self, prev: tuple, curr: tuple) -> str | None:
        """Определение факта и направления пересечения линии"""
        lx1, ly1 = self.count_line[0]
        lx2, ly2 = self.count_line[1]

        # Векторное произведение для определения стороны
        d1 = self._cross_product(prev, (lx1, ly1), (lx2, ly2))
        d2 = self._cross_product(curr, (lx1, ly1), (lx2, ly2))

        if d1 * d2 < 0:  # пересечение
            return 'forward' if d1 < 0 else 'backward'
        return None

    def _cross_product(self, point, line_start, line_end):
        return ((line_end[0] - line_start[0]) * (point[1] - line_start[1]) -
                (line_end[1] - line_start[1]) * (point[0] - line_start[0]))

    def _get_counts(self) -> dict:
        return {
            'count_in': self.count_in,
            'count_out': self.count_out,
            'current_occupancy': self.count_in - self.count_out
        }

Тепловая карта передвижений

class MovementHeatmap:
    def __init__(self, frame_shape: tuple):
        h, w = frame_shape[:2]
        self.accumulator = np.zeros((h, w), dtype=np.float32)
        self.decay = 0.995  # забываем старые данные

    def update(self, track_positions: list[tuple]):
        self.accumulator *= self.decay
        for x, y in track_positions:
            if 0 <= x < self.accumulator.shape[1] and \
               0 <= y < self.accumulator.shape[0]:
                self.accumulator[y, x] += 1.0

        # Gaussian blur для сглаживания
        self.accumulator = cv2.GaussianBlur(
            self.accumulator, (21, 21), 0
        )

    def get_heatmap(self, frame: np.ndarray) -> np.ndarray:
        normalized = cv2.normalize(
            self.accumulator, None, 0, 255, cv2.NORM_MINMAX
        ).astype(np.uint8)
        colormap = cv2.applyColorMap(normalized, cv2.COLORMAP_JET)
        return cv2.addWeighted(frame, 0.6, colormap, 0.4, 0)

Аналитика и отчётность

Данные подсчёта идут в time-series базу (InfluxDB) и доступны в Grafana:

• Трафик за день/неделю/месяц
• Пиковые часы посещаемости
• Конверсионная воронка (по зонам)
• Соответствие ограничениям наполняемости

Точность систем подсчёта