Safety Inspection Bot

Data Structure

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

class HazardSeverity(Enum):
    CRITICAL = "Critical"      # Immediate danger to life
    HIGH = "High"              # Serious injury potential
    MEDIUM = "Medium"          # Injury potential
    LOW = "Low"                # Minor hazard
    OBSERVATION = "Observation"  # Best practice

class HazardStatus(Enum):
    OPEN = "Open"
    IN_PROGRESS = "In Progress"
    CORRECTED = "Corrected"
    VERIFIED = "Verified Closed"

class InspectionType(Enum):
    DAILY_PREWORK = "Daily Pre-Work"
    TOOLBOX_TALK = "Toolbox Talk"
    AREA_INSPECTION = "Area Inspection"
    EQUIPMENT_INSPECTION = "Equipment Inspection"
    HOT_WORK_PERMIT = "Hot Work Permit"
    CONFINED_SPACE = "Confined Space Entry"
    CRANE_LIFT = "Crane/Lift Inspection"
    SCAFFOLD = "Scaffold Inspection"
    EXCAVATION = "Excavation Inspection"
    INCIDENT = "Incident Report"

@dataclass
class Hazard:
    hazard_id: str
    inspection_id: str
    description: str
    severity: HazardSeverity
    location: str
    photo_urls: List[str] = field(default_factory=list)
    assigned_to: str = ""
    due_date: date = None
    status: HazardStatus = HazardStatus.OPEN
    corrective_action: str = ""
    corrected_date: date = None
    corrected_by: str = ""
    verification_date: date = None
    verified_by: str = ""

@dataclass
class Inspection:
    inspection_id: str
    inspection_type: InspectionType
    project_id: str
    date: date
    inspector: str
    location: str
    checklist_items: List[dict] = field(default_factory=list)
    hazards_found: List[Hazard] = field(default_factory=list)
    overall_rating: str = ""  # Pass/Fail/Conditional
    notes: str = ""
    photos: List[str] = field(default_factory=list)
    signatures: List[dict] = field(default_factory=list)
    weather: str = ""
    created_at: datetime = field(default_factory=datetime.now)

@dataclass
class Incident:
    incident_id: str
    project_id: str
    date: datetime
    type: str  # Near Miss, First Aid, Recordable, Lost Time
    description: str
    location: str
    injured_party: str = ""
    witness_names: List[str] = field(default_factory=list)
    immediate_actions: str = ""
    root_cause: str = ""
    corrective_actions: str = ""
    reported_by: str = ""
    photos: List[str] = field(default_factory=list)
    osha_recordable: bool = False
    days_away: int = 0
    days_restricted: int = 0

Python Implementation

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

