resource-allocation-optimizer

What this skill does

# Resource Allocation Optimizer

## Overview

Optimize resource allocation in construction schedules. Level workforce and equipment utilization, resolve over-allocations, and balance workload across the project duration.

> "Resource leveling reduces peak demand by 30% and improves productivity" — DDC Community

## Resource Leveling Concept

```
Before Leveling:                    After Leveling:
Workers                             Workers
  20│    ████                         15│  ████████████
  15│  ████████                        10│████████████████
  10│████████████                       5│████████████████████
   5│██████████████████                 0└──────────────────────
   0└────────────────────                  Week 1  2  3  4  5  6
      Week 1  2  3  4  5
                                       Peak reduced, duration extended
```

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from collections import defaultdict
import heapq

@dataclass
class Resource:
    id: str
    name: str
    resource_type: str  # labor, equipment, material
    capacity: float  # units available per day
    cost_per_unit: float = 0.0
    skills: List[str] = field(default_factory=list)

@dataclass
class ResourceAssignment:
    activity_id: str
    resource_id: str
    units: float  # units required per day
    start_day: int
    end_day: int

@dataclass
class Activity:
    id: str
    name: str
    duration: int
    early_start: int
    late_start: int
    total_float: int
    resource_requirements: Dict[str, float] = field(default_factory=dict)
    is_critical: bool = False

@dataclass
class ResourceProfile:
    resource_id: str
    daily_usage: Dict[int, float]  # day -> units used
    peak_usage: float
    average_usage: float
    utilization_rate: float

@dataclass
class LevelingResult:
    original_duration: int
    new_duration: int
    activities_shifted: List[Tuple[str, int, int]]  # (id, old_start, new_start)
    resource_profiles: Dict[str, ResourceProfile]
    peak_reduction: Dict[str, float]

class ResourceOptimizer:
    """Optimize construction resource allocation."""

    def __init__(self):
        self.resources: Dict[str, Resource] = {}
        self.activities: Dict[str, Activity] = {}
        self.assignments: List[ResourceAssignment] = []

    def add_resource(self, id: str, name: str, resource_type: str,
                    capacity: float, cost_per_unit: float = 0.0,
                    skills: List[str] = None) -> Resource:
        """Add resource to pool."""
        resource = Resource(
            id=id,
            name=name,
            resource_type=resource_type,
            capacity=capacity,
            cost_per_unit=cost_per_unit,
            skills=skills or []
        )
        self.resources[id] = resource
        return resource

    def add_activity(self, id: str, name: str, duration: int,
                    early_start: int, late_start: int,
                    resource_requirements: Dict[str, float] = None,
                    is_critical: bool = False) -> Activity:
        """Add activity with resource requirements."""
        activity = Activity(
            id=id,
            name=name,
            duration=duration,
            early_start=early_start,
            late_start=late_start,
            total_float=late_start - early_start,
            resource_requirements=resource_requirements or {},
            is_critical=is_critical
        )
        self.activities[id] = activity

        # Create assignments
        for res_id, units in activity.resource_requirements.items():
            assignment = ResourceAssignment(
                activity_id=id,
                resource_id=res_id,
                units=units,
                start_day=early_start,
                end_day=early_start + duration
            )
            self.assignments.append(assignment)

        return activity

    def calculate_resource_profile(self, resource_id: str,
                                  activity_starts: Dict[str, int] = None) -> ResourceProfile:
        """Calculate daily resource usage profile."""
        if resource_id not in self.resources:
            raise ValueError(f"Resource {resource_id} not found")

        resource = self.resources[resource_id]
        daily_usage = defaultdict(float)

        # Use provided starts or early starts
        starts = activity_starts or {act.id: act.early_start for act in self.activities.values()}

        for assignment in self.assignments:
            if assignment.resource_id != resource_id:
                continue

            act_start = starts.get(assignment.activity_id, assignment.start_day)
            act = self.activities[assignment.activity_id]

            for day in range(act_start, act_start + act.duration):
                daily_usage[day] += assignment.units

        usage_values = list(daily_usage.values()) if daily_usage else [0]
        project_duration = max(daily_usage.keys()) + 1 if daily_usage else 0

        return ResourceProfile(
            resource_id=resource_id,
            daily_usage=dict(daily_usage),
            peak_usage=max(usage_values),
            average_usage=sum(usage_values) / len(usage_values) if usage_values else 0,
            utilization_rate=sum(usage_values) / (project_duration * resource.capacity) if project_duration else 0
        )

    def identify_overallocations(self) -> Dict[str, List[Tuple[int, float]]]:
        """Identify days where resources are over-allocated."""
        overallocations = {}

        for resource in self.resources.values():
            profile = self.calculate_resource_profile(resource.id)
            over_days = [
                (day, usage - resource.capacity)
                for day, usage in profile.daily_usage.items()
                if usage > resource.capacity
            ]
            if over_days:
                overallocations[resource.id] = over_days

        return overallocations

    def level_resources(self, resource_ids: List[str] = None,
                       allow_duration_extension: bool = True,
                       max_extension_days: int = 30) -> LevelingResult:
        """Level resources by shifting non-critical activities."""
        resource_ids = resource_ids or list(self.resources.keys())

        # Store original starts
        original_starts = {act.id: act.early_start for act in self.activities.values()}
        original_duration = max(act.early_start + act.duration for act in self.activities.values())

        # Current activity starts (will be modified)
        current_starts = dict(original_starts)

        # Sort activities by float (most float = most flexibility)
        sorted_activities = sorted(
            [a for a in self.activities.values() if not a.is_critical],
            key=lambda a: -a.total_float
        )

        activities_shifted = []

        # Iteratively resolve overallocations
        for _ in range(100):  # Max iterations
            overallocations = self._check_overallocations(current_starts, resource_ids)

            if not overallocations:
                break

            # Find activity to shift
            shifted = False
            for act in sorted_activities:
                if act.id in [o[0] for o in overallocations]:
                    # Try to shift this activity
                    new_start = self._find_valid_start(
                        act, current_starts, resource_ids,
                        allow_duration_extension, max_extension_days
                    )

                    if new_start is not None and new_start != current_starts[act.id]:
                        old_start = current_starts[act.id]
                        current_starts[act.id] = new_start
                        activities_shifted.append((act.id, old_start, new_start))
                        shifted = True
                        break