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