Safety Inspection System for Construction

Comprehensive digital safety management system for construction sites with inspection checklists, hazard tracking, and incident reporting.

Business Case

Problem: Paper-based safety management leads to:

• Incomplete inspections (20-30% of items skipped)

• Lost documentation

• Delayed incident reporting

• Difficulty tracking corrective actions

• Compliance audit failures

Solution: Digital system that:

• Enforces complete checklist completion

• Photos attached to each finding

• Instant notifications for hazards

• Tracks corrective actions to closure

• Generates compliance reports

ROI: 40% reduction in recordable incidents, 100% audit compliance

Safety Inspection Types

┌──────────────────────────────────────────────────────────────────────┐ │ SAFETY INSPECTION TYPES │ ├──────────────────────────────────────────────────────────────────────┤ │ │ │ DAILY WEEKLY SPECIAL │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Pre-Work │ │ Area Walk │ │ Pre-Pour │ │ │ │ • Housekeeping│ │ • Fire ext. │ │ • Formwork │ │ │ │ • PPE │ │ • First aid │ │ • Shoring │ │ │ │ • Equipment │ │ • Scaffolds │ │ └─────────────┘ │ │ └─────────────┘ │ • Trenches │ ┌─────────────┐ │ │ ┌─────────────┐ └─────────────┘ │ Crane Setup │ │ │ │ Toolbox Talk│ ┌─────────────┐ │ • Ground │ │ │ │ • Topic │ │ Equipment │ │ • Load chart│ │ │ │ • Attendees │ │ • Cranes │ │ • Rigging │ │ │ │ • Sign-off │ │ • Lifts │ └─────────────┘ │ │ └─────────────┘ │ • Vehicles │ ┌─────────────┐ │ │ ┌─────────────┐ └─────────────┘ │ Hot Work │ │ │ │ End of Day │ │ • Permit │ │ │ │ • Secured │ │ • Fire watch│ │ │ │ • Barricades│ └─────────────┘ │ │ └─────────────┘ │ │ │ └──────────────────────────────────────────────────────────────────────┘

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

name: Safety Inspection Bot commands: /inspection: - Select inspection type - Show checklist - Record results (✅/❌) - Capture photos - Submit

/hazard: - Describe hazard - Select severity - Take photo - GPS location - Assign responsible party

/incident: - Report type - Description - Photos - Witness info - Immediate actions

"Safety is not a priority - it's a value. Priorities change, values don't."