safety-compliance-checker

name	safety-compliance-checker
description	Automated safety compliance verification for construction sites. Check PPE usage, zone access, working at heights regulations, and generate compliance reports using rule-based and ML approaches.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

Safety Compliance Checker

Overview

This skill implements automated safety compliance checking for construction projects. Verify regulatory requirements, track safety metrics, and identify potential violations before they become incidents.

Compliance Areas:

• Personal Protective Equipment (PPE)
• Working at heights regulations
• Confined space entry
• Hot work permits
• Excavation safety
• Electrical safety
• Fire prevention

Quick Start

from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime, date

class ComplianceStatus(Enum):
    COMPLIANT = "compliant"
    NON_COMPLIANT = "non_compliant"
    PARTIAL = "partial"
    NOT_APPLICABLE = "not_applicable"
    PENDING_REVIEW = "pending_review"

@dataclass
class ComplianceCheck:
    rule_id: str
    rule_name: str
    status: ComplianceStatus
    findings: List[str]
    evidence: Optional[str]
    checked_at: datetime
    checked_by: str

# Quick compliance check
def check_ppe_compliance(workers: List[Dict]) -> List[ComplianceCheck]:
    """Check PPE compliance for workers"""
    checks = []

    for worker in workers:
        findings = []
        required_ppe = worker.get('required_ppe', ['helmet', 'vest', 'boots'])
        actual_ppe = worker.get('actual_ppe', [])

        missing = set(required_ppe) - set(actual_ppe)
        if missing:
            findings.append(f"Missing PPE: {', '.join(missing)}")

        status = ComplianceStatus.COMPLIANT if not missing else ComplianceStatus.NON_COMPLIANT

        checks.append(ComplianceCheck(
            rule_id="PPE-001",
            rule_name="Personal Protective Equipment",
            status=status,
            findings=findings,
            evidence=f"Worker ID: {worker.get('id')}",
            checked_at=datetime.now(),
            checked_by="automated_system"
        ))

    return checks

# Example usage
workers = [
    {'id': 'W001', 'required_ppe': ['helmet', 'vest', 'boots'], 'actual_ppe': ['helmet', 'vest']},
    {'id': 'W002', 'required_ppe': ['helmet', 'vest', 'boots'], 'actual_ppe': ['helmet', 'vest', 'boots']}
]

results = check_ppe_compliance(workers)
for r in results:
    print(f"{r.evidence}: {r.status.value} - {r.findings}")

Comprehensive Safety Compliance System

Safety Rules Engine

from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional, Any
from enum import Enum
from datetime import datetime, date, timedelta
import json

