Розробка 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-менеджера