| name | resource-pool-optimizer |
| description | Optimize shared resource pools across multiple construction projects. Balance resource allocation, minimize conflicts, and maximize utilization across the portfolio. |
| homepage | https://datadrivenconstruction.io |
| metadata | {"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}} |
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
daily_cost: float = 0.0
home_location: str = ""
mobilization_days: int =
skills: [] = field(default_factory=)
:
:
project_id:
project_name:
resource_type: ResourceType
required_skills: []
start_date: datetime
end_date: datetime
quantity_needed: =
priority: Priority = Priority.NORMAL
is_critical_path: =
flexibility_days: =
notes: =
:
:
resource_id:
request_id:
project_id:
start_date: datetime
end_date: datetime
quantity:
status: AllocationStatus = AllocationStatus.ALLOCATED
:
resource_id:
resource_name:
date: datetime
requests: [ResourceRequest]
total_demand:
available:
resolution_options: []
:
resource_id:
resource_name:
period_start: datetime
period_end: datetime
total_days:
allocated_days:
utilization_pct:
idle_periods: [[datetime, datetime]]
cost:
:
():
.pool_name = pool_name
.resources: [, Resource] = {}
.requests: [, ResourceRequest] = {}
.allocations: [, Allocation] = {}
() -> Resource:
resource = Resource(
=,
name=name,
resource_type=resource_type,
capacity=capacity,
daily_cost=daily_cost,
skills=skills []
)
.resources[] = resource
resource
() -> ResourceRequest:
request_id =
request = ResourceRequest(
=request_id,
project_id=project_id,
project_name=project_name,
resource_type=resource_type,
required_skills=required_skills [],
start_date=start_date,
end_date=end_date,
quantity_needed=quantity,
priority=priority,
is_critical_path=is_critical_path,
flexibility_days=flexibility_days
)
.requests[request_id] = request
request
() -> [[Resource, ]]:
available = []
resource .resources.values():
resource.resource_type != request.resource_type:
request.required_skills:
(skill resource.skills skill request.required_skills):
available_qty = ._get_availability(
resource., request.start_date, request.end_date
)
available_qty > :
available.append((resource, available_qty))
(available, key= x: -x[])
() -> :
resource = .resources.get(resource_id)
resource:
allocated =
alloc .allocations.values():
alloc.resource_id != resource_id:
alloc.start_date < end alloc.end_date > start:
allocated = (allocated, alloc.quantity)
resource.capacity - allocated
() -> Allocation:
request_id .requests:
ValueError()
resource_id .resources:
ValueError()
request = .requests[request_id]
resource = .resources[resource_id]
quantity :
quantity = (request.quantity_needed, resource.capacity)
available = ._get_availability(
resource_id, request.start_date, request.end_date
)
available < quantity:
ValueError()
alloc_id =
allocation = Allocation(
=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
)
.allocations[alloc_id] = allocation
allocation
() -> [, [Allocation]]:
results = {: [], : [], : []}
sorted_requests = (
.requests.values(),
key= r: (r.priority.value, r.is_critical_path, r.start_date)
)
request sorted_requests:
existing = [a a .allocations.values()
a.request_id == request.]
existing:
available = .find_available_resources(request)
available:
results[].append(request)
resource, qty = available[]
:
allocation = .allocate(request., resource.,
(request.quantity_needed, qty))
results[].append(allocation)
ValueError:
results[].append(request)
results
() -> [Conflict]:
conflicts = []
by_type: [ResourceType, [ResourceRequest]] = {}
req .requests.values():
req.resource_type by_type:
by_type[req.resource_type] = []
by_type[req.resource_type].append(req)
resource_type, requests by_type.items():
resources = [r r .resources.values()
r.resource_type == resource_type]
total_capacity = (r.capacity r resources)
all_dates = ()
req requests:
current = req.start_date
current <= req.end_date:
all_dates.add(current)
current += timedelta(days=)
date (all_dates):
day_requests = [r r requests
r.start_date <= date <= r.end_date]
total_demand = (r.quantity_needed r day_requests)
total_demand > total_capacity:
options = []
req (day_requests, key= x: x.priority.value, reverse=):
req.flexibility_days > :
options.append()
options.append()
options.append()
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)
conflicts
() -> [UtilizationReport]:
reports = []
total_days = (end_date - start_date).days
resource .resources.values():
allocs = [a a .allocations.values()
a.resource_id == resource.
a.start_date < end_date
a.end_date > start_date]
allocated_dates = ()
alloc allocs:
current = (alloc.start_date, start_date)
current < (alloc.end_date, end_date):
allocated_dates.add(current)
current += timedelta(days=)
allocated_days = (allocated_dates)
utilization = (allocated_days / total_days * ) total_days >
idle_periods = []
all_dates = ()
current = start_date
current < end_date:
all_dates.add(current)
current += timedelta(days=)
idle_dates = (all_dates - allocated_dates)
idle_dates:
period_start = idle_dates[]
i, date (idle_dates[:], ):
(date - idle_dates[i-]).days > :
idle_periods.append((period_start, idle_dates[i-]))
period_start = date
idle_periods.append((period_start, idle_dates[-]))
cost = allocated_days * resource.daily_cost
reports.append(UtilizationReport(
resource_id=resource.,
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
))
(reports, key= x: -x.utilization_pct)
() -> []:
suggestions = []
util = .calculate_utilization(
datetime.now(),
datetime.now() + timedelta(days=)
)
report util:
report.utilization_pct < :
suggestions.append({
: ,
: report.resource_name,
: report.utilization_pct,
:
})
conflicts = .detect_conflicts()
conflict conflicts[:]:
suggestions.append({
: ,
: conflict.date,
: conflict.total_demand,
: conflict.available,
: conflict.resolution_options[] conflict.resolution_options
})
suggestions
() -> :
lines = [
,
,
,
,
,
,
,
,
]
r .resources.values():
lines.append(
)
lines.extend([
,
,
,
,
])
alloc (.allocations.values(), key= x: x.start_date):
resource = .resources.get(alloc.resource_id)
lines.append(
)
util = .calculate_utilization(
datetime.now(),
datetime.now() + timedelta(days=)
)
lines.extend([
,
,
,
,
])
report util:
idle_str = report.idle_periods
lines.append(
)
conflicts = .detect_conflicts()
conflicts:
lines.extend([
,
,
,
,
])
c conflicts[:]:
lines.append(
)
.join(lines)
Quick Start
from datetime import datetime, timedelta
optimizer = ResourcePoolOptimizer("Regional Equipment Pool")
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"])
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=
)
results = optimizer.auto_allocate()
()
()
conflicts = optimizer.detect_conflicts()
()
suggestions = optimizer.suggest_optimization()
s suggestions[:]:
()
(optimizer.generate_report())
Requirements
pip install (no external dependencies)
