Разработка AI-системы детекции отсутствия СИЗ у строителей

AI-детекция отсутствия средств индивидуальной защиты (СИЗ)

Узкая, но критически важная задача: автоматически определять, надета ли на рабочего каска, жилет, очки, перчатки. Звучит просто, пока не сталкиваешься с реальными условиями: рабочий в тени, боком, на высоте 20м, в клубах пыли.

Датасеты и модели для PPE detection

Открытые датасеты: Safety Helmet Detection Dataset (7000+ изображений), PPE-Detection Dataset (Roboflow, 8000+ изображений), CHV (Construction Helmet and Vest).

Для production всегда нужно дообучение на собственных данных. Разные стройплощадки — разные типы касок, жилетов, освещения.

from ultralytics import YOLO
import cv2
import numpy as np
from collections import defaultdict

class PPEDetector:
    """
    Модель обнаруживает как наличие, так и отсутствие СИЗ напрямую.
    Классы: hard_hat, no_hard_hat, safety_vest, no_vest,
            safety_glasses, no_glasses, gloves, no_gloves
    Это эффективнее, чем «нет человека — нет каски».
    """
    def __init__(self, model_path: str, site_config: dict):
        self.model = YOLO(model_path)
        self.required_ppe = site_config.get('required_ppe', ['hard_hat', 'safety_vest'])
        self.violation_threshold = site_config.get('violation_threshold', 0.5)

        # Для подавления дублирующих тревог
        self.active_violations: dict[int, dict] = {}
        self.cooldown_frames = 30  # 1 сек @ 30fps

    def detect(self, frame: np.ndarray) -> dict:
        results = self.model.track(frame, persist=True, conf=0.4)

        workers_status = {}
        all_detections = []

        for box in results[0].boxes:
            cls = self.model.names[int(box.cls)]
            conf = float(box.conf)
            bbox = list(map(int, box.xyxy[0]))
            track_id = int(box.id) if box.id is not None else -1

            all_detections.append({
                'class': cls, 'conf': conf,
                'bbox': bbox, 'track_id': track_id
            })

        # Группируем по рабочим (person = anchor)
        persons = [d for d in all_detections if d['class'] == 'person']

        for person in persons:
            pid = person['track_id']
            violations = []

            for req_ppe in self.required_ppe:
                no_ppe_class = f'no_{req_ppe}'

                # Есть явный класс "без СИЗ" рядом с рабочим?
                for det in all_detections:
                    if det['class'] == no_ppe_class:
                        if self._near_person(det['bbox'], person['bbox']):
                            if det['conf'] > self.violation_threshold:
                                violations.append({
                                    'type': no_ppe_class,
                                    'confidence': det['conf']
                                })

            workers_status[pid] = {
                'bbox': person['bbox'],
                'violations': violations,
                'compliant': len(violations) == 0
            }

        return {
            'workers': workers_status,
            'total_workers': len(persons),
            'violations_count': sum(
                len(w['violations']) for w in workers_status.values()
            ),
            'compliance_rate': (
                sum(1 for w in workers_status.values() if w['compliant'])
                / max(len(persons), 1)
            )
        }

    def _near_person(self, ppe_bbox: list, person_bbox: list,
                      expand: float = 0.3) -> bool:
        """СИЗ считается относящимся к рабочему, если его bbox близко"""
        px1, py1, px2, py2 = person_bbox
        pw = px2 - px1
        ph = py2 - py1

        # Расширяем bbox рабочего
        ex1 = px1 - pw * expand
        ey1 = py1 - ph * expand
        ex2 = px2 + pw * expand
        ey2 = py2 + ph * expand

        cx = (ppe_bbox[0] + ppe_bbox[2]) / 2
        cy = (ppe_bbox[1] + ppe_bbox[3]) / 2

        return ex1 <= cx <= ex2 and ey1 <= cy <= ey2

Трудные случаи и как с ними работать

1. Частичное перекрытие: рабочий виден наполовину за конструкцией. Голова в кадре — каска проверяется. Если голова не видна — не штрафуем.

def is_head_visible(self, person_bbox: list,
                     frame_height: int) -> bool:
    """Оцениваем, видна ли голова рабочего"""
    h = person_bbox[3] - person_bbox[1]
    # Голова занимает верхние ~15% тела
    head_region_y = person_bbox[1] + h * 0.15
    return head_region_y < frame_height * 0.95  # не у нижнего края

2. Мелкие объекты на дальнем плане: рабочий на 40 метрах, bbox 30×90 px. Каска 8×8 px. YOLOv8l на таких разрешениях даёт recall ~70%. Решение: PTZ-камеры с автозумом или дополнительные камеры на дальних секциях.

3. Похожие объекты: строительный мусор, ткань на строительных лесах похожи на каску при плохом освещении. Hard negative mining при дообучении — собираем 200–300 таких примеров и добавляем в тренировочный набор.

Статистика и дашборд

class PPEComplianceDashboard:
    def daily_summary(self, detection_log: list) -> dict:
        total_detections = len(detection_log)
        violations_by_type = defaultdict(int)
        violations_by_hour = defaultdict(int)
        violators = set()

        for record in detection_log:
            for violation in record.get('violations', []):
                violations_by_type[violation['type']] += 1
                hour = record['timestamp'].hour
                violations_by_hour[hour] += 1
                violators.add(record['worker_id'])

        return {
            'total_inspections': total_detections,
            'unique_violators': len(violators),
            'violations_by_type': dict(violations_by_type),
            'peak_violation_hour': max(violations_by_hour,
                                        key=violations_by_hour.get,
                                        default=None),
            'compliance_rate': 1 - (len(violators) / max(total_detections, 1))
        }