Schedule Delay Analyzer

Technical Implementation

import pandas as pd from datetime import date, timedelta from typing import Dict, Any, List, Optional from dataclasses import dataclass, field from enum import Enum

class DelayType(Enum): EXCUSABLE_COMPENSABLE = "excusable_compensable" EXCUSABLE_NON_COMPENSABLE = "excusable_non_compensable" NON_EXCUSABLE = "non_excusable" CONCURRENT = "concurrent"

class DelayCause(Enum): OWNER_CHANGE = "owner_change" DESIGN_ERROR = "design_error" WEATHER = "weather" DIFFERING_CONDITIONS = "differing_conditions" CONTRACTOR_ISSUE = "contractor_issue" MATERIAL_DELAY = "material_delay" LABOR_SHORTAGE = "labor_shortage" PERMIT_DELAY = "permit_delay" OTHER = "other"

@dataclass class DelayEvent: delay_id: str activity_id: str activity_name: str delay_type: DelayType cause: DelayCause start_date: date end_date: date delay_days: int on_critical_path: bool description: str documentation: List[str] = field(default_factory=list) cost_impact: float = 0.0

@dataclass class ScheduleBaseline: baseline_date: date planned_completion: date activities: Dict[str, Dict[str, date]] # activity_id: {start, end}

class ScheduleDelayAnalyzer: def init(self, project_name: str, contract_completion: date): self.project_name = project_name self.contract_completion = contract_completion self.baselines: List[ScheduleBaseline] = [] self.delays: Dict[str, DelayEvent] = {} self._counter = 0

def add_baseline(self, baseline_date: date, planned_completion: date,
                activities: Dict[str, Dict[str, date]]):
    baseline = ScheduleBaseline(baseline_date, planned_completion, activities)
    self.baselines.append(baseline)

def record_delay(self, activity_id: str, activity_name: str,
                delay_type: DelayType, cause: DelayCause,
                start_date: date, end_date: date,
                on_critical_path: bool, description: str,
                cost_impact: float = 0) -> DelayEvent:
    self._counter += 1
    delay_id = f"DLY-{self._counter:04d}"

    delay = DelayEvent(
        delay_id=delay_id,
        activity_id=activity_id,
        activity_name=activity_name,
        delay_type=delay_type,
        cause=cause,
        start_date=start_date,
        end_date=end_date,
        delay_days=(end_date - start_date).days,
        on_critical_path=on_critical_path,
        description=description,
        cost_impact=cost_impact
    )
    self.delays[delay_id] = delay
    return delay

def calculate_project_delay(self) -> int:
    """Calculate total critical path delay."""
    critical_delays = [d for d in self.delays.values() if d.on_critical_path]
    return sum(d.delay_days for d in critical_delays)

def analyze_by_type(self) -> Dict[str, Dict[str, Any]]:
    analysis = {}
    for delay in self.delays.values():
        dtype = delay.delay_type.value
        if dtype not in analysis:
            analysis[dtype] = {'count': 0, 'days': 0, 'cost': 0}
        analysis[dtype]['count'] += 1
        analysis[dtype]['days'] += delay.delay_days
        analysis[dtype]['cost'] += delay.cost_impact
    return analysis

def analyze_by_cause(self) -> Dict[str, int]:
    by_cause = {}
    for delay in self.delays.values():
        cause = delay.cause.value
        by_cause[cause] = by_cause.get(cause, 0) + delay.delay_days
    return by_cause

def calculate_time_extension_claim(self) -> Dict[str, Any]:
    """Calculate basis for time extension claim."""
    excusable = [d for d in self.delays.values()
                if d.delay_type in [DelayType.EXCUSABLE_COMPENSABLE,
                                    DelayType.EXCUSABLE_NON_COMPENSABLE]
                and d.on_critical_path]

    compensable = [d for d in excusable
                  if d.delay_type == DelayType.EXCUSABLE_COMPENSABLE]

    return {
        'excusable_delays': len(excusable),
        'excusable_days': sum(d.delay_days for d in excusable),
        'compensable_delays': len(compensable),
        'compensable_days': sum(d.delay_days for d in compensable),
        'total_cost_impact': sum(d.cost_impact for d in compensable),
        'recommended_extension': sum(d.delay_days for d in excusable)
    }

def get_summary(self) -> Dict[str, Any]:
    critical_delay = self.calculate_project_delay()
    projected_completion = self.contract_completion + timedelta(days=critical_delay)

    return {
        'project': self.project_name,
        'contract_completion': self.contract_completion,
        'projected_completion': projected_completion,
        'total_delays': len(self.delays),
        'critical_path_delays': sum(1 for d in self.delays.values() if d.on_critical_path),
        'total_delay_days': critical_delay,
        'by_type': self.analyze_by_type(),
        'by_cause': self.analyze_by_cause()
    }

def export_analysis(self, output_path: str):
    data = [{
        'ID': d.delay_id,
        'Activity': d.activity_name,
        'Type': d.delay_type.value,
        'Cause': d.cause.value,
        'Start': d.start_date,
        'End': d.end_date,
        'Days': d.delay_days,
        'Critical': d.on_critical_path,
        'Cost Impact': d.cost_impact,
        'Description': d.description
    } for d in self.delays.values()]
    pd.DataFrame(data).to_excel(output_path, index=False)

Quick Start

analyzer = ScheduleDelayAnalyzer("Office Tower", date(2024, 12, 31))

delay = analyzer.record_delay( activity_id="A-300", activity_name="Foundation Work", delay_type=DelayType.EXCUSABLE_COMPENSABLE, cause=DelayCause.OWNER_CHANGE, start_date=date(2024, 3, 1), end_date=date(2024, 3, 15), on_critical_path=True, description="Owner requested additional scope", cost_impact=50000 )

summary = analyzer.get_summary() print(f"Project delayed by {summary['total_delay_days']} days")

claim = analyzer.calculate_time_extension_claim() print(f"Recommended extension: {claim['recommended_extension']} days")

Resources

• DDC Book: Chapter 3.3 - Schedule Management