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