safety-compliance-checker

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