Schedule Delay Analyzer

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)

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