AI Construction Worker PPE Absence Detection System

AI detection of lack of personal protective equipment (PPE)

A narrow but critical task: automatically detecting whether a worker is wearing a hard hat, vest, goggles, or gloves. It sounds simple until you encounter real-life conditions: the worker is in the shade, sideways, at a height of 20 meters, surrounded by clouds of dust.

Datasets and models for PPE detection

Open datasets: Safety Helmet Detection Dataset (7000+ images), PPE-Detection Dataset (Roboflow, 8000+ images), CHV (Construction Helmet and Vest).

For production, additional training on your own data is always necessary. Different construction sites require different types of hard hats, vests, and lighting.

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

Difficult cases and how to handle them

1. Partial obstruction: the worker is half visible behind the structure. The head is in the frame—the helmet is being checked. If the head is not visible, there is no penalty.

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. Small objects in the background: worker at 40 meters, bbox 30×90 px. Hard hat 8×8 px. YOLOv8l provides ~70% recall at these resolutions. Solution: PTZ cameras with autozoom or additional cameras in the far sections.

3. Similar objects: construction debris and fabric on scaffolding resemble a hard hat in low light. Hard negative mining during retraining—we collect 200–300 such examples and add them to the training set.

Statistics and dashboard

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))
        }