class SafetyManager:
    """Construction site safety management system"""

    def __init__(self, project_id: str, storage_path: str = None):
        self.project_id = project_id
        self.storage_path = storage_path or f"safety_{project_id}"
        self.inspections: Dict[str, Inspection] = {}
        self.hazards: Dict[str, Hazard] = {}
        self.incidents: Dict[str, Incident] = {}

        # Load checklists
        self.checklists = self._load_checklists()

    def _load_checklists(self) -> Dict[str, List[dict]]:
        """Load inspection checklists"""
        return {
            InspectionType.DAILY_PREWORK.value: [
                {"id": "DP01", "item": "Work area clean and organized", "category": "Housekeeping"},
                {"id": "DP02", "item": "Walking surfaces clear of debris", "category": "Housekeeping"},
                {"id": "DP03", "item": "All workers have required PPE", "category": "PPE"},
                {"id": "DP04", "item": "Hard hats worn in designated areas", "category": "PPE"},
                {"id": "DP05", "item": "Safety glasses worn where required", "category": "PPE"},
                {"id": "DP06", "item": "High-visibility vests worn", "category": "PPE"},
                {"id": "DP07", "item": "Fall protection in use above 6 feet", "category": "Fall Protection"},
                {"id": "DP08", "item": "Guardrails/covers on floor openings", "category": "Fall Protection"},
                {"id": "DP09", "item": "Ladders in good condition", "category": "Equipment"},
                {"id": "DP10", "item": "Extension cords not damaged", "category": "Electrical"},
                {"id": "DP11", "item": "GFCIs in use for power tools", "category": "Electrical"},
                {"id": "DP12", "item": "Fire extinguishers accessible", "category": "Fire Safety"},
                {"id": "DP13", "item": "Emergency exits clear", "category": "Emergency"},
                {"id": "DP14", "item": "First aid kit stocked", "category": "Emergency"},
                {"id": "DP15", "item": "SDS sheets available", "category": "Hazcom"},
            ],
            InspectionType.SCAFFOLD.value: [
                {"id": "SC01", "item": "Base plates/mudsills in place", "category": "Foundation"},
                {"id": "SC02", "item": "All legs plumb and level", "category": "Structure"},
                {"id": "SC03", "item": "Cross bracing complete", "category": "Structure"},
                {"id": "SC04", "item": "Planking fully decked", "category": "Platform"},
                {"id": "SC05", "item": "No gaps >1 inch between planks", "category": "Platform"},
                {"id": "SC06", "item": "Guardrails at 42 inches", "category": "Guardrails"},
                {"id": "SC07", "item": "Midrails at 21 inches", "category": "Guardrails"},
                {"id": "SC08", "item": "Toeboards installed", "category": "Guardrails"},
                {"id": "SC09", "item": "Access ladder provided", "category": "Access"},
                {"id": "SC10", "item": "Tied to structure every 26 feet vertical", "category": "Ties"},
                {"id": "SC11", "item": "Inspection tag current", "category": "Documentation"},
                {"id": "SC12", "item": "Competent person inspection today", "category": "Documentation"},
            ],
            InspectionType.EXCAVATION.value: [
                {"id": "EX01", "item": "Excavation permit obtained", "category": "Permits"},
                {"id": "EX02", "item": "Utilities located and marked", "category": "Utilities"},
                {"id": "EX03", "item": "Competent person on site", "category": "Supervision"},
                {"id": "EX04", "item": "Soil classification completed", "category": "Soil"},
                {"id": "EX05", "item": "Appropriate protective system in place", "category": "Protection"},
                {"id": "EX06", "item": "Spoil pile 2+ feet from edge", "category": "Housekeeping"},
                {"id": "EX07", "item": "Ladder within 25 feet of workers", "category": "Egress"},
                {"id": "EX08", "item": "Barricades around excavation", "category": "Protection"},
                {"id": "EX09", "item": "Water accumulation addressed", "category": "Conditions"},
                {"id": "EX10", "item": "Atmosphere tested if >4 feet", "category": "Air Quality"},
            ],
        }

    def create_inspection(
        self,
        inspection_type: InspectionType,
        inspector: str,
        location: str,
        weather: str = ""
    ) -> Inspection:
        """Create new inspection"""

        inspection_id = f"INS-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        # Get checklist for this type
        checklist = self.checklists.get(inspection_type.value, [])
        checklist_items = [
            {**item, "result": None, "notes": "", "photo": None}
            for item in checklist
        ]

        inspection = Inspection(
            inspection_id=inspection_id,
            inspection_type=inspection_type,
            project_id=self.project_id,
            date=date.today(),
            inspector=inspector,
            location=location,
            checklist_items=checklist_items,
            weather=weather
        )

        self.inspections[inspection_id] = inspection
        return inspection

    def complete_checklist_item(
        self,
        inspection_id: str,
        item_id: str,
        result: str,  # "Pass", "Fail", "N/A"
        notes: str = "",
        photo_url: str = None
    ) -> Inspection:
        """Complete a checklist item"""

        inspection = self.inspections.get(inspection_id)
        if not inspection:
            raise ValueError(f"Inspection {inspection_id} not found")

        for item in inspection.checklist_items:
            if item["id"] == item_id:
                item["result"] = result
                item["notes"] = notes
                item["photo"] = photo_url
                break

        # If failed, prompt for hazard creation
        if result == "Fail":
            print(f"⚠️ Item {item_id} failed - create hazard record")

        return inspection

    def add_hazard(
        self,
        inspection_id: str,
        description: str,
        severity: HazardSeverity,
        location: str,
        assigned_to: str = "",
        due_date: date = None,
        photo_urls: List[str] = None
    ) -> Hazard:
        """Record a hazard finding"""

        hazard_id = f"HAZ-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        if due_date is None:
            # Default due dates by severity
            days = {
                HazardSeverity.CRITICAL: 0,    # Immediate
                HazardSeverity.HIGH: 1,        # 24 hours
                HazardSeverity.MEDIUM: 3,      # 3 days
                HazardSeverity.LOW: 7,         # 1 week
                HazardSeverity.OBSERVATION: 14 # 2 weeks
            }
            due_date = date.today() + timedelta(days=days.get(severity, 7))

        hazard = Hazard(
            hazard_id=hazard_id,
            inspection_id=inspection_id,
            description=description,
            severity=severity,
            location=location,
            assigned_to=assigned_to,
            due_date=due_date,
            photo_urls=photo_urls or []
        )

        self.hazards[hazard_id] = hazard

        # Add to inspection
        if inspection_id in self.inspections:
            self.inspections[inspection_id].hazards_found.append(hazard)

        # Notify for critical/high severity
        if severity in [HazardSeverity.CRITICAL, HazardSeverity.HIGH]:
            self._notify_hazard(hazard)

        return hazard

    def correct_hazard(
        self,
        hazard_id: str,
        corrective_action: str,
        corrected_by: str,
        photo_url: str = None
    ) -> Hazard:
        """Record hazard correction"""

        hazard = self.hazards.get(hazard_id)
        if not hazard:
            raise ValueError(f"Hazard {hazard_id} not found")

        hazard.corrective_action = corrective_action
        hazard.corrected_by = corrected_by
        hazard.corrected_date = date.today()
        hazard.status = HazardStatus.CORRECTED

        if photo_url:
            hazard.photo_urls.append(photo_url)

        return hazard

    def verify_hazard_closure(
        self,
        hazard_id: str,
        verified_by: str
    ) -> Hazard:
        """Verify hazard has been properly corrected"""

        hazard = self.hazards.get(hazard_id)
        if not hazard:
            raise ValueError(f"Hazard {hazard_id} not found")

        if hazard.status != HazardStatus.CORRECTED:
            raise ValueError(f"Hazard {hazard_id} not corrected yet")

        hazard.verified_by = verified_by
        hazard.verification_date = date.today()
        hazard.status = HazardStatus.VERIFIED

        return hazard

    def report_incident(
        self,
        incident_type: str,
        description: str,
        location: str,
        injured_party: str = "",
        witness_names: List[str] = None,
        immediate_actions: str = "",
        reported_by: str = "",
        photo_urls: List[str] = None
    ) -> Incident:
        """Report safety incident"""

        incident_id = f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        incident = Incident(
            incident_id=incident_id,
            project_id=self.project_id,
            date=datetime.now(),
            type=incident_type,
            description=description,
            location=location,
            injured_party=injured_party,
            witness_names=witness_names or [],
            immediate_actions=immediate_actions,
            reported_by=reported_by,
            photos=photo_urls or []
        )

        self.incidents[incident_id] = incident

        # Immediate notification for all incidents
        self._notify_incident(incident)

        return incident

    def get_open_hazards(self) -> List[Hazard]:
        """Get all open hazards"""
        return [
            h for h in self.hazards.values()
            if h.status in [HazardStatus.OPEN, HazardStatus.IN_PROGRESS]
        ]

    def get_overdue_hazards(self) -> List[Hazard]:
        """Get overdue hazards"""
        today = date.today()
        return [
            h for h in self.hazards.values()
            if h.status == HazardStatus.OPEN and h.due_date < today
        ]

    def get_statistics(self, period_days: int = 30) -> dict:
        """Get safety statistics"""

        cutoff = date.today() - timedelta(days=period_days)

        # Filter by period
        period_inspections = [
            i for i in self.inspections.values()
            if i.date >= cutoff
        ]
        period_hazards = [
            h for h in self.hazards.values()
            # Get creation date from inspection
        ]
        period_incidents = [
            i for i in self.incidents.values()
            if i.date.date() >= cutoff
        ]

        # Calculate metrics
        total_inspections = len(period_inspections)
        total_hazards = len(self.hazards)
        open_hazards = len(self.get_open_hazards())
        overdue_hazards = len(self.get_overdue_hazards())

        # Incident metrics
        near_misses = len([i for i in period_incidents if i.type == "Near Miss"])
        first_aid = len([i for i in period_incidents if i.type == "First Aid"])
        recordables = len([i for i in period_incidents if i.osha_recordable])

        # Calculate TRIR (Total Recordable Incident Rate)
        # TRIR = (Recordables × 200,000) / Total Hours Worked
        # Assuming 50 workers × 8 hours × 22 days = 8,800 hours/month
        estimated_hours = 8800 * (period_days / 30)
        trir = (recordables * 200000 / estimated_hours) if estimated_hours > 0 else 0

        return {
            'period_days': period_days,
            'inspections_completed': total_inspections,
            'hazards_identified': total_hazards,
            'hazards_open': open_hazards,
            'hazards_overdue': overdue_hazards,
            'hazards_by_severity': self._count_by_severity(),
            'incidents_total': len(period_incidents),
            'near_misses': near_misses,
            'first_aid': first_aid,
            'recordables': recordables,
            'trir': round(trir, 2),
            'days_since_last_recordable': self._days_since_recordable()
        }

    def _count_by_severity(self) -> dict:
        """Count hazards by severity"""
        result = {s.value: 0 for s in HazardSeverity}
        for hazard in self.hazards.values():
            if hazard.status != HazardStatus.VERIFIED:
                result[hazard.severity.value] += 1
        return result

    def _days_since_recordable(self) -> int:
        """Calculate days since last recordable incident"""
        recordables = [
            i for i in self.incidents.values()
            if i.osha_recordable
        ]
        if not recordables:
            return 365  # Assume 1 year if no recordables

        last = max(recordables, key=lambda x: x.date)
        return (datetime.now() - last.date).days

    def _notify_hazard(self, hazard: Hazard):
        """Send notification for high-severity hazard"""
        print(f"🚨 HAZARD ALERT: {hazard.severity.value}")
        print(f"   Location: {hazard.location}")
        print(f"   Description: {hazard.description}")
        print(f"   Assigned to: {hazard.assigned_to}")
        print(f"   Due: {hazard.due_date}")

    def _notify_incident(self, incident: Incident):
        """Send notification for incident"""
        print(f"⚠️ INCIDENT REPORTED: {incident.type}")
        print(f"   Location: {incident.location}")
        print(f"   Description: {incident.description}")
        if incident.injured_party:
            print(f"   Injured: {incident.injured_party}")

    def generate_daily_safety_report(self) -> str:
        """Generate daily safety briefing"""

        stats = self.get_statistics(period_days=1)
        open_hazards = self.get_open_hazards()
        overdue = self.get_overdue_hazards()

        report = f"""
╔══════════════════════════════════════════════════════════════╗
║                   DAILY SAFETY BRIEFING                       ║
║   Project: {self.project_id:<40}       ║
║   Date: {date.today().strftime('%d.%m.%Y'):<43}       ║
╠══════════════════════════════════════════════════════════════╣

📊 TODAY'S METRICS
───────────────────────────────────────────────────────────────
   Days Without Recordable Incident: {stats['days_since_last_recordable']}
   Open Hazards: {stats['hazards_open']}
   Overdue Hazards: {stats['hazards_overdue']}

⚠️ OPEN HAZARDS REQUIRING ATTENTION
───────────────────────────────────────────────────────────────
"""
        if overdue:
            report += "🔴 OVERDUE:\n"
            for h in overdue:
                report += f"   • {h.hazard_id}: {h.description[:50]}... (Due: {h.due_date})\n"

        critical_high = [h for h in open_hazards
                        if h.severity in [HazardSeverity.CRITICAL, HazardSeverity.HIGH]]
        if critical_high:
            report += "\n🟠 CRITICAL/HIGH PRIORITY:\n"
            for h in critical_high:
                report += f"   • {h.hazard_id}: {h.description[:50]}... ({h.severity.value})\n"

        report += """
📋 REQUIRED INSPECTIONS TODAY
───────────────────────────────────────────────────────────────
   ☐ Pre-work safety inspection
   ☐ Toolbox talk (Topic: ________________)
   ☐ Equipment inspections

💡 SAFETY FOCUS OF THE DAY
───────────────────────────────────────────────────────────────
   [Insert daily focus topic]

╚══════════════════════════════════════════════════════════════╝
"""
        return report


