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.allocati