critical-path-analyzer

name	critical-path-analyzer
description	Analyze project critical path from schedule data. Identify critical activities, calculate float, and assess schedule risk.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"📅","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

Critical Path Analyzer

Business Case

Problem Statement

Schedule management requires understanding:

• Which activities are critical?
• How much float exists?
• What delays impact completion?
• Where to focus resources?

Solution

Analyze schedule network to identify critical path, calculate float, and provide actionable schedule insights.

Technical Implementation

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
 datetime  date, timedelta
 enum  Enum
 collections  defaultdict


 ():
    NOT_STARTED = 
    IN_PROGRESS = 
    COMPLETED = 
    DELAYED = 



 :
    activity_id: 
    name: 
    duration:   
    predecessors: []
    early_start:  = 
    early_finish:  = 
    late_start:  = 
    late_finish:  = 
    total_float:  = 
    free_float:  = 
    is_critical:  = 
    status: ActivityStatus = ActivityStatus.NOT_STARTED
    percent_complete:  = 
    actual_start: [date] = 
    actual_finish: [date] = 



 :
    critical_path: []
    project_duration: 
    activities: [, Activity]
    near_critical: []  
    total_float_days: 


 :
    

    NEAR_CRITICAL_THRESHOLD =   

     ():
        .project_start = project_start
        .activities: [, Activity] = {}

     ():
        

        .activities[activity_id] = Activity(
            activity_id=activity_id,
            name=name,
            duration=duration,
            predecessors=predecessors  []
        )

     ():
        
         _, row  df.iterrows():
            preds = row.get(, )
             pd.isna(preds):
                pred_list = []
            :
                pred_list = [p.strip()  p  (preds).split()  p.strip()]

            .add_activity(
                activity_id=(row[]),
                name=row[],
                duration=(row[]),
                predecessors=pred_list
            )

     ():
        

        
        sorted_activities = ._topological_sort()

         activity_id  sorted_activities:
            activity = .activities[activity_id]

            
              activity.predecessors:
                activity.early_start = 
            :
                activity.early_start = (
                    .activities[pred].early_finish
                     pred  activity.predecessors
                     pred  .activities
                )

            activity.early_finish = activity.early_start + activity.duration

     ():
        

        
        project_duration = (a.early_finish  a  .activities.values())

        
        successors = defaultdict()
         activity_id, activity  .activities.items():
             pred  activity.predecessors:
                 pred  .activities:
                    successors[pred].append(activity_id)

        
        sorted_activities = ._topological_sort()[::-]

         activity_id  sorted_activities:
            activity = .activities[activity_id]

            
             activity_id   successors   successors[activity_id]:
                activity.late_finish = project_duration
            :
                activity.late_finish = (
                    .activities[succ].late_start
                     succ  successors[activity_id]
                )

            activity.late_start = activity.late_finish - activity.duration

            
            activity.total_float = activity.late_start - activity.early_start
            activity.is_critical = activity.total_float == 

     () -> []:
        

        visited = ()
        result = []

         ():
             activity_id  visited:
                
            visited.add(activity_id)

            activity = .activities.get(activity_id)
             activity:
                 pred  activity.predecessors:
                     pred  .activities:
                        visit(pred)
                result.append(activity_id)

         activity_id  .activities:
            visit(activity_id)

         result

     () -> CriticalPathResult:
        

        ._forward_pass()
        ._backward_pass()

        
        critical_activities = [
            a.activity_id  a  .activities.values()
             a.is_critical
        ]

        
        near_critical = [
            a.activity_id  a  .activities.values()
              < a.total_float <= .NEAR_CRITICAL_THRESHOLD
        ]

        project_duration = (a.early_finish  a  .activities.values())
        total_float = (a.total_float  a  .activities.values())

         CriticalPathResult(
            critical_path=critical_activities,
            project_duration=project_duration,
            activities=.activities,
            near_critical=near_critical,
            total_float_days=total_float
        )

     () -> pd.DataFrame:
        

        data = []
         activity  .activities.values():
            early_start_date = .project_start + timedelta(days=activity.early_start)
            early_finish_date = .project_start + timedelta(days=activity.early_finish)
            late_start_date = .project_start + timedelta(days=activity.late_start)
            late_finish_date = .project_start + timedelta(days=activity.late_finish)

            data.append({
                : activity.activity_id,
                : activity.name,
                : activity.duration,
                : early_start_date,
                : early_finish_date,
                : late_start_date,
                : late_finish_date,
                : activity.total_float,
                :   activity.is_critical  
            })

         pd.DataFrame(data)

     () -> [, ]:
        

        activity = .activities.get(activity_id)
          activity:
             {}

        absorbed_by_float = (delay_days, activity.total_float)
        project_delay = (, delay_days - activity.total_float)

        
        affected = []
         project_delay > :
            
             a  .activities.values():
                 activity_id  a.predecessors:
                    affected.append(a.activity_id)

         {
            : activity_id,
            : delay_days,
            : activity.total_float,
            : absorbed_by_float,
            : project_delay,
            : affected,
            : project_delay > 
        }

     () -> [[, ]]:
        

        result = .calculate_critical_path()
        suggestions = []

        
         activity_id  result.critical_path:
            activity = .activities[activity_id]

            
            max_reduction = (activity.duration * )

             max_reduction > :
                suggestions.append({
                    : activity_id,
                    : activity.name,
                    : activity.duration,
                    : max_reduction,
                    : 
                })

        
         (suggestions, key= x: x[], reverse=)

     () -> :
        

        result = .calculate_critical_path()

         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary_df = pd.DataFrame([{
                : .project_start,
                : result.project_duration,
                : .project_start + timedelta(days=result.project_duration),
                : (result.critical_path),
                : (result.near_critical),
                : result.total_float_days
            }])
            summary_df.to_excel(writer, sheet_name=, index=)

            
            schedule_df = .get_schedule_dates()
            schedule_df.to_excel(writer, sheet_name=, index=)

            
            critical_df = pd.DataFrame([
                {
                    : a_id,
                    : .activities[a_id].name,
                    : .activities[a_id].duration
                }
                 a_id  result.critical_path
            ])
            critical_df.to_excel(writer, sheet_name=, index=)

         output_path