Разработка AI-системы детекции нарушений техники безопасности на стройплощадке

AI-система обнаружения нарушений охраны труда на стройплощадке

Строительство — одна из наиболее травмоопасных отраслей. Большинство несчастных случаев связаны с нарушением правил безопасности: отсутствие СИЗ, нахождение в опасной зоне, работа без страховки на высоте. Система видеоаналитики закрывает разрыв между периодическими проверками инспектора (раз в неделю) и непрерывным контролем.

Классы нарушений и методы обнаружения

Реализация детектора СИЗ

import cv2
import numpy as np
from ultralytics import YOLO
from dataclasses import dataclass

@dataclass
class SafetyViolation:
    violation_type: str
    worker_id: int
    bbox: list
    confidence: float
    zone: str
    severity: str  # 'warning', 'critical'

class ConstructionSafetyMonitor:
    def __init__(self, model_path: str, config: dict):
        # YOLOv8l дообученный на Safety Helmet Dataset + custom PPE data
        # Классы: person, hard_hat, safety_vest, no_hard_hat, no_vest,
        #         safety_glasses, gloves, harness
        self.model = YOLO(model_path)

        self.danger_zones = config['danger_zones']
        self.required_ppe = config.get('required_ppe',
                                        ['hard_hat', 'safety_vest'])
        self.violation_history = {}  # worker_track_id -> violations

        # Дополнительный pose estimator для проверки страховки на высоте
        self.pose_estimator = YOLO('yolov8l-pose.pt')

    def _worker_has_ppe(self, worker_bbox: list,
                         ppe_detections: list,
                         ppe_class: str) -> tuple[bool, float]:
        """Проверяем, есть ли у конкретного рабочего нужный СИЗ"""
        wx1, wy1, wx2, wy2 = worker_bbox
        worker_upper_half = [wx1, wy1, wx2, wy1 + (wy2 - wy1) * 0.6]

        best_iou = 0.0
        for ppe in ppe_detections:
            if ppe['class'] == ppe_class:
                iou = self._iou(worker_upper_half, ppe['bbox'])
                best_iou = max(best_iou, iou)

        # IoU > 0.1 = СИЗ находится в области тела рабочего
        return best_iou > 0.1, best_iou

    def detect_violations(self, frame: np.ndarray) -> list[SafetyViolation]:
        results = self.model.track(frame, persist=True, conf=0.4)
        violations = []

        persons = []
        ppe_items = []

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

            if cls == 'person':
                persons.append({'bbox': bbox, 'track_id': track_id})
            elif cls in ['hard_hat', 'safety_vest', 'safety_glasses',
                          'gloves', 'harness']:
                ppe_items.append({'class': cls, 'bbox': bbox, 'conf': conf})

        # Для каждого рабочего проверяем наличие СИЗ
        for worker in persons:
            zone = self._get_zone(worker['bbox'])

            for required in self.required_ppe:
                has_ppe, iou_score = self._worker_has_ppe(
                    worker['bbox'], ppe_items, required
                )

                if not has_ppe:
                    vtype = f'no_{required}'
                    violations.append(SafetyViolation(
                        violation_type=vtype,
                        worker_id=worker['track_id'],
                        bbox=worker['bbox'],
                        confidence=1.0 - iou_score,
                        zone=zone,
                        severity='critical' if required == 'hard_hat' else 'warning'
                    ))

            # Проверка нахождения в запретной зоне
            if zone in self.danger_zones:
                cx = (worker['bbox'][0] + worker['bbox'][2]) // 2
                cy = (worker['bbox'][1] + worker['bbox'][3]) // 2
                if self._in_polygon(cx, cy,
                                     self.danger_zones[zone]['polygon']):
                    violations.append(SafetyViolation(
                        violation_type='unauthorized_zone_entry',
                        worker_id=worker['track_id'],
                        bbox=worker['bbox'],
                        confidence=0.95,
                        zone=zone,
                        severity='critical'
                    ))

        return violations

    def _iou(self, box1: list, box2: list) -> float:
        x1 = max(box1[0], box2[0])
        y1 = max(box1[1], box2[1])
        x2 = min(box1[2], box2[2])
        y2 = min(box1[3], box2[3])

        inter = max(0, x2-x1) * max(0, y2-y1)
        area1 = (box1[2]-box1[0]) * (box1[3]-box1[1])
        area2 = (box2[2]-box2[0]) * (box2[3]-box2[1])
        union = area1 + area2 - inter
        return inter / max(union, 1e-6)

Кейс: строительство жилого комплекса, 200 рабочих

12 IP-камер на строительной площадке. До внедрения: инспектор обходил площадку раз в день, фиксировал нарушения вручную. Нарушения СИЗ доходили до 30–40 в день, часть оставалась незамеченной.

После внедрения системы:

• Охват: 100% зон видимости камер в режиме реального времени
• Обнаружено в первую неделю: 847 нарушений (vs 40–50 вручную)
• После месяца эксплуатации: снижение нарушений на 73%
• 2 критических инцидента предотвращены (нахождение в зоне работы крана)

Точность на тестовом наборе: 91% для касок, 87% для жилетов (сложный случай — жилет надет, но расстёгнут).

Уведомления и интеграция

• Оповещение охраны по Telegram-боту с фото нарушения
• Автоматическое создание акта нарушения с кадром, временем, зоной
• Экспорт статистики нарушений в Excel/Power BI для safety-менеджера