Resource Allocation Optimizer
Overview
Optimize resource allocation in construction schedules. Level workforce and equipment utilization, resolve over-allocations, and balance workload across the project duration.
"Resource leveling reduces peak demand by 30% and improves productivity" — DDC Community
Resource Leveling Concept
Before Leveling: After Leveling:
Workers Workers
20│ ████ 15│ ████████████
15│ ████████ 10│████████████████
10│████████████ 5│████████████████████
5│██████████████████ 0└──────────────────────
0└──────────────────── Week 1 2 3 4 5 6
Week 1 2 3 4 5
Peak reduced, duration extended
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from collections import defaultdict
import heapq
@dataclass
class Resource:
id: str
name: str
resource_type: str # labor, equipment, material
capacity: float # units available per day
cost_per_unit: float = 0.0
skills: List[str] = field(default_factory=list)
@dataclass
class ResourceAssignment:
activity_id: str
resource_id: str
units: float # units required per day
start_day: int
end_day: int
@dataclass
class Activity:
id: str
name: str
duration: int
early_start: int
late_start: int
total_float: int
resource_requirements: Dict[str, float] = field(default_factory=dict)
is_critical: bool = False
@dataclass
class ResourceProfile:
resource_id: str
daily_usage: Dict[int, float] # day -> units used
peak_usage: float
average_usage: float
utilization_rate: float
@dataclass
class LevelingResult:
original_duration: int
new_duration: int
activities_shifted: List[Tuple[str, int, int]] # (id, old_start, new_start)
resource_profiles: Dict[str, ResourceProfile]
peak_reduction: Dict[str, float]
class ResourceOptimizer:
"""Optimize construction resource allocation."""
def __init__(self):
self.resources: Dict[str, Resource] = {}
self.activities: Dict[str, Activity] = {}
self.assignments: List[ResourceAssignment] = []
def add_resource(self, id: str, name: str, resource_type: str,
capacity: float, cost_per_unit: float = 0.0,
skills: List[str] = None) -> Resource:
"""Add resource to pool."""
resource = Resource(
id=id,
name=name,
resource_type=resource_type,
capacity=capacity,
cost_per_unit=cost_per_unit,
skills=skills or []
)
self.resources[id] = resource
return resource
def add_activity(self, id: str, name: str, duration: int,
early_start: int, late_start: int,
resource_requirements: Dict[str, float] = None,
is_critical: bool = False) -> Activity:
"""Add activity with resource requirements."""
activity = Activity(
id=id,
name=name,
duration=duration,
early_start=early_start,
late_start=late_start,
total_float=late_start - early_start,
resource_requirements=resource_requirements or {},
is_critical=is_critical
)
self.activities[id] = activity
# Create assignments
for res_id, units in activity.resource_requirements.items():
assignment = ResourceAssignment(
activity_id=id,
resource_id=res_id,
units=units,
start_day=early_start,
end_day=early_start + duration
)
self.assignments.append(assignment)
return activity
def calculate_resource_profile(self, resource_id: str,
activity_starts: Dict[str, int] = None) -> ResourceProfile:
"""Calculate daily resource usage profile."""
if resource_id not in self.resources:
raise ValueError(f"Resource {resource_id} not found")
resource = self.resources[resource_id]
daily_usage = defaultdict(float)
# Use provided starts or early starts
starts = activity_starts or {act.id: act.early_start for act in self.activities.values()}
for assignment in self.assignments:
if assignment.resource_id != resource_id:
continue
act_start = starts.get(assignment.activity_id, assignment.start_day)
act = self.activities[assignment.activity_id]
for day in range(act_start, act_start + act.duration):
daily_usage[day] += assignment.units
usage_values = list(daily_usage.values()) if daily_usage else [0]
project_duration = max(daily_usage.keys()) + 1 if daily_usage else 0
return ResourceProfile(
resource_id=resource_id,
daily_usage=dict(daily_usage),
peak_usage=max(usage_values),
average_usage=sum(usage_values) / len(usage_values) if usage_values else 0,
utilization_rate=sum(usage_values) / (project_duration * resource.capacity) if project_duration else 0
)
def identify_overallocations(self) -> Dict[str, List[Tuple[int, float]]]:
"""Identify days where resources are over-allocated."""
overallocations = {}
for resource in self.resources.values():
profile = self.calculate_resource_profile(resource.id)
over_days = [
(day, usage - resource.capacity)
for day, usage in profile.daily_usage.items()
if usage > resource.capacity
]
if over_days:
overallocations[resource.id] = over_days
return overallocations
def level_resources(self, resource_ids: List[str] = None,
allow_duration_extension: bool = True,
max_extension_days: int = 30) -> LevelingResult:
"""Level resources by shifting non-critical activities."""
resource_ids = resource_ids or list(self.resources.keys())
# Store original starts
original_starts = {act.id: act.early_start for act in self.activities.values()}
original_duration = max(act.early_start + act.duration for act in self.activities.values())
# Current activity starts (will be modified)
current_starts = dict(original_starts)
# Sort activities by float (most float = most flexibility)
sorted_activities = sorted(
[a for a in self.activities.values() if not a.is_critical],
key=lambda a: -a.total_float
)
activities_shifted = []
# Iteratively resolve overallocations
for _ in range(100): # Max iterations
overallocations = self._check_overallocations(current_starts, resource_ids)
if not overallocations:
break
# Find activity to shift
shifted = False
for act in sorted_activities:
if act.id in [o[0] for o in overallocations]:
# Try to shift this activity
new_start = self._find_valid_start(
act, current_starts, resource_ids,
allow_duration_extension, max_extension_days
)
if new_start is not None and new_start != current_starts[act.id]:
old_start = current_starts[act.id]
current_starts[act.id] = new_start
activities_shifted.append((act.id, old_start, new_start))
shifted = True
break