class RiskLevel(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

class RuleCategory(Enum):
    PPE = "Personal Protective Equipment"
    FALL_PROTECTION = "Fall Protection"
    ELECTRICAL = "Electrical Safety"
    EXCAVATION = "Excavation Safety"
    CONFINED_SPACE = "Confined Space"
    HOT_WORK = "Hot Work"
    FIRE = "Fire Prevention"
    HAZMAT = "Hazardous Materials"
    CRANE = "Crane & Lifting"
    SCAFFOLDING = "Scaffolding"

@dataclass
class SafetyRule:
    rule_id: str
    name: str
    description: str
    category: RuleCategory
    risk_level: RiskLevel
    regulation_ref: str  # OSHA, local codes
    check_function: Optional[Callable] = None
    parameters: Dict = field(default_factory=dict)


 :
    rule: SafetyRule
    location: 
    description: 
    detected_at: datetime
    severity: RiskLevel
    corrective_action: 
    deadline: date
    status:  = 
    assigned_to: [] = 

 :
    

     ():
        .rules: [, SafetyRule] = {}
        .violations: [Violation] = []
        ._load_default_rules()

     ():
        
        default_rules = [
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.PPE,
                risk_level=RiskLevel.HIGH,
                regulation_ref=,
                parameters={: []}
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.FALL_PROTECTION,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref=,
                parameters={: }
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.EXCAVATION,
                risk_level=RiskLevel.HIGH,
                regulation_ref=,
                parameters={: }
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.CONFINED_SPACE,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref=,
                parameters={: }
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.HOT_WORK,
                risk_level=RiskLevel.HIGH,
                regulation_ref=,
                parameters={: }
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.SCAFFOLDING,
                risk_level=RiskLevel.HIGH,
                regulation_ref=,
                parameters={: }
            ),
            SafetyRule(
                rule_id=,
                name=,
                description=,
                category=RuleCategory.ELECTRICAL,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref=,
                parameters={: }
            )
        ]

         rule  default_rules:
            .rules[rule.rule_id] = rule

     ():
        
        .rules[rule.rule_id] = rule

     () -> [ComplianceCheck]:
        
        checks = []

        activity_type = activity.get(, )
        location = activity.get(, )
        height = activity.get(, )
        workers = activity.get(, [])
        permits = activity.get(, [])
        equipment = activity.get(, [])

        
         height >= :
            fall_rule = .rules.get()
             fall_rule:
                has_protection = (
                      w.get(, [])    equipment
                     w  workers
                )
                checks.append(ComplianceCheck(
                    rule_id=fall_rule.rule_id,
                    rule_name=fall_rule.name,
                    status=ComplianceStatus.COMPLIANT  has_protection  ComplianceStatus.NON_COMPLIANT,
                    findings=[]  has_protection  [],
                    evidence=,
                    checked_at=datetime.now(),
                    checked_by=
                ))

        
         worker  workers:
            ppe_rule = .rules.get()
             ppe_rule:
                has_helmet =   worker.get(, [])
                checks.append(ComplianceCheck(
                    rule_id=ppe_rule.rule_id,
                    rule_name=ppe_rule.name,
                    status=ComplianceStatus.COMPLIANT  has_helmet  ComplianceStatus.NON_COMPLIANT,
                    findings=[]  has_helmet  [],
                    evidence=,
                    checked_at=datetime.now(),
                    checked_by=
                ))

        
         activity_type  [, , ]:
            hot_work_rule = .rules.get()
             hot_work_rule:
                has_permit =   permits
                has_fire_watch = activity.get(, )
                has_extinguisher =   equipment

                findings = []
                  has_permit:
                    findings.append()
                  has_fire_watch:
                    findings.append()
                  has_extinguisher:
                    findings.append()

                checks.append(ComplianceCheck(
                    rule_id=hot_work_rule.rule_id,
                    rule_name=hot_work_rule.name,
                    status=ComplianceStatus.COMPLIANT   findings  ComplianceStatus.NON_COMPLIANT,
                    findings=findings,
                    evidence=,
                    checked_at=datetime.now(),
                    checked_by=
                ))

        
         activity.get(, ):
            cs_rule = .rules.get()
             cs_rule:
                has_permit =   permits
                has_attendant = activity.get(, )
                atmospheric_tested = activity.get(, )

                findings = []
                  has_permit:
                    findings.append()
                  has_attendant:
                    findings.append()
                  atmospheric_tested:
                    findings.append()

                checks.append(ComplianceCheck(
                    rule_id=cs_rule.rule_id,
                    rule_name=cs_rule.name,
                    status=ComplianceStatus.COMPLIANT   findings  ComplianceStatus.NON_COMPLIANT,
                    findings=findings,
                    evidence=,
                    checked_at=datetime.now(),
                    checked_by=
                ))

         checks

     () -> Violation:
        
        rule = .rules.get(check.rule_id)

        
         rule.risk_level == RiskLevel.CRITICAL:
            deadline = date.today()  
         rule.risk_level == RiskLevel.HIGH:
            deadline = date.today() + timedelta(days=)
         rule.risk_level == RiskLevel.MEDIUM:
            deadline = date.today() + timedelta(days=)
        :
            deadline = date.today() + timedelta(days=)

        violation = Violation(
            rule=rule,
            location=location,
            description=.join(check.findings),
            detected_at=check.checked_at,
            severity=rule.risk_level,
            corrective_action=._get_corrective_action(rule, check.findings),
            deadline=deadline
        )

        .violations.append(violation)
         violation

     () -> :
        
        actions = {
            RuleCategory.PPE: ,
            RuleCategory.FALL_PROTECTION: ,
            RuleCategory.ELECTRICAL: ,
            RuleCategory.EXCAVATION: ,
            RuleCategory.CONFINED_SPACE: ,
            RuleCategory.HOT_WORK: ,
            RuleCategory.SCAFFOLDING: ,
            RuleCategory.CRANE: 
        }

         actions.get(rule.category, )

