cpm_engine.py

"""
Ask SURI - Critical Path Method (CPM) Engine
This file contains a simplified conceptual implementation of the CPM forward and backward pass algorithms.
"""

from typing import List, Dict, Set
from datetime import datetime, timedelta

class Activity:
    def __init__(self, id: str, duration: int):
        self.id = id
        self.duration = duration
        self.predecessors: List['Activity'] = []
        self.successors: List['Activity'] = []
        
        # Computed dates (integer days from project start for simplicity)
        self.early_start = 0
        self.early_finish = 0
        self.late_start = 0
        self.late_finish = 0
        self.total_float = 0

class CPMEngine:
    def __init__(self, activities: List[Activity]):
        self.activities = activities
        self.project_duration = 0
        
    def forward_pass(self):
        """
        Computes Early Start (ES) and Early Finish (EF) for all activities.
        """
        # Topological sort ensures we process predecessors before successors
        sorted_activities = self._topological_sort()
        
        for activity in sorted_activities:
            if not activity.predecessors:
                activity.early_start = 0
            else:
                # ES is the maximum EF of all predecessors
                activity.early_start = max(pred.early_finish for pred in activity.predecessors)
                
            activity.early_finish = activity.early_start + activity.duration
            
        # Project duration is the maximum EF of all activities
        self.project_duration = max(act.early_finish for act in self.activities)
        
    def backward_pass(self):
        """
        Computes Late Start (LS), Late Finish (LF), and Total Float (TF) for all activities.
        """
        # Reverse topological sort ensures we process successors before predecessors
        sorted_activities = reversed(self._topological_sort())
        
        for activity in sorted_activities:
            if not activity.successors:
                activity.late_finish = self.project_duration
            else:
                # LF is the minimum LS of all successors
                activity.late_finish = min(succ.late_start for succ in activity.successors)
                
            activity.late_start = activity.late_finish - activity.duration
            
            # Total Float = LS - ES (or LF - EF)
            activity.total_float = activity.late_start - activity.early_start
            
    def get_critical_path(self) -> List[Activity]:
        """
        Returns activities on the critical path (Total Float <= 0).
        """
        return [act for act in self.activities if act.total_float <= 0]
        
    def _topological_sort(self) -> List[Activity]:
        """
        Helper method to sort activities topologically based on dependencies.
        (Implementation omitted for brevity)
        """
        # In a real implementation, this would use Kahn's algorithm or DFS
        return self.activities