Incident Reporting System

Overview

Comprehensive incident reporting system for construction safety. Capture near-misses, injuries, and property damage. Conduct root cause analysis and track corrective actions to prevent recurrence.

"Near-miss reporting prevents 90% of future serious incidents" — DDC Community

Incident Pyramid

                    △
                   /│\        Fatality (1)
                  / │ \
                 /  │  \      Serious Injury (10)
                /   │   \
               /    │    \    Minor Injury (30)
              /     │     \
             /      │      \  Near Miss (300)
            /       │       \
           /        │        \ Unsafe Acts (3000)
          ──────────┴──────────

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime, timedelta
import json

class IncidentType(Enum):
    NEAR_MISS = "near_miss"
    FIRST_AID = "first_aid"
    MEDICAL_TREATMENT = "medical_treatment"
    LOST_TIME = "lost_time"
    FATALITY = "fatality"
    PROPERTY_DAMAGE = "property_damage"
    ENVIRONMENTAL = "environmental"

class IncidentCategory(Enum):
    FALL = "fall"
    STRUCK_BY = "struck_by"
    CAUGHT_IN = "caught_in"
    ELECTROCUTION = "electrocution"
    VEHICLE = "vehicle"
    MATERIAL_HANDLING = "material_handling"
    TOOL_EQUIPMENT = "tool_equipment"
    SLIP_TRIP = "slip_trip"
    FIRE = "fire"
    CHEMICAL = "chemical"
    OTHER = "other"

class InvestigationStatus(Enum):
    REPORTED = "reported"
    UNDER_INVESTIGATION = "under_investigation"
    ROOT_CAUSE_IDENTIFIED = "root_cause_identified"
    CORRECTIVE_ACTIONS_ASSIGNED = "corrective_actions_assigned"
    IN_REMEDIATION = "in_remediation"
    CLOSED = "closed"

@dataclass
class Person:
    name: str
    company: str
    role: str
    contact: str
    years_experience: int = 0

@dataclass
class CorrectiveAction:
    id: str
    description: str
    assigned_to: str
    due_date: datetime
    status: str = "open"
    completed_date: Optional[datetime] = None
    verification_notes: str = ""

@dataclass
class Incident:
    id: str
    incident_type: IncidentType
    category: IncidentCategory
    date_time: datetime
    location: str
    project_id: str
    project_name: str

    # Description
    description: str
    immediate_actions: str

    # People involved
    injured_person: Optional[Person] = None
    witnesses: List[Person] = field(default_factory=list)
    reported_by: str = ""

    # Investigation
    status: InvestigationStatus = InvestigationStatus.REPORTED
    root_causes: List[str] = field(default_factory=list)
    contributing_factors: List[str] = field(default_factory=list)
    corrective_actions: List[CorrectiveAction] = field(default_factory=list)

    # Documentation
    photos: List[str] = field(default_factory=list)
    weather_conditions: str = ""
    equipment_involved: List[str] = field(default_factory=list)

    # Metrics
    days_lost: int = 0
    property_damage_cost: float = 0.0
    osha_recordable: bool = False

class IncidentManager:
    """Manage construction incident reporting and investigation."""

    # 5 Whys root cause categories
    ROOT_CAUSE_CATEGORIES = [
        "Training/Competency",
        "Procedures/Work Instructions",
        "Equipment/Tools",
        "Supervision",
        "Communication",
        "Housekeeping",
        "PPE",
        "Work Environment",
        "Physical/Mental State",
        "Management System"
    ]

    def __init__(self):
        self.incidents: Dict[str, Incident] = {}
        self.corrective_actions: Dict[str, CorrectiveAction] = {}

    def report_incident(self, incident_type: IncidentType,
                       category: IncidentCategory,
                       date_time: datetime,
                       location: str,
                       project_id: str,
                       project_name: str,
                       description: str,
                       immediate_actions: str,
                       reported_by: str,
                       injured_person: Dict = None) -> Incident:
        """Report new incident."""
        incident_id = f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}"

        injured = None
        if injured_person:
            injured = Person(**injured_person)

        incident = Incident(
            id=incident_id,
            incident_type=incident_type,
            category=category,
            date_time=date_time,
            location=location,
            project_id=project_id,
            project_name=project_name,
            description=description,
            immediate_actions=immediate_actions,
            reported_by=reported_by,
            injured_person=injured
        )

        # Auto-flag OSHA recordable
        if incident_type in [IncidentType.MEDICAL_TREATMENT,
                            IncidentType.LOST_TIME,
                            IncidentType.FATALITY]:
            incident.osha_recordable = True

        self.incidents[incident_id] = incident
        return incident

    def add_witness(self, incident_id: str, witness: Dict) -> Incident:
        """Add witness to incident."""
        if incident_id not in self.incidents:
            raise ValueError(f"Incident {incident_id} not found")

        self.incidents[incident_id].witnesses.append(Person(**witness))
        return self.incidents[incident_id]

    def conduct_investigation(self, incident_id: str,
                             root_causes: List[str],
                             contributing_factors: List[str]) -> Incident:
        """Record investigation findings."""
        if incident_id not in self.incidents:
            raise ValueError(f"Incident {incident_id} not found")

        incident = self.incidents[incident_id]
        incident.root_causes = root_causes
        incident.contributing_factors = contributing_factors
        incident.status = InvestigationStatus.ROOT_CAUSE_IDENTIFIED
        return incident

    def five_whys_analysis(self, incident_id: str, whys: List[str]) -> Dict:
        """Conduct 5 Whys analysis."""
        if incident_id not in self.incidents:
            raise ValueError(f"Incident {incident_id} not found")

        analysis = {
            "incident_id": incident_id,
            "analysis_date": datetime.now().isoformat(),
            "whys": []
        }

        for i, why in enumerate(whys):
            analysis["whys"].append({
                "level": i + 1,
                "question": f"Why #{i+1}?",
                "answer": why
            })

        # The last "why" is typically the root cause
        if whys:
            self.incidents[incident_id].root_causes.append(whys[-1])

        return analysis

    def assign_corrective_action(self, incident_id: str,
                                description: str,
                                assigned_to: str,
                                due_days: int = 7) -> CorrectiveAction:
        """Assign corrective action."""
        if incident_id not in self.incidents:
            raise ValueError(f"Incident {incident_id} not found")

        action_id = f"CA-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        action = CorrectiveAction(
            id=action_id,
            description=description,
            assigned_to=assigned_to,
            due_date=datetime.now() + timedelta(days=due_days)
        )

        self.incidents[incident_id].corrective_actions.append(action)
        self.incidents[incident_id].status = InvestigationStatus.CORRECTIVE_ACTIONS_ASSIGNED
        self.corrective_actions[action_id] = action
        return action

    def complete_corrective_action(self, action_id: str,
                                   verification_notes: str) -> CorrectiveAction:
        """Mark corrective action complete."""
        if action_id not in self.corrective_actions:
            raise ValueError(f"Corrective action {action_id}