PPE Detection Integration

class PPEComplianceChecker:
    """PPE compliance checking with CV integration support"""

    def __init__(self):
        self.ppe_requirements = self._load_requirements()

    def _load_requirements(self) -> Dict[str, List[str]]:
        """Load PPE requirements by work type"""
        return {
            'general': ['helmet', 'safety_vest', 'safety_boots'],
            'welding': ['helmet', 'welding_mask', 'gloves', 'apron', 'safety_boots'],
            'excavation': ['helmet', 'safety_vest', 'safety_boots', 'gloves'],
            'electrical': ['helmet', 'safety_glasses', 'insulated_gloves', 'safety_boots'],
            'concrete': ['helmet', 'safety_glasses', 'gloves', 'safety_boots', 'knee_pads'],
            'demolition': ['helmet', 'safety_glasses', 'dust_mask', 'gloves', 'safety_boots'],
            : [, , , ],
            : [, , , , ],
            : [, , , ]
        }

     () -> :
        
        required = .ppe_requirements.get(work_type, .ppe_requirements[])

        missing = (required) - (detected_ppe)
        extra = (detected_ppe) - (required)

        compliance_score = ((required) - (missing)) / (required) * 

         {
            : worker_id,
            : work_type,
            : required,
            : detected_ppe,
            : (missing),
            : compliance_score,
            : (missing) == ,
            : [  item  missing]
        }

     () -> []:
        
        results = []

         detection  detections:
            worker_id = detection.get(, )
            work_type = detection.get(, )
            detected_items = detection.get(, [])

            result = .check_worker_ppe(worker_id, work_type, detected_items)
            result[] = detection.get()
            result[] = detection.get()
            result[] = detection.get()

            results.append(result)

         results

Site Inspection System

from datetime import datetime, date
from typing import List, Dict, Optional
import pandas as pd

@dataclass
class InspectionItem:
    item_id: str
    category: str
    description: str
    is_compliant: bool
    notes: str
    photo_refs: List[str] = field(default_factory=list)
    corrective_action: Optional[str] = None

@dataclass
class SiteInspection:
    inspection_id: str
    site_id: str
    inspector: str
    inspection_date: date
    weather: str
    items: List[InspectionItem]
    overall_score: float
    recommendations: List[str]

