Resource Pool Optimizer
Overview
Manage and optimize shared resources (equipment, specialized labor, materials) across multiple construction projects. Identify conflicts, balance allocations, and maximize resource utilization while minimizing idle time and project delays.
Resource Pool Concept
┌─────────────────────────────────────────────────────────────────┐
│ RESOURCE POOL OPTIMIZATION │
├─────────────────────────────────────────────────────────────────┤
│ │
│ RESOURCE POOL PROJECTS │
│ ───────────── ──────── │
│ 🏗️ Tower Crane A ───────────→ Project 1 (Weeks 1-8) │
│ 🏗️ Tower Crane B ───────────→ Project 2 (Weeks 3-12) │
│ 🔧 Excavator Fleet ─────┬─────→ Project 1 (Weeks 1-4) │
│ └─────→ Project 3 (Weeks 5-10) │
│ 👷 Steel Crew A ───────────→ Project 2 (Weeks 6-15) │
│ 👷 Steel Crew B ─────┬─────→ Project 1 (Weeks 8-14) │
│ └─────→ Project 3 (Weeks 15-20) │
│ │
│ OPTIMIZATION GOALS: │
│ • Minimize idle time between projects │
│ • Avoid double-booking conflicts │
│ • Prioritize critical path activities │
│ • Balance utilization across resources │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Set
from datetime import datetime, timedelta
from enum import Enum
import heapq
class ResourceType(Enum):
EQUIPMENT = "equipment"
LABOR_CREW = "labor_crew"
MATERIAL = "material"
SPECIALTY = "specialty"
class AllocationStatus(Enum):
AVAILABLE = "available"
ALLOCATED = "allocated"
MAINTENANCE = "maintenance"
CONFLICT = "conflict"
class Priority(Enum):
CRITICAL = 1
HIGH = 2
NORMAL = 3
LOW = 4
@dataclass
class Resource:
id: str
name: str
resource_type: ResourceType
capacity: float = 1.0 # Can be split (e.g., crew size)
daily_cost: float = 0.0
home_location: str = ""
mobilization_days: int = 1
skills: List[str] = field(default_factory=list)
@dataclass
class ResourceRequest:
id: str
project_id: str
project_name: str
resource_type: ResourceType
required_skills: List[str]
start_date: datetime
end_date: datetime
quantity_needed: float = 1.0
priority: Priority = Priority.NORMAL
is_critical_path: bool = False
flexibility_days: int = 0 # Can shift by this many days
notes: str = ""
@dataclass
class Allocation:
id: str
resource_id: str
request_id: str
project_id: str
start_date: datetime
end_date: datetime
quantity: float
status: AllocationStatus = AllocationStatus.ALLOCATED
@dataclass
class Conflict:
resource_id: str
resource_name: str
date: datetime
requests: List[ResourceRequest]
total_demand: float
available: float
resolution_options: List[str]
@dataclass
class UtilizationReport:
resource_id: str
resource_name: str
period_start: datetime
period_end: datetime
total_days: int
allocated_days: int
utilization_pct: float
idle_periods: List[Tuple[datetime, datetime]]
cost: float
class ResourcePoolOptimizer:
"""Optimize shared resources across projects."""
def __init__(self, pool_name: str):
self.pool_name = pool_name
self.resources: Dict[str, Resource] = {}
self.requests: Dict[str, ResourceRequest] = {}
self.allocations: Dict[str, Allocation] = {}
def add_resource(self, id: str, name: str, resource_type: ResourceType,
capacity: float = 1.0, daily_cost: float = 0.0,
skills: List[str] = None) -> Resource:
"""Add resource to pool."""
resource = Resource(
id=id,
name=name,
resource_type=resource_type,
capacity=capacity,
daily_cost=daily_cost,
skills=skills or []
)
self.resources[id] = resource
return resource
def add_request(self, project_id: str, project_name: str,
resource_type: ResourceType, start_date: datetime,
end_date: datetime, quantity: float = 1.0,
priority: Priority = Priority.NORMAL,
required_skills: List[str] = None,
is_critical_path: bool = False,
flexibility_days: int = 0) -> ResourceRequest:
"""Add resource request from project."""
request_id = f"REQ-{project_id}-{len(self.requests)+1:03d}"
request = ResourceRequest(
id=request_id,
project_id=project_id,
project_name=project_name,
resource_type=resource_type,
required_skills=required_skills or [],
start_date=start_date,
end_date=end_date,
quantity_needed=quantity,
priority=priority,
is_critical_path=is_critical_path,
flexibility_days=flexibility_days
)
self.requests[request_id] = request
return request
def find_available_resources(self, request: ResourceRequest) -> List[Tuple[Resource, float]]:
"""Find resources that can fulfill request."""
available = []
for resource in self.resources.values():
# Check type match
if resource.resource_type != request.resource_type:
continue
# Check skills match
if request.required_skills:
if not all(skill in resource.skills for skill in request.required_skills):
continue
# Check availability
available_qty = self._get_availability(
resource.id, request.start_date, request.end_date
)
if available_qty > 0:
available.append((resource, available_qty))
return sorted(available, key=lambda x: -x[1])
def _get_availability(self, resource_id: str, start: datetime, end: datetime) -> float:
"""Get available capacity for resource in period."""
resource = self.resources.get(resource_id)
if not resource:
return 0
# Find overlapping allocations
allocated = 0
for alloc in self.allocations.values():
if alloc.resource_id != resource_id:
continue
# Check overlap
if alloc.start_date < end and alloc.end_date > start:
allocated = max(allocated, alloc.quantity)
return resource.capacity - allocated
def allocate(self, request_id: str, resource_id: str,
quantity: float = None) -> Allocation:
"""Allocate resource to request."""
if request_id not in self.requests:
raise ValueError(f"Request {request_id} not found")
if resource_id not in self.resources:
raise ValueError(f"Resource {resource_id} not found")
request = self.requests[request_id]
resource = self.resources[resource_id]
if quantity is None:
quantity = min(request.quantity_needed, resource.capacity)
# Check availability
available = self._get_availability(
resource_id, request.start_date, request.end_date
)
if available < quantity:
raise ValueError(f"Insufficient capacity. Available: {available}, Requested: {quantity}")
alloc_id = f"ALLOC-{len(self.allocations)+1:04d}"
allocation = Allocation(
id=alloc_id,
resource_id=resource_id,
request_id=request_id,
project_id=request.project_id,
start_date=request.start_date,
end_date=request.end_date,
quantity=quantity
)
self.allocations[alloc_id] = allocation
return allocation
def auto_allocate(self) -> Dict[str, List[Allocation]]:
"""Automatically allocate resources using priority-based algorithm."""
results = {"allocated": [], "unallocated": [], "conflicts": []}
# Sort requests by priority and critical path
sorted_requests = sorted(
self.requests.values(),
key=lambda r: (r.priority.value, not r.is_critical_path, r.start_date)
)
for request in sorted_requests:
# Check if already allocated
existing = [a for a in self.allocations.values()
if a.request_id == request.id]
if existing:
continue
# Find available resources
available = self.find_available_resources(request)
if not available:
results["unallocated"].append(request)
continue
# Allocate best match
resource, qty = available[0]
try:
allocation = self.allocate(request.id, resource.id,
min(request.quantity_needed, qty))
results["allocated"].append(allocation)
except ValueError:
results["conflicts"].append(request)
return results
def detect_conflicts(self) -> List[Conflict]:
"""Detect resource conflicts across all requests."""
conflicts = []
# Group requests by resource type
by_type: Dict[ResourceType, List[ResourceRequest]] = {}
for req in self.requests.values():
if req.resource_type not in by_type:
by_type[req.resource_type] = []
by_type[req.resource_type].append(req)
# Check each resource type
for resource_type, requests in by_type.items():
# Get all resources of this type
resources = [r for r in self.resources.values()
if r.resource_type == resource_type]
total_capacity = sum(r.capacity for r in resources)
# Check each day for conflicts
all_dates = set()
for req in requests:
current = req.start_date
while current <= req.end_date:
all_dates.add(current)
current += timedelta(days=1)
for date in sorted(all_dates):
# Sum demand for this date
day_requests = [r for r in requests
if r.start_date <= date <= r.end_date]
total_demand = sum(r.quantity_needed for r in day_requests)
if total_demand > total_capacity:
# Generate resolution options
options = []
for req in sorted(day_requests, key=lambda x: x.priority.value, reverse=True):
if req.flexibility_days > 0:
options.append(f"Shift {req.project_name} by {req.flexibility_days} days")
options.append("Add additional resource")
options.append("Extend work hours")
conflict = Conflict(
resource_id=resource_type.value,
resource_name=resource_type.value,
date=date,
requests=day_requests,
total_demand=total_demand,
available=total_capacity,
resolution_options=options
)
conflicts.append(conflict)
return conflicts
def calculate_utilization(self, start_date: datetime,
end_date: datetime) -> List[UtilizationReport]:
"""Calculate utilization for all resources."""
reports = []
total_days = (end_date - start_date).days
for resource in self.resources.values():
# Find allocations in period
allocs = [a for a in self.allocations.values()
if a.resource_id == resource.id
and a.start_date < end_date
and a.end_date > start_date]
# Calculate allocated days
allocated_dates = set()
for alloc in allocs:
current = max(alloc.start_date, start_date)
while current < min(alloc.end_date, end_date):
allocated_dates.add(current)
current += timedelta(days=1)
allocated_days = len(allocated_dates)
utilization = (allocated_days / total_days * 100) if total_days > 0 else 0
# Find idle periods
idle_periods = []
all_dates = set()
current = start_date
while current < end_date:
all_dates.add(current)
current += timedelta(days=1)
idle_dates = sorted(all_dates - allocated_dates)
if idle_dates:
# Group consecutive idle dates
period_start = idle_dates[0]
for i, date in enumerate(idle_dates[1:], 1):
if (date - idle_dates[i-1]).days > 1:
idle_periods.append((period_start, idle_dates[i-1]))
period_start = date
idle_periods.append((period_start, idle_dates[-1]))
cost = allocated_days * resource.daily_cost
reports.append(UtilizationReport(
resource_id=resource.id,
resource_name=resource.name,
period_start=start_date,
period_end=end_date,
total_days=total_days,
allocated_days=allocated_days,
utilization_pct=utilization,
idle_periods=idle_periods,
cost=cost
))
return sorted(reports, key=lambda x: -x.utilization_pct)
def suggest_optimization(self) -> List[Dict]:
"""Suggest optimizations for resource allocation."""
suggestions = []
# Find underutilized resources
util = self.calculate_utilization(
datetime.now(),
datetime.now() + timedelta(days=90)
)
for report in util:
if report.utilization_pct < 50:
suggestions.append({
"type": "underutilization",
"resource": report.resource_name,
"utilization": report.utilization_pct,
"suggestion": f"Consider reassigning or releasing {report.resource_name}"
})
# Find conflicts
conflicts = self.detect_conflicts()
for conflict in conflicts[:5]:
suggestions.append({
"type": "conflict",
"date": conflict.date,
"demand": conflict.total_demand,
"available": conflict.available,
"suggestion": conflict.resolution_options[0] if conflict.resolution_options else "Review allocation"
})
return suggestions
def generate_report(self) -> str:
"""Generate resource pool report."""
lines = [
"# Resource Pool Optimization Report",
"",
f"**Pool:** {self.pool_name}",
f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
"",
"## Resource Inventory",
"",
"| Resource | Type | Capacity | Daily Cost |",
"|----------|------|----------|------------|"
]
for r in self.resources.values():
lines.append(
f"| {r.name} | {r.resource_type.value} | {r.capacity} | ${r.daily_cost:,.0f} |"
)
lines.extend([
"",
"## Current Allocations",
"",
"| Resource | Project | Start | End | Qty |",
"|----------|---------|-------|-----|-----|"
])
for alloc in sorted(self.allocations.values(), key=lambda x: x.start_date):
resource = self.resources.get(alloc.resource_id)
lines.append(
f"| {resource.name if resource else alloc.resource_id} | "
f"{alloc.project_id} | {alloc.start_date.strftime('%Y-%m-%d')} | "
f"{alloc.end_date.strftime('%Y-%m-%d')} | {alloc.quantity} |"
)
# Utilization
util = self.calculate_utilization(
datetime.now(),
datetime.now() + timedelta(days=90)
)
lines.extend([
"",
"## 90-Day Utilization Forecast",
"",
"| Resource | Utilization | Idle Periods |",
"|----------|-------------|--------------|"
])
for report in util:
idle_str = f"{len(report.idle_periods)} gaps" if report.idle_periods else "None"
lines.append(
f"| {report.resource_name} | {report.utilization_pct:.0f}% | {idle_str} |"
)
# Conflicts
conflicts = self.detect_conflicts()
if conflicts:
lines.extend([
"",
f"## Conflicts Detected ({len(conflicts)})",
"",
"| Date | Resource | Demand | Available |",
"|------|----------|--------|-----------|"
])
for c in conflicts[:10]:
lines.append(
f"| {c.date.strftime('%Y-%m-%d')} | {c.resource_name} | "
f"{c.total_demand} | {c.available} |"
)
return "\n".join(lines)Quick Start
from datetime import datetime, timedelta
# Initialize optimizer
optimizer = ResourcePoolOptimizer("Regional Equipment Pool")
# Add resources
optimizer.add_resource("CR-001", "Tower Crane Alpha", ResourceType.EQUIPMENT,
capacity=1, daily_cost=2500)
optimizer.add_resource("CR-002", "Tower Crane Beta", ResourceType.EQUIPMENT,
capacity=1, daily_cost=2500)
optimizer.add_resource("EX-001", "Excavator Fleet", ResourceType.EQUIPMENT,
capacity=3, daily_cost=1500)
optimizer.add_resource("SC-001", "Steel Crew A", ResourceType.LABOR_CREW,
capacity=1, daily_cost=8000, skills=["structural", "welding"])
# Add requests from projects
optimizer.add_request(
"PRJ-001", "Downtown Tower",
ResourceType.EQUIPMENT,
start_date=datetime(2025, 2, 1),
end_date=datetime(2025, 6, 30),
quantity=1,
priority=Priority.HIGH,
is_critical_path=True
)
optimizer.add_request(
"PRJ-002", "Hospital Wing",
ResourceType.EQUIPMENT,
start_date=datetime(2025, 3, 1),
end_date=datetime(2025, 8, 31),
quantity=1,
priority=Priority.NORMAL,
flexibility_days=14
)
# Auto-allocate resources
results = optimizer.auto_allocate()
print(f"Allocated: {len(results['allocated'])}")
print(f"Unallocated: {len(results['unallocated'])}")
# Detect conflicts
conflicts = optimizer.detect_conflicts()
print(f"Conflicts: {len(conflicts)}")
# Get optimization suggestions
suggestions = optimizer.suggest_optimization()
for s in suggestions[:5]:
print(f"{s['type']}: {s['suggestion']}")
# Generate report
print(optimizer.generate_report())Requirements
pip install (no external dependencies)