schedule-compression

name	schedule-compression
description	Compress construction schedules using crashing and fast-tracking techniques. Analyze cost-time tradeoffs and find optimal acceleration strategies.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"⏱️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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

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 = .normal_duration - .crash_duration
         duration_diff == :
             ()
         (.crash_cost - .normal_cost) / duration_diff


     () -> :
        
         .current_duration - .crash_duration


 :
    activity1_id: 
    activity2_id: 
    overlap_days: 
    risk_level:   
    risk_description: 
    rework_probability: 
    potential_rework_cost: 


 :
    target_reduction: 
    achieved_reduction: 
    crash_activities: [[, ]]  
    fast_track_options: [FastTrackOption]
    total_additional_cost: 
    new_project_duration: 
    risk_assessment: 

 :
    

     ():
        .activities: [, Activity] = {}
        .fast_track_options: [FastTrackOption] = []
        .project_duration:  = 
        .critical_path: [] = []

     () -> Activity:
        
        activity = Activity(
            =,
            name=name,
            normal_duration=normal_duration,
            crash_duration=crash_duration,
            normal_cost=normal_cost,
            crash_cost=crash_cost,
            predecessors=predecessors  [],
            is_critical=is_critical
        )
        .activities[] = activity
         activity

     () -> FastTrackOption:
        
        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
        )
        .fast_track_options.append(option)
         option

     () -> :
        
        
        finish_times = {}

         () -> :
             act_id  finish_times:
                 finish_times[act_id]

            act = .activities[act_id]
              act.predecessors:
                start = 
            :
                start = (get_finish(p)  p  act.predecessors)

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

         act_id  .activities:
            get_finish(act_id)

        .project_duration = (finish_times.values())  finish_times  
         .project_duration

     () -> []:
        
        
        .calculate_project_duration()

        
        critical = [act.  act  .activities.values()  act.is_critical]
        .critical_path = critical
         critical

     () -> []:
        
        crash_options = []

         act  .activities.values():
             act.days_available_to_crash >   act.is_critical:
                crash_options.append({
                    : act.,
                    : act.name,
                    : act.crash_slope,
                    : act.days_available_to_crash,
                    : act.current_duration,
                    : act.crash_duration
                })

        
         (crash_options, key= x: x[])

     () -> CompressionPlan:
        
        .calculate_project_duration()
        original_duration = .project_duration

        crashed_activities = []
        total_cost = 
        days_achieved = 

        
        options = .analyze_crash_options()

         days_achieved < target_days  options:
            
            best_option = 
             opt  options:
                 opt[] > :
                    best_option = opt
                    

              best_option:
                

            
            act = .activities[best_option[]]
            crash_cost = act.crash_slope

             total_cost + crash_cost > max_budget:
                

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

            
            best_option[] -= 

            
            existing = ((c  c  crashed_activities  c[] == act.), )
             existing:
                crashed_activities.remove(existing)
                crashed_activities.append((act., existing[] + ))
            :
                crashed_activities.append((act., ))

            
            options = .analyze_crash_options()

        new_duration = .calculate_project_duration()

         CompressionPlan(
            target_reduction=target_days,
            achieved_reduction=days_achieved,
            crash_activities=crashed_activities,
            fast_track_options=[],
            total_additional_cost=total_cost,
            new_project_duration=new_duration,
            risk_assessment=
        )

     () -> CompressionPlan:
        
        risk_order = {: , : , : }
        max_risk_level = risk_order.get(max_risk, )

        
        viable_options = [
            opt  opt  .fast_track_options
             risk_order.get(opt.risk_level, ) <= max_risk_level
        ]

        
        viable_options.sort(key= x: -x.overlap_days)

        selected_options = []
        total_overlap = 
        total_risk_cost = 

         opt  viable_options:
             total_overlap >= target_days:
                

            selected_options.append(opt)
            total_overlap += opt.overlap_days
            total_risk_cost += opt.rework_probability * opt.potential_rework_cost

        .calculate_project_duration()
        new_duration = .project_duration - total_overlap

        risk_text =   max_risk ==     max_risk ==   

         CompressionPlan(
            target_reduction=target_days,
            achieved_reduction=total_overlap,
            crash_activities=[],
            fast_track_options=selected_options,
            total_additional_cost=total_risk_cost,
            new_project_duration=new_duration,
            risk_assessment=
        )

     () -> CompressionPlan:
        
        
        crash_plan = .crash_schedule(target_days, max_budget)

         crash_plan.achieved_reduction >= target_days:
             crash_plan

        
        remaining_days = target_days - crash_plan.achieved_reduction
        remaining_budget = max_budget - crash_plan.total_additional_cost

        fast_track_plan = .fast_track_schedule(remaining_days, max_risk)

        
        total_reduction = crash_plan.achieved_reduction + fast_track_plan.achieved_reduction
        total_cost = crash_plan.total_additional_cost + fast_track_plan.total_additional_cost

         CompressionPlan(
            target_reduction=target_days,
            achieved_reduction=total_reduction,
            crash_activities=crash_plan.crash_activities,
            fast_track_options=fast_track_plan.fast_track_options,
            total_additional_cost=total_cost,
            new_project_duration=.project_duration - total_reduction,
            risk_assessment=
        )

     () -> []:
        
        curve = []
        .calculate_project_duration()
        original_duration = .project_duration

        
         act  .activities.values():
            act.current_duration = act.normal_duration

        base_cost = (act.normal_cost  act  .activities.values())

        curve.append({
            : original_duration,
            : base_cost,
            : 
        })

         days  (, max_compression + ):
            
             act  .activities.values():
                act.current_duration = act.normal_duration

            plan = .crash_schedule(days)

             plan.achieved_reduction < days:
                

            curve.append({
                : plan.new_project_duration,
                : base_cost + plan.total_additional_cost,
                : days
            })

         curve

     () -> :
        
        lines = [
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            ,
            
        ]

         plan.crash_activities:
            lines.append()
            lines.append()
            lines.append()
            lines.append()
             act_id, days  plan.crash_activities:
                act = .activities[act_id]
                cost = days * act.crash_slope
                lines.append()
            lines.append()

         plan.fast_track_options:
            lines.append()
            lines.append()
             opt  plan.fast_track_options:
                lines.append()
                lines.append()
            lines.append()

         .join(lines)

Quick Start

# Initialize compressor
compressor = ScheduleCompressor()

# Add activities with crash data
compressor.add_activity(
    "A", "Foundation",
    normal_duration=20, crash_duration=15,
    normal_cost=100000, crash_cost=130000,
    is_critical=True
)
compressor.add_activity(
    "B", "Steel Erection",
    normal_duration=30, crash_duration=22,
    normal_cost=200000, crash_cost=260000,
    predecessors=["A"],
    is_critical=True
)
compressor.add_activity(
    "C", "MEP Rough-in",
    normal_duration=25, crash_duration=20,
    normal_cost=150000, crash_cost=180000,
    predecessors=["B"],
    is_critical=True
)

# Add fast-track options
compressor.add_fast_track_option(
    "A", "B", overlap_days=5,
    risk_level="medium",
    risk_description="Steel may need rework if foundation changes",
    rework_probability=0.15,
    potential_rework_cost=30000
)

# Analyze crash options
print("Crash Options (sorted by cost/day):")
for opt in compressor.analyze_crash_options():
    print(f"  {opt['activity_name']}: ${opt['crash_slope']:.0f}/day, max  days")


plan = compressor.optimize_compression(
    target_days=,
    max_budget=,
    max_risk=
)

(compressor.generate_compression_report(plan))


curve = compressor.generate_cost_curve()
 point  curve:
    ()

Requirements

pip install (no external dependencies)