class SiteInspectionSystem:
    """Construction site safety inspection management"""

    def __init__(self):
        self.checklists = self._load_checklists()
        self.inspections: List[SiteInspection] = []

    def _load_checklists(self) -> Dict[str, []]:
        
         {
            : [
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : }
            ],
            : [
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : }
            ],
            : [
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : }
            ],
            : [
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : }
            ],
            : [
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : },
                {: , : , : }
            ]
        }

     () -> SiteInspection:
        
        items = []

         checklist_type  checklist_types:
            checklist = .checklists.get(checklist_type, [])

             check  checklist:
                item_id = check[]
                response = responses.get(item_id, {})

                items.append(InspectionItem(
                    item_id=item_id,
                    category=check[],
                    description=check[],
                    is_compliant=response.get(, ),
                    notes=response.get(, ),
                    photo_refs=response.get(, []),
                    corrective_action=response.get()   response.get(, )  
                ))

        
        compliant_count = (  item  items  item.is_compliant)
        overall_score = (compliant_count / (items) * )  items  

        
        recommendations = []
         item  items:
              item.is_compliant:
                recommendations.append()

        inspection = SiteInspection(
            inspection_id=,
            site_id=site_id,
            inspector=inspector,
            inspection_date=date.today(),
            weather=,  
            items=items,
            overall_score=overall_score,
            recommendations=recommendations[:]  
        )

        .inspections.append(inspection)
         inspection

     () -> pd.DataFrame:
        
        data = []

         item  inspection.items:
            data.append({
                : inspection.inspection_id,
                : inspection.inspection_date,
                : inspection.inspector,
                : item.item_id,
                : item.category,
                : item.description,
                :   item.is_compliant  ,
                : item.notes,
                : item.corrective_action  
            })

         pd.DataFrame(data)

     () -> :
        
        site_inspections = [
            i  i  .inspections
             i.site_id == site_id 
            (date.today() - i.inspection_date).days <= days
        ]

          site_inspections:
             {: }

        scores = [i.overall_score  i  site_inspections]
        dates = [i.inspection_date  i  site_inspections]

        
        category_compliance = {}
         inspection  site_inspections:
             item  inspection.items:
                 item.category   category_compliance:
                    category_compliance[item.category] = {: , : }
                category_compliance[item.category][] += 
                 item.is_compliant:
                    category_compliance[item.category][] += 

         {
            : site_id,
            : (site_inspections),
            : (scores) / (scores),
            : ((dates, scores)),
            : {
                k: v[] / v[] * 
                 k, v  category_compliance.items()
            },
            : site_inspections[-].inspection_id
        }

Compliance Dashboard Data

class SafetyDashboard:
    """Generate safety compliance dashboard data"""

    def __init__(self, engine: SafetyComplianceEngine,
                 inspection_system: SiteInspectionSystem):
        self.engine = engine
        self.inspections = inspection_system

    def get_dashboard_data(self, site_id: str) -> Dict:
        """Get comprehensive dashboard data"""
        # Active violations
        open_violations = [v for v in self.engine.violations if v.status == 'open']

        # Violations by severity
        by_severity = {}
        for v in open_violations:
            sev = v.severity.name
            by_severity[sev] = by_severity.get(sev, 0) + 1

        # Violations by category
        by_category = {}
        for v in open_violations:
            cat = v.rule.category.value
            by_category[cat] = by_category.get(cat, 0) + 1

        # Overdue violations
        overdue = [v for v in open_violations if v.deadline < date.today()]

        # Compliance trends
        trends = self.inspections.get_compliance_trends(site_id)

        return {
            'summary': {
                'total_open_violations': len(open_violations),
                : by_severity.get(, ),
                : (overdue),
                : trends.get(, )
            },
            : by_severity,
            : by_category,
            : [
                {
                    : v.rule.name,
                    : v.location,
                    : v.deadline.isoformat(),
                    : (date.today() - v.deadline).days
                }
                 v  overdue
            ],
            : trends.get(, []),
            : trends.get(, {})
        }

Quick Reference

Check Type	OSHA Reference	Risk Level	Frequency
PPE Compliance	1926.100-106	High	Continuous
Fall Protection	1926.501-503	Critical	Daily
Scaffolding	1926.451-454	High	Before each use
Excavation	1926.651-652	High	Daily
Electrical	1926.400-449	Critical	Daily
Confined Space	1926.1200	Critical	Before entry
Hot Work	1926.350-354	High	Per activity

Resources

• OSHA Construction Standards: https://www.osha.gov/laws-regs/regulations/standardnumber/1926
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See progress-monitoring-cv for PPE detection with computer vision
• See risk-assessment-ml for predictive safety analytics
• See document-classification-nlp for safety document processing