schedule-compression

What this skill does

# Schedule Compression

## Overview

Compress construction schedules when project deadlines are at risk. Apply crashing (adding resources) and fast-tracking (parallel activities) to accelerate delivery while managing cost and risk.

> "Strategic compression can recover 20% of schedule with 10% cost increase" — DDC Community

## Compression Techniques

```
┌─────────────────────────────────────────────────────────────────┐
│                  SCHEDULE COMPRESSION                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  CRASHING                          FAST-TRACKING                 │
│  ────────                          ─────────────                 │
│  Add resources to reduce           Overlap sequential            │
│  activity duration                 activities                    │
│                                                                  │
│  Before:  ████████ (10d)          Before:  A ──→ B ──→ C        │
│  After:   █████ (5d) + $$$         After:   A ──→ B             │
│                                              └──→ C              │
│  Cost: Higher labor/OT             Risk: Rework if A changes    │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
```

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum

class CompressionMethod(Enum):
    CRASH = "crash"
    FAST_TRACK = "fast_track"
    HYBRID = "hybrid"

@dataclass
class Activity:
    id: str
    name: str
    normal_duration: int
    crash_duration: int  # Minimum possible duration
    normal_cost: float
    crash_cost: float  # Cost at crash duration
    predecessors: List[str] = field(default_factory=list)
    is_critical: bool = False
    current_duration: int = 0

    def __post_init__(self):
        if self.current_duration == 0:
            self.current_duration = self.normal_duration

    @property
    def crash_slope(self) -> float:
        """Cost per day of crashing."""
        duration_diff = self.normal_duration - self.crash_duration
        if duration_diff == 0:
            return float('inf')
        return (self.crash_cost - self.normal_cost) / duration_diff

    @property
    def days_available_to_crash(self) -> int:
        """Days activity can still be crashed."""
        return self.current_duration - self.crash_duration

@dataclass
class FastTrackOption:
    activity1_id: str
    activity2_id: str
    overlap_days: int
    risk_level: str  # low, medium, high
    risk_description: str
    rework_probability: float
    potential_rework_cost: float

@dataclass
class CompressionPlan:
    target_reduction: int
    achieved_reduction: int
    crash_activities: List[Tuple[str, int]]  # (activity_id, days_crashed)
    fast_track_options: List[FastTrackOption]
    total_additional_cost: float
    new_project_duration: int
    risk_assessment: str

class ScheduleCompressor:
    """Compress construction schedules using crashing and fast-tracking."""

    def __init__(self):
        self.activities: Dict[str, Activity] = {}
        self.fast_track_options: List[FastTrackOption] = []
        self.project_duration: int = 0
        self.critical_path: List[str] = []

    def add_activity(self, id: str, name: str,
                    normal_duration: int, crash_duration: int,
                    normal_cost: float, crash_cost: float,
                    predecessors: List[str] = None,
                    is_critical: bool = False) -> Activity:
        """Add activity with crash data."""
        activity = Activity(
            id=id,
            name=name,
            normal_duration=normal_duration,
            crash_duration=crash_duration,
            normal_cost=normal_cost,
            crash_cost=crash_cost,
            predecessors=predecessors or [],
            is_critical=is_critical
        )
        self.activities[id] = activity
        return activity

    def add_fast_track_option(self, activity1_id: str, activity2_id: str,
                             overlap_days: int, risk_level: str,
                             risk_description: str,
                             rework_probability: float = 0.1,
                             potential_rework_cost: float = 0) -> FastTrackOption:
        """Add fast-tracking option between activities."""
        option = FastTrackOption(
            activity1_id=activity1_id,
            activity2_id=activity2_id,
            overlap_days=overlap_days,
            risk_level=risk_level,
            risk_description=risk_description,
            rework_probability=rework_probability,
            potential_rework_cost=potential_rework_cost
        )
        self.fast_track_options.append(option)
        return option

    def calculate_project_duration(self) -> int:
        """Calculate current project duration using CPM."""
        # Simple forward pass
        finish_times = {}

        def get_finish(act_id: str) -> int:
            if act_id in finish_times:
                return finish_times[act_id]

            act = self.activities[act_id]
            if not act.predecessors:
                start = 0
            else:
                start = max(get_finish(p) for p in act.predecessors)

            finish_times[act_id] = start + act.current_duration
            return finish_times[act_id]

        for act_id in self.activities:
            get_finish(act_id)

        self.project_duration = max(finish_times.values()) if finish_times else 0
        return self.project_duration

    def identify_critical_path(self) -> List[str]:
        """Identify critical path activities."""
        # Simplified - in practice, use full CPM
        self.calculate_project_duration()

        # Mark activities with zero float as critical
        critical = [act.id for act in self.activities.values() if act.is_critical]
        self.critical_path = critical
        return critical

    def analyze_crash_options(self) -> List[Dict]:
        """Analyze all crashing options sorted by cost efficiency."""
        crash_options = []

        for act in self.activities.values():
            if act.days_available_to_crash > 0 and act.is_critical:
                crash_options.append({
                    'activity_id': act.id,
                    'activity_name': act.name,
                    'crash_slope': act.crash_slope,
                    'max_days': act.days_available_to_crash,
                    'current_duration': act.current_duration,
                    'crash_duration': act.crash_duration
                })

        # Sort by crash slope (cost per day)
        return sorted(crash_options, key=lambda x: x['crash_slope'])

    def crash_schedule(self, target_days: int,
                      max_budget: float = float('inf')) -> CompressionPlan:
        """Crash schedule to reduce duration by target days."""
        self.calculate_project_duration()
        original_duration = self.project_duration

        crashed_activities = []
        total_cost = 0
        days_achieved = 0

        # Get crash options
        options = self.analyze_crash_options()

        while days_achieved < target_days and options:
            # Find cheapest option
            best_option = None
            for opt in options:
                if opt['max_days'] > 0:
                    best_option = opt
                    break

            if not best_option:
                break

            # Crash by 1 day
            act = self.activities[best_option['activity_id']]
            crash_cost = act.crash_slope

            if total_cost + crash_cost > max_budget:
                break

            act.current_duration -= 1
            total_cost += crash_cost
            days_achieved += 1

            # Update option
            best_option['max_days'] -= 1