Resource Leveler for Construction

Overview

Optimize resource allocation across construction schedules. Level labor and equipment to avoid peaks, balance workload, and maintain project deadlines while reducing costs.

Business Case

Resource leveling provides:

• Cost Reduction: Avoid overtime and idle time
• Workforce Stability: Consistent crew sizes
• Equipment Optimization: Reduce rental costs
• Realistic Schedules: Achievable resource plans

Technical Implementation

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

@dataclass
class Resource:
    id: str
    name: str
    resource_type: str  # labor, equipment, material
    max_units: float
    cost_per_unit: float
    unit: str  # hours, days, each

@dataclass
class ResourceAssignment:
    task_id: str
    resource_id: str
    units: float
    start_date: date
    end_date: date

@dataclass
class Task:
    id: str
    name: str
    duration: int  # days
    start_date: date
    end_date: date
    predecessors: List[str]
    total_float: int
    is_critical: bool
    resource_assignments: List[ResourceAssignment] = field(default_factory=list)

@dataclass
class LevelingResult:
    success: bool
    original_end_date: date
    leveled_end_date: date
    tasks_moved: int
    peak_reduction: Dict[str, float]
    warnings: List[str]

class ConstructionResourceLeveler:
    """Level resources across construction schedules."""

    def __init__(self):
        self.resources: Dict[str, Resource] = {}
        self.tasks: Dict[str, Task] = {}
        self.assignments: List[ResourceAssignment] = []

    def add_resource(self, resource: Resource):
        """Add a resource to the pool."""
        self.resources[resource.id] = resource

    def add_task(self, task: Task):
        """Add a task to the schedule."""
        self.tasks[task.id] = task

    def add_assignment(self, assignment: ResourceAssignment):
        """Assign a resource to a task."""
        self.assignments.append(assignment)
        if assignment.task_id in self.tasks:
            self.tasks[assignment.task_id].resource_assignments.append(assignment)

    def calculate_resource_usage(self, start_date: date = None,
                                  end_date: date = None) -> pd.DataFrame:
        """Calculate daily resource usage."""
        if not self.assignments:
            return pd.DataFrame()

        # Determine date range
        if start_date is None:
            start_date = min(a.start_date for a in self.assignments)
        if end_date is None:
            end_date = max(a.end_date for a in self.assignments)

        # Create date range
        dates = pd.date_range(start_date, end_date, freq='D')

        # Initialize usage matrix
        usage = {r_id: [0.0] * len(dates) for r_id in self.resources}

        # Fill in usage
        for assignment in self.assignments:
            if assignment.resource_id in usage:
                for i, d in enumerate(dates):
                    if assignment.start_date <= d.date() <= assignment.end_date:
                        usage[assignment.resource_id][i] += assignment.units

        df = pd.DataFrame(usage, index=dates)
        df.index.name = 'date'

        return df

    def identify_overallocations(self) -> List[Dict]:
        """Identify resource overallocations."""
        usage = self.calculate_resource_usage()
        overallocations = []

        for resource_id, resource in self.resources.items():
            if resource_id in usage.columns:
                daily_usage = usage[resource_id]
                over_days = daily_usage[daily_usage > resource.max_units]

                if len(over_days) > 0:
                    overallocations.append({
                        'resource_id': resource_id,
                        'resource_name': resource.name,
                        'max_units': resource.max_units,
                        'peak_usage': daily_usage.max(),
                        'over_by': daily_usage.max() - resource.max_units,
                        'days_overallocated': len(over_days),
                        'first_overallocation': over_days.index[0].date(),
                        'worst_day': daily_usage.idxmax().date()
                    })

        return overallocations

    def level_resources(self, method: str = 'float_priority',
                        protect_critical_path: bool = True,
                        max_extension: int = 30) -> LevelingResult:
        """Level resources to resolve overallocations."""

        original_end = max(t.end_date for t in self.tasks.values())
        tasks_moved = 0
        warnings = []

        # Get initial overallocations
        initial_over = self.identify_overallocations()
        if not initial_over:
            return LevelingResult(
                success=True,
                original_end_date=original_end,
                leveled_end_date=original_end,
                tasks_moved=0,
                peak_reduction={},
                warnings=["No overallocations found"]
            )

        # Track peak usage before
        usage_before = self.calculate_resource_usage()
        peaks_before = {r: usage_before[r].max() for r in usage_before.columns}

        # Leveling loop
        iteration = 0
        max_iterations = len(self.tasks) * 2

        while iteration < max_iterations:
            iteration += 1
            overallocations = self.identify_overallocations()

            if not overallocations:
                break

            # Find task to move
            moved = False
            for over in overallocations:
                resource_id = over['resource_id']
                worst_day = over['worst_day']

                # Find tasks using this resource on worst day
                candidates = self._find_movable_tasks(
                    resource_id, worst_day, protect_critical_path
                )

                if candidates:
                    # Sort by priority (lowest float first to preserve options)
                    candidates.sort(key=lambda t: -t.total_float)
                    task_to_move = candidates[0]

                    # Calculate new dates
                    new_start, new_end = self._calculate_shift(
                        task_to_move, resource_id, max_extension
                    )

                    if new_start:
                        self._shift_task(task_to_move.id, new_start, new_end)
                        tasks_moved += 1
                        moved = True
                        break

            if not moved:
                warnings.append("Could not resolve all overallocations")
                break

        # Calculate results
        usage_after = self.calculate_resource_usage()
        peaks_after = {r: usage_after[r].max() for r in usage_after.columns}

        peak_reduction = {}
        for r in peaks_before:
            if r in peaks_after:
                reduction = (peaks_before[r] - peaks_after[r]) / peaks_before[r] * 100
                peak_reduction[r] = reduction

        leveled_end = max(t.end_date for t in self.tasks.values())

        if leveled_end > original_end + timedelta(days=max_extension):
            warnings.append(f"Project extended beyond max allowed ({max_extension} days)")

        remaining_over = self.identify_overallocations()

        return LevelingResult(
            success=len(remaining_over) == 0,
            original_end_date=original_end,
            leveled_end_date=leveled_end,
            tasks_moved=tasks_moved,
            peak_reduction=peak_reduction,
            warnings=warnings
        )

    def _find_movable_tasks(self, resource_id: str, on_date: date,
                            protect_critical: bool) -> List[Task]:
        """Find tasks that can be moved to reduce overallocation."""