# Usage Example
if __name__ == "__main__":
    # Initialize safety manager
    safety = SafetyManager(project_id="PROJECT-2026-001")

    # Create daily pre-work inspection
    inspection = safety.create_inspection(
        inspection_type=InspectionType.DAILY_PREWORK,
        inspector="Ivan Petrov",
        location="Building A, Floor 5",
        weather="Clear, -5°C"
    )

    print(f"Created inspection: {inspection.inspection_id}")

    # Complete checklist items
    safety.complete_checklist_item(
        inspection_id=inspection.inspection_id,
        item_id="DP01",
        result="Pass"
    )

    safety.complete_checklist_item(
        inspection_id=inspection.inspection_id,
        item_id="DP07",
        result="Fail",
        notes="Worker on scaffold without harness"
    )

    # Add hazard for failed item
    hazard = safety.add_hazard(
        inspection_id=inspection.inspection_id,
        description="Worker observed on scaffold without fall protection harness",
        severity=HazardSeverity.CRITICAL,
        location="Building A, Floor 5, West side",
        assigned_to="Site Foreman"
    )

    print(f"Hazard created: {hazard.hazard_id}")

    # Correct hazard
    safety.correct_hazard(
        hazard_id=hazard.hazard_id,
        corrective_action="Worker provided harness and retrained on fall protection requirements",
        corrected_by="Ivan Petrov"
    )

    # Verify closure
    safety.verify_hazard_closure(
        hazard_id=hazard.hazard_id,
        verified_by="Safety Manager"
    )

    # Generate daily report
    print(safety.generate_daily_safety_report())

Mobile App Integration

# Telegram bot for field safety inspections