Resource Pool Optimizer skill

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)