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.