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."""
        candidates = []

        for task in self.tasks.values():
            # Check if task uses this resource on this date
            uses_resource = any(
                a.resource_id == resource_id and
                a.start_date <= on_date <= a.end_date
                for a in task.resource_assignments
            )

            if not uses_resource:
                continue

            # Check if critical path protected
            if protect_critical and task.is_critical:
                continue

            # Check if has float
            if task.total_float > 0:
                candidates.append(task)

        return candidates

    def _calculate_shift(self, task: Task, resource_id: str,
                         max_extension: int) -> Tuple[date, date]:
        """Calculate optimal shift for a task."""
        resource = self.resources[resource_id]

        # Try shifting forward
        for days in range(1, min(task.total_float + 1, max_extension + 1)):
            new_start = task.start_date + timedelta(days=days)
            new_end = task.end_date + timedelta(days=days)

            # Check if this resolves overallocation
            temp_usage = self._calculate_usage_if_moved(task.id, new_start, new_end)

            if temp_usage.get(resource_id, 0) <= resource.max_units:
                return new_start, new_end

        return None, None

    def _calculate_usage_if_moved(self, task_id: str, new_start: date,
                                   new_end: date) -> Dict[str, float]:
        """Calculate resource usage if task were moved."""
        # Simplified: calculate peak on affected dates
        usage = defaultdict(float)

        for assignment in self.assignments:
            if assignment.task_id == task_id:
                # Use new dates
                for d in pd.date_range(new_start, new_end):
                    usage[assignment.resource_id] = max(
                        usage[assignment.resource_id],
                        assignment.units
                    )
            else:
                # Use existing dates
                for d in pd.date_range(assignment.start_date, assignment.end_date):
                    usage[assignment.resource_id] = max(
                        usage[assignment.resource_id],
                        assignment.units
                    )

        return dict(usage)

    def _shift_task(self, task_id: str, new_start: date, new_end: date):
        """Shift a task to new dates."""
        task = self.tasks[task_id]
        delta = new_start - task.start_date

        # Update task
        task.start_date = new_start
        task.end_date = new_end

        # Update assignments
        for assignment in self.assignments:
            if assignment.task_id == task_id:
                assignment.start_date += delta
                assignment.end_date += delta

    def optimize_crew_size(self, resource_id: str,
                            target_utilization: float = 0.85) -> Dict:
        """Recommend optimal crew size for a resource."""
        usage = self.calculate_resource_usage()

        if resource_id not in usage.columns:
            return None

        daily_usage = usage[resource_id]
        resource = self.resources[resource_id]

        # Calculate statistics
        peak = daily_usage.max()
        avg = daily_usage.mean()
        working_days = (daily_usage > 0).sum()

        # Current utilization
        current_util = avg / resource.max_units if resource.max_units > 0 else 0

        # Optimal size for target utilization
        optimal_size = avg / target_utilization

        return {
            'resource_id': resource_id,
            'current_max_units': resource.max_units,
            'peak_usage': peak,
            'average_usage': avg,
            'working_days': int(working_days),
            'current_utilization': current_util,
            'recommended_max_units': round(optimal_size, 1),
            'potential_savings': (resource.max_units - optimal_size) * resource.cost_per_unit * working_days
        }

    def generate_histogram(self, resource_id: str) -> pd.DataFrame:
        """Generate resource histogram data."""
        usage = self.calculate_resource_usage()

        if resource_id not in usage.columns:
            return pd.DataFrame()

        resource = self.resources[resource_id]

        df = pd.DataFrame({
            'date': usage.index,
            'usage': usage[resource_id].values,
            'capacity': resource.max_units,
            'overallocated': usage[resource_id].values > resource.max_units
        })

        return df

    def generate_report(self) -> str:
        """Generate resource leveling report."""
        lines = ["# Resource Leveling Report", ""]
        lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        lines.append(f"**Resources:** {len(self.resources)}")
        lines.append(f"**Tasks:** {len(self.tasks)}")
        lines.append("")

        # Overallocations
        overallocations = self.identify_overallocations()
        if overallocations:
            lines.append("## Overallocations Found")
            for over in overallocations:
                lines.append(f"\n### {over['resource_name']}")
                lines.append(f"- **Max Units:** {over['max_units']}")
                lines.append(f"- **Peak Usage:** {over['peak_usage']}")
                lines.append(f"- **Days Overallocated:** {over['days_overallocated']}")
                lines.append(f"- **Worst Day:** {over['worst_day']}")
        else:
            lines.append("## No Overallocations")
            lines.append("All resources are within capacity.")

        # Resource utilization
        lines.append("\n## Resource Utilization")
        for resource_id in self.resources:
            opt = self.optimize_crew_size(resource_id)
            if opt:
                lines.append(f"\n### {self.resources[resource_id].name}")
                lines.append(f"- **Utilization:** {opt['current_utilization']:.1%}")
                lines.append(f"- **Peak:** {opt['peak_usage']:.1f}")
                lines.append(f"- **Average:** {opt['average_usage']:.1f}")

        return "\n".join(lines)

Quick Start

from datetime import date

# Initialize leveler
leveler = ConstructionResourceLeveler()

# Add resources
leveler.add_resource(Resource(
    id="CARP",
    name="Carpenters",
    resource_type="labor",
    max_units=10,
    cost_per_unit=75,
    unit="hours"
))

# Add tasks
leveler.add_task(Task(
    id="T1",
    name="Frame Level 1",
    duration=10,
    start_date=date(2026, 3, 1),
    end_date=date(2026, 3, 14),
    predecessors=[],
    total_float=5,
    is_critical=False
))

# Add assignments
leveler.add_assignment(ResourceAssignment(
    task_id="T1",
    resource_id="CARP",
    units=8,
    start_date=date(2026, 3, 1),
    end_date=date(2026, 3, 14)
))

# Check overallocations
overallocations = leveler.identify_overallocations()
for over in overallocations:
    print(f"{over['resource_name']}: {over['peak_usage']} vs {over['max_units']} max")

# Level resources
result = leveler.level_resources(protect_critical_path=True)
print(f"Tasks moved: {result.tasks_moved}")
print(f"End date change: {result.original_end_date} -> {result.leveled_end_date}")

# Generate report
print(leveler.generate_report())

Dependencies

pip install pandas numpy