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)

@dataclass
class Violation:
    rule: SafetyRule
    location: str
    description: str
    detected_at: datetime
    severity: RiskLevel
    corrective_action: str
    deadline: date
    status: str = "open"
    assigned_to: Optional[str] = None

class SafetyComplianceEngine:
    """Rule-based safety compliance checking engine"""

    def __init__(self):
        self.rules: Dict[str, SafetyRule] = {}
        self.violations: List[Violation] = []
        self._load_default_rules()

    def _load_default_rules(self):
        """Load default safety rules"""
        default_rules = [
            SafetyRule(
                rule_id="OSHA-1926.100",
                name="Head Protection",
                description="Employees working in areas where there is a possible danger of head injury shall wear protective helmets",
                category=RuleCategory.PPE,
                risk_level=RiskLevel.HIGH,
                regulation_ref="OSHA 29 CFR 1926.100",
                parameters={"required_ppe": ["helmet"]}
            ),
            SafetyRule(
                rule_id="OSHA-1926.501",
                name="Fall Protection - General",
                description="Each employee on walking/working surfaces with unprotected sides 6 feet or more above lower level shall be protected",
                category=RuleCategory.FALL_PROTECTION,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref="OSHA 29 CFR 1926.501",
                parameters={"height_threshold_ft": 6}
            ),
            SafetyRule(
                rule_id="OSHA-1926.651",
                name="Excavation General Requirements",
                description="Daily inspections of excavations, adjacent areas, and protective systems",
                category=RuleCategory.EXCAVATION,
                risk_level=RiskLevel.HIGH,
                regulation_ref="OSHA 29 CFR 1926.651",
                parameters={"inspection_frequency": "daily"}
            ),
            SafetyRule(
                rule_id="OSHA-1926.1200",
                name="Confined Space Entry",
                description="Permit-required confined space program for construction",
                category=RuleCategory.CONFINED_SPACE,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref="OSHA 29 CFR 1926.1200",
                parameters={"permit_required": True}
            ),
            SafetyRule(
                rule_id="OSHA-1926.352",
                name="Fire Prevention - Welding",
                description="Fire watch and extinguisher required for hot work",
                category=RuleCategory.HOT_WORK,
                risk_level=RiskLevel.HIGH,
                regulation_ref="OSHA 29 CFR 1926.352",
                parameters={"fire_watch_duration_min": 30}
            ),
            SafetyRule(
                rule_id="OSHA-1926.451",
                name="Scaffolding General Requirements",
                description="Scaffold platforms shall be fully planked between front uprights and guardrail supports",
                category=RuleCategory.SCAFFOLDING,
                risk_level=RiskLevel.HIGH,
                regulation_ref="OSHA 29 CFR 1926.451",
                parameters={"max_height_without_tie": 26}
            ),
            SafetyRule(
                rule_id="OSHA-1926.405",
                name="Electrical Wiring Methods",
                description="All electrical equipment and circuits shall be grounded",
                category=RuleCategory.ELECTRICAL,
                risk_level=RiskLevel.CRITICAL,
                regulation_ref="OSHA 29 CFR 1926.405",
                parameters={"gfci_required": True}
            )
        ]

        for rule in default_rules:
            self.rules[rule.rule_id] = rule

    def add_rule(self, rule: SafetyRule):
        """Add custom safety rule"""
        self.rules[rule.rule_id] = rule

    def check_work_activity(self, activity: Dict) -> List[ComplianceCheck]:
        """Check compliance for a work activity"""
        checks = []

        activity_type = activity.get('type', '')
        location = activity.get('location', '')
        height = activity.get('height_ft', 0)
        workers = activity.get('workers', [])
        permits = activity.get('permits', [])
        equipment = activity.get('equipment', [])

        # Fall protection check
        if height >= 6:
            fall_rule = self.rules.get("OSHA-1926.501")
            if fall_rule:
                has_protection = any(
                    'harness' in w.get('ppe', []) or 'guardrail' in equipment
                    for w in workers
                )
                checks.append(ComplianceCheck(
                    rule_id=fall_rule.rule_id,
                    rule_name=fall_rule.name,
                    status=ComplianceStatus.COMPLIANT if has_protection else ComplianceStatus.NON_COMPLIANT,
                    findings=[] if has_protection else [f"Working at {height}ft without fall protection"],
                    evidence=f"Activity: {activity_type} at {location}",
                    checked_at=datetime.now(),
                    checked_by="automated_system"
                ))

        # PPE check for all workers
        for worker in workers:
            ppe_rule = self.rules.get("OSHA-1926.100")
            if ppe_rule:
                has_helmet = 'helmet' in worker.get('ppe', [])
                checks.append(ComplianceCheck(
                    rule_id=ppe_rule.rule_id,
                    rule_name=ppe_rule.name,
                    status=ComplianceStatus.COMPLIANT if has_helmet else ComplianceStatus.NON_COMPLIANT,
                    findings=[] if has_helmet else ["Missing hard hat"],
                    evidence=f"Worker: {worker.get('id')}",
                    checked_at=datetime.now(),
                    checked_by="automated_system"
                ))

        # Hot work permit check
        if activity_type in ['welding', 'cutting', 'brazing']:
            hot_work_rule = self.rules.get("OSHA-1926.352")
            if hot_work_rule:
                has_permit = 'hot_work' in permits
                has_fire_watch = activity.get('fire_watch', False)
                has_extinguisher = 'fire_extinguisher' in equipment

                findings = []
                if not has_permit:
                    findings.append("Missing hot work permit")
                if not has_fire_watch:
                    findings.append("No fire watch assigned")
                if not has_extinguisher:
                    findings.append("No fire extinguisher on site")

                checks.append(ComplianceCheck(
                    rule_id=hot_work_rule.rule_id,
                    rule_name=hot_work_rule.name,
                    status=ComplianceStatus.COMPLIANT if not findings else ComplianceStatus.NON_COMPLIANT,
                    findings=findings,
                    evidence=f"Activity: {activity_type} at {location}",
                    checked_at=datetime.now(),
                    checked_by="automated_system"
                ))

        # Confined space check
        if activity.get('confined_space', False):
            cs_rule = self.rules.get("OSHA-1926.1200")
            if cs_rule:
                has_permit = 'confined_space' in permits
                has_attendant = activity.get('attendant', False)
                atmospheric_tested = activity.get('atmospheric_test', False)

                findings = []
                if not has_permit:
                    findings.append("Missing confined space entry permit")
                if not has_attendant:
                    findings.append("No attendant stationed at entry")
                if not atmospheric_tested:
                    findings.append("Atmospheric testing not performed")

                checks.append(ComplianceCheck(
                    rule_id=cs_rule.rule_id,
                    rule_name=cs_rule.name,
                    status=ComplianceStatus.COMPLIANT if not findings else ComplianceStatus.NON_COMPLIANT,
                    findings=findings,
                    evidence=f"Confined space: {location}",
                    checked_at=datetime.now(),
                    checked_by="automated_system"
                ))

        return checks

    def create_violation(self, check: ComplianceCheck, location: str) -> Violation:
        """Create violation record from failed check"""
        rule = self.rules.get(check.rule_id)

        # Determine deadline based on severity
        if rule.risk_level == RiskLevel.CRITICAL:
            deadline = date.today()  # Immediate
        elif rule.risk_level == RiskLevel.HIGH:
            deadline = date.today() + timedelta(days=1)
        elif rule.risk_level == RiskLevel.MEDIUM:
            deadline = date.today() + timedelta(days=3)
        else:
            deadline = date.today() + timedelta(days=7)

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

        self.violations.append(violation)
        return violation

    def _get_corrective_action(self, rule: SafetyRule, findings: List[str]) -> str:
        """Generate corrective action based on rule and findings"""
        actions = {
            RuleCategory.PPE: "Provide required PPE to workers and ensure proper usage",
            RuleCategory.FALL_PROTECTION: "Install guardrails, safety nets, or provide personal fall arrest systems",
            RuleCategory.ELECTRICAL: "De-energize equipment, install GFCI protection, verify grounding",
            RuleCategory.EXCAVATION: "Perform required inspections, install shoring/sloping as needed",
            RuleCategory.CONFINED_SPACE: "Stop work, evacuate space, obtain required permits",
            RuleCategory.HOT_WORK: "Obtain hot work permit, assign fire watch, provide extinguishers",
            RuleCategory.SCAFFOLDING: "Inspect scaffold, install missing components, tag out if unsafe",
            RuleCategory.CRANE: "Verify operator certification, check rigging, establish swing radius"
        }

        return actions.get(rule.category, "Review and correct violation immediately")

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'],
            'roofing': ['helmet', 'harness', 'safety_boots', 'gloves'],
            'painting': ['helmet', 'respirator', 'safety_glasses', 'gloves', 'coveralls'],
            'scaffolding': ['helmet', 'harness', 'safety_boots', 'gloves']
        }

    def check_worker_ppe(self, worker_id: str, work_type: str,
                         detected_ppe: List[str]) -> Dict:
        """Check worker PPE against requirements"""
        required = self.ppe_requirements.get(work_type, self.ppe_requirements['general'])

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

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

        return {
            'worker_id': worker_id,
            'work_type': work_type,
            'required_ppe': required,
            'detected_ppe': detected_ppe,
            'missing_ppe': list(missing),
            'compliance_score': compliance_score,
            'is_compliant': len(missing) == 0,
            'recommendations': [f"Provide {item}" for item in missing]
        }

    def process_cv_detections(self, detections: List[Dict]) -> List[Dict]:
        """Process detections from computer vision system"""
        results = []

        for detection in detections:
            worker_id = detection.get('worker_id', 'unknown')
            work_type = detection.get('work_type', 'general')
            detected_items = detection.get('ppe_items', [])

            result = self.check_worker_ppe(worker_id, work_type, detected_items)
            result['timestamp'] = detection.get('timestamp')
            result['location'] = detection.get('location')
            result['image_ref'] = detection.get('image_path')

            results.append(result)

        return 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, List[Dict]]:
        """Load inspection checklists by category"""
        return {
            'general_safety': [
                {'id': 'GS-001', 'item': 'Site access controlled and secured', 'category': 'Access'},
                {'id': 'GS-002', 'item': 'Safety signage posted at entrance', 'category': 'Signage'},
                {'id': 'GS-003', 'item': 'First aid station accessible and stocked', 'category': 'Emergency'},
                {'id': 'GS-004', 'item': 'Emergency contact numbers posted', 'category': 'Emergency'},
                {'id': 'GS-005', 'item': 'Fire extinguishers accessible and inspected', 'category': 'Fire'},
                {'id': 'GS-006', 'item': 'Evacuation routes clear and marked', 'category': 'Emergency'},
                {'id': 'GS-007', 'item': 'Housekeeping adequate (no tripping hazards)', 'category': 'Housekeeping'},
                {'id': 'GS-008', 'item': 'Material storage organized and stable', 'category': 'Storage'},
                {'id': 'GS-009', 'item': 'Adequate lighting in work areas', 'category': 'Environment'},
                {'id': 'GS-010', 'item': 'Toilets and washing facilities clean', 'category': 'Welfare'}
            ],
            'fall_protection': [
                {'id': 'FP-001', 'item': 'Guardrails installed where required', 'category': 'Guardrails'},
                {'id': 'FP-002', 'item': 'Floor openings covered or protected', 'category': 'Openings'},
                {'id': 'FP-003', 'item': 'Ladders in good condition and secured', 'category': 'Ladders'},
                {'id': 'FP-004', 'item': 'Harnesses inspected and in good condition', 'category': 'PPE'},
                {'id': 'FP-005', 'item': 'Anchor points adequate and tested', 'category': 'Anchors'},
                {'id': 'FP-006', 'item': 'Safety nets installed where required', 'category': 'Nets'},
                {'id': 'FP-007', 'item': 'Stairways have handrails', 'category': 'Stairs'},
                {'id': 'FP-008', 'item': 'Roof edge protection in place', 'category': 'Roof'}
            ],
            'scaffolding': [
                {'id': 'SC-001', 'item': 'Scaffold erected by competent person', 'category': 'Erection'},
                {'id': 'SC-002', 'item': 'Base plates and mudsills in place', 'category': 'Foundation'},
                {'id': 'SC-003', 'item': 'All platforms fully planked', 'category': 'Platforms'},
                {'id': 'SC-004', 'item': 'Guardrails on all open sides', 'category': 'Protection'},
                {'id': 'SC-005', 'item': 'Access ladders provided', 'category': 'Access'},
                {'id': 'SC-006', 'item': 'Tied to structure at required intervals', 'category': 'Stability'},
                {'id': 'SC-007', 'item': 'Inspection tag current', 'category': 'Documentation'},
                {'id': 'SC-008', 'item': 'No damage to components', 'category': 'Condition'}
            ],
            'electrical': [
                {'id': 'EL-001', 'item': 'GFCI protection on all outlets', 'category': 'Protection'},
                {'id': 'EL-002', 'item': 'Extension cords in good condition', 'category': 'Equipment'},
                {'id': 'EL-003', 'item': 'Panel boxes closed and labeled', 'category': 'Panels'},
                {'id': 'EL-004', 'item': 'No exposed live wires', 'category': 'Hazards'},
                {'id': 'EL-005', 'item': 'Lockout/tagout procedures followed', 'category': 'LOTO'},
                {'id': 'EL-006', 'item': 'Temporary wiring properly supported', 'category': 'Installation'}
            ],
            'excavation': [
                {'id': 'EX-001', 'item': 'Competent person inspection completed', 'category': 'Inspection'},
                {'id': 'EX-002', 'item': 'Utilities located and marked', 'category': 'Utilities'},
                {'id': 'EX-003', 'item': 'Protective systems in place (shoring/sloping)', 'category': 'Protection'},
                {'id': 'EX-004', 'item': 'Spoil pile minimum 2ft from edge', 'category': 'Materials'},
                {'id': 'EX-005', 'item': 'Safe access/egress within 25ft', 'category': 'Access'},
                {'id': 'EX-006', 'item': 'Water accumulation managed', 'category': 'Water'},
                {'id': 'EX-007', 'item': 'Barricades around excavation', 'category': 'Barriers'}
            ]
        }

    def conduct_inspection(self, site_id: str, inspector: str,
                           checklist_types: List[str],
                           responses: Dict[str, Dict]) -> SiteInspection:
        """Conduct site inspection"""
        items = []

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

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

                items.append(InspectionItem(
                    item_id=item_id,
                    category=check['category'],
                    description=check['item'],
                    is_compliant=response.get('compliant', False),
                    notes=response.get('notes', ''),
                    photo_refs=response.get('photos', []),
                    corrective_action=response.get('action') if not response.get('compliant', False) else None
                ))

        # Calculate score
        compliant_count = sum(1 for item in items if item.is_compliant)
        overall_score = (compliant_count / len(items) * 100) if items else 0

        # Generate recommendations
        recommendations = []
        for item in items:
            if not item.is_compliant:
                recommendations.append(f"{item.item_id}: {item.corrective_action or 'Address non-compliance'}")

        inspection = SiteInspection(
            inspection_id=f"INS-{site_id}-{datetime.now().strftime('%Y%m%d%H%M')}",
            site_id=site_id,
            inspector=inspector,
            inspection_date=date.today(),
            weather="Clear",  # Would be input
            items=items,
            overall_score=overall_score,
            recommendations=recommendations[:10]  # Top 10
        )

        self.inspections.append(inspection)
        return inspection

    def generate_inspection_report(self, inspection: SiteInspection) -> pd.DataFrame:
        """Generate inspection report DataFrame"""
        data = []