Safety Compliance Checker Agent Skill

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

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

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

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 = []

        for item in inspection.items:
            data.append({
                'Inspection ID': inspection.inspection_id,
                'Date': inspection.inspection_date,
                'Inspector': inspection.inspector,
                'Item ID': item.item_id,
                'Category': item.category,
                'Description': item.description,
                'Compliant': 'Yes' if item.is_compliant else 'No',
                'Notes': item.notes,
                'Corrective Action': item.corrective_action or ''
            })

        return pd.DataFrame(data)

    def get_compliance_trends(self, site_id: str, days: int = 30) -> Dict:
        """Get compliance trends for a site"""
        site_inspections = [
            i for i in self.inspections
            if i.site_id == site_id and
            (date.today() - i.inspection_date).days <= days
        ]

        if not site_inspections:
            return {'message': 'No inspections found'}

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

        # Category breakdown
        category_compliance = {}
        for inspection in site_inspections:
            for item in inspection.items:
                if item.category not in category_compliance:
                    category_compliance[item.category] = {'compliant': 0, 'total': 0}
                category_compliance[item.category]['total'] += 1
                if item.is_compliant:
                    category_compliance[item.category]['compliant'] += 1

        return {
            'site_id': site_id,
            'inspection_count': len(site_inspections),
            'average_score': sum(scores) / len(scores),
            'score_trend': list(zip(dates, scores)),
            'category_compliance': {
                k: v['compliant'] / v['total'] * 100
                for k, v in category_compliance.items()
            },
            'latest_inspection': site_inspections[-1].inspection_id
        }

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),
                'critical_violations': by_severity.get('CRITICAL', 0),
                'overdue_violations': len(overdue),
                'latest_inspection_score': trends.get('average_score', 0)
            },
            'violations_by_severity': by_severity,
            'violations_by_category': by_category,
            'overdue_items': [
                {
                    'rule': v.rule.name,
                    'location': v.location,
                    'deadline': v.deadline.isoformat(),
                    'days_overdue': (date.today() - v.deadline).days
                }
                for v in overdue
            ],
            'compliance_trend': trends.get('score_trend', []),
            'category_compliance': trends.get('category_compliance', {})
        }