Critical Path Analyzer
Overview
Analyze construction schedules using Critical Path Method (CPM). Identify critical activities, calculate total and free float, simulate what-if scenarios, and find opportunities to compress schedule.
"Understanding the critical path is essential for effective schedule management" — DDC Community
CPM Concepts
┌─────────────────────────────────────────────────────────────────┐
│ CRITICAL PATH METHOD │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Activity A (5d) ──→ Activity C (10d) ──→ Activity E (3d) │
│ ↓ ↑ │
│ Activity B (8d) ──────────→ Activity D (6d) ───┘ │
│ │
│ Critical Path: A → C → E (18 days) │
│ Float on B→D: 4 days │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set, Tuple
from datetime import datetime, timedelta
from collections import defaultdict
import heapq
@dataclass
class Activity:
id: str
name: str
duration: int # days
predecessors: List[str] = field(default_factory=list)
successors: List[str] = field(default_factory=list)
# Calculated values
early_start: int = 0
early_finish: int = 0
late_start: int = 0
late_finish: int = 0
total_float: int = 0
free_float: int = 0
is_critical: bool = False
# Optional attributes
resources: List[str] = field(default_factory=list)
cost: float = 0.0
actual_start: Optional[datetime] = None
actual_finish: Optional[datetime] = None
percent_complete: float = 0.0
@dataclass
class ScheduleAnalysis:
project_duration: int
critical_path: List[str]
critical_activities: List[Activity]
near_critical_activities: List[Activity] # Float <= threshold
total_activities: int
float_distribution: Dict[int, int]
class CriticalPathAnalyzer:
"""Analyze construction schedule critical path."""
def __init__(self):
self.activities: Dict[str, Activity] = {}
self.analysis: Optional[ScheduleAnalysis] = None
def add_activity(self, id: str, name: str, duration: int,
predecessors: List[str] = None,
resources: List[str] = None,
cost: float = 0.0) -> Activity:
"""Add activity to schedule."""
activity = Activity(
id=id,
name=name,
duration=duration,
predecessors=predecessors or [],
resources=resources or [],
cost=cost
)
self.activities[id] = activity
# Update successor relationships
for pred_id in activity.predecessors:
if pred_id in self.activities:
self.activities[pred_id].successors.append(id)
return activity
def import_schedule(self, activities: List[Dict]) -> int:
"""Import activities from list of dicts."""
count = 0
for act in activities:
self.add_activity(
id=act['id'],
name=act['name'],
duration=act['duration'],
predecessors=act.get('predecessors', []),
resources=act.get('resources', []),
cost=act.get('cost', 0)
)
count += 1
return count
def calculate_critical_path(self, near_critical_threshold: int = 5) -> ScheduleAnalysis:
"""Calculate critical path using forward and backward pass."""
if not self.activities:
raise ValueError("No activities to analyze")
# Reset calculations
for act in self.activities.values():
act.early_start = 0
act.early_finish = 0
act.late_start = 0
act.late_finish = 0
act.total_float = 0
act.free_float = 0
act.is_critical = False
# Forward pass - calculate early start/finish
self._forward_pass()
# Determine project duration
project_duration = max(act.early_finish for act in self.activities.values())
# Backward pass - calculate late start/finish
self._backward_pass(project_duration)
# Calculate float and identify critical activities
critical_path = []
critical_activities = []
near_critical = []
for act in self.activities.values():
act.total_float = act.late_start - act.early_start
# Calculate free float
if act.successors:
min_succ_es = min(self.activities[s].early_start for s in act.successors)
act.free_float = min_succ_es - act.early_finish
else:
act.free_float = project_duration - act.early_finish
# Identify critical (zero float)
if act.total_float == 0:
act.is_critical = True
critical_activities.append(act)
elif act.total_float <= near_critical_threshold:
near_critical.append(act)
# Build critical path sequence
critical_path = self._build_critical_path_sequence(critical_activities)
# Float distribution
float_dist = defaultdict(int)
for act in self.activities.values():
float_dist[act.total_float] += 1
self.analysis = ScheduleAnalysis(
project_duration=project_duration,
critical_path=critical_path,
critical_activities=critical_activities,
near_critical_activities=near_critical,
total_activities=len(self.activities),
float_distribution=dict(float_dist)
)
return self.analysis
def _forward_pass(self):
"""Forward pass to calculate early start/finish."""
# Topological sort
in_degree = {act_id: len(act.predecessors) for act_id, act in self.activities.items()}
queue = [act_id for act_id, degree in in_degree.items() if degree == 0]
while queue:
act_id = queue.pop(0)
act = self.activities[act_id]
# Early start is max of predecessor early finishes
if act.predecessors:
act.early_start = max(
self.activities[pred].early_finish for pred in act.predecessors
)
else:
act.early_start = 0
act.early_finish = act.early_start + act.duration
# Process successors
for succ_id in act.successors:
in_degree[succ_id] -= 1
if in_degree[succ_id] == 0:
queue.append(succ_id)
def _backward_pass(self, project_duration: int):
"""Backward pass to calculate late start/finish."""
# Start from end activities
end_activities = [act for act in self.activities.values() if not act.successors]
for act in end_activities:
act.late_finish = project_duration
act.late_start = act.late_finish - act.duration
# Process in reverse topological order
processed = set(act.id for act in end_activities)
queue = list(end_activities)
while queue:
act = queue.pop(0)
for pred_id in act.predecessors:
pred = self.activities[pred_id]
# Late finish is min of successor late starts
new_late_finish = act.late_start
if pred.late_finish == 0 or new_late_finish < pred.late_finish:
pred.late_finish = new_late_finish
pred.