safety-compliance-checker

What this skill does

# 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

```python
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

```python
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