AI Construction Site Safety Violation Detection System

AI-based system for detecting occupational safety violations at construction sites

Construction is one of the most hazardous industries. Most accidents are related to safety violations: lack of personal protective equipment, exposure to hazardous areas, and working at heights without safety equipment. A video analytics system bridges the gap between periodic inspector checks (weekly) and continuous monitoring.

Classes of violations and detection methods

Implementation of the PPE detector

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)

Case: Construction of a residential complex, 200 workers

12 IP cameras at the construction site. Before implementation, an inspector walked the site once a day, manually recording violations. PPE violations reached 30–40 per day, some going unnoticed.

After the system is implemented:

• Coverage: 100% of camera visibility zones in real time
• Detected in the first week: 847 violations (vs 40–50 manually)
• After a month of operation: a 73% reduction in violations
• 2 critical incidents prevented (being in the crane operating area)

Accuracy on the test set: 91% for helmets, 87% for vests (difficult case: the vest is on but unbuttoned).

Notifications and Integration

• Notification to security via Telegram bot with a photo of the violation
• Automatic creation of a violation report with a frame, time, and zone
• Export violation statistics to Excel/Power BI for the safety manager