resource-leveler

建設向けリソース平準化

概要

建設スケジュール全体でリソース配分を最適化します。労働力と設備を平準化してピークを回避し、作業負荷のバランスを取り、コストを削減しながらプロジェクトの納期を守ります。

ビジネス上のメリット

リソースの平準化には、次のメリットがあります。

• コスト削減：残業や待機時間を回避
• 労働力の安定：作業員の人数を一定に維持
• 設備の最適化：レンタル費用を削減
• 現実的なスケジュール：実現可能なリソース計画

技術的な実装

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime, date, timedelta
import pandas as pd
import numpy as np
from collections import defaultdict

@dataclass
class Resource:
    id: str
    name: str
    resource_type: str  # labor, equipment, material
    max_units: float
    cost_per_unit: float
    unit: str  # hours, days, each

@dataclass
class ResourceAssignment:
    task_id: str
    resource_id: str
    units: float
    start_date: date
    end_date: date

@dataclass
class Task:
    id: str
    name: str
    duration: int  # days
    start_date: date
    end_date: date
    predecessors: List[str]
    total_float: int
    is_critical: bool
    resource_assignments: List[ResourceAssignment] = field(default_factory=list)

@dataclass
class LevelingResult:
    success: bool
    original_end_date: date
    leveled_end_date: date
    tasks_moved: int
    peak_reduction: Dict[str, float]
    warnings: List[str]

class ConstructionResourceLeveler:
    """Level resources across construction schedules."""

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

    def add_resource(self, resource: Resource):
        """Add a resource to the pool."""
        self.resources[resource.id] = resource

    def add_task(self, task: Task):
        """Add a task to the schedule."""
        self.tasks[task.id] = task

    def add_assignment(self, assignment: ResourceAssignment):
        """Assign a resource to a task."""
        self.assignments.append(assignment)
        if assignment.task_id in self.tasks:
            self.tasks[assignment.task_id].resource_assignments.append(assignment)

    def calculate_resource_usage(self, start_date: date = None,
                                  end_date: date = None) -> pd.DataFrame:
        """Calculate daily resource usage."""
        if not self.assignments:
            return pd.DataFrame()

        # Determine date range
        if start_date is None:
            start_date = min(a.start_date for a in self.assignments)
        if end_date is None:
            end_date = max(a.end_date for a in self.assignments)

        # Create date range
        dates = pd.date_range(start_date, end_date, freq='D')

        # Initialize usage matrix
        usage = {r_id: [0.0] * len(dates) for r_id in self.resources}

        # Fill in usage
        for assignment in self.assignments:
            if assignment.resource_id in usage:
                for i, d in enumerate(dates):
                    if assignment.start_date <= d.date() <= assignment.end_date:
                        usage[assignment.resource_id][i] += assignment.units

        df = pd.DataFrame(usage, index=dates)
        df.index.name = 'date'

        return df

    def identify_overallocations(self) -> List[Dict]:
        """Identify resource overallocations."""
        usage = self.calculate_resource_usage()
        overallocations = []

        for resource_id, resource in self.resources.items():
            if resource_id in usage.columns:
                daily_usage = usage[resource_id]
                over_days = daily_usage[daily_usage > resource.max_units]

                if len(over_days) > 0:
                    overallocations.append({
                        'resource_id': resource_id,
                        'resource_name': resource.name,
                        'max_units': resource.max_units,
                        'peak_usage': daily_usage.max(),
                        'over_by': daily_usage.max() - resource.max_units,
                        'days_overallocated': len(over_days),
                        'first_overallocation': over_days.index[0].date(),
                        'worst_day': daily_usage.idxmax().date()
                    })

        return overallocations

    def level_resources(self, method: str = 'float_priority',
                        protect_critical_path: bool = True,
                        max_extension: int = 30) -> LevelingResult:
        """Level resources to resolve overallocations."""

        original_end = max(t.end_date for t in self.tasks.values())
        tasks_moved = 0
        warnings = []

        # Get initial overallocations
        initial_over = self.identify_overallocations()
        if not initial_over:
            return LevelingResult(
                success=True,
                original_end_date=original_end,
                leveled_end_date=original_end,
                tasks_moved=0,
                peak_reduction={},
                warnings=["No overallocations found"]
            )

        # Track peak usage before
        usage_before = self.calculate_resource_usage()
        peaks_before = {r: usage_before[r].max() for r in usage_before.columns}

        # Leveling loop
        iteration = 0
        max_iterations = len(self.tasks) * 2

        while iteration < max_iterations:
            iteration += 1
            overallocations = self.identify_overallocations()

            if not overallocations:
                break

            # Find task to move
            moved = False
            for over in overallocations:
                resource_id = over['resource_id']
                worst_day = over['worst_day']

                # Find tasks using this resource on worst day
                candidates = self._find_movable_tasks(
                    resource_id, worst_day, protect_critical_path
                )

                if candidates:
                    # Sort by priority (lowest float first to preserve options)
                    candidates.sort(key=lambda t: -t.total_float)
                    task_to_move = candidates[0]

                    # Calculate new dates
                    new_start, new_end = self._calculate_shift(
                        task_to_move, resource_id, max_extension
                    )

                    if new_start:
                        self._shift_task(task_to_move.id, new_start, new_end)
                        tasks_moved += 1
                        moved = True
                        break

            if not moved:
                warnings.append("Could not resolve all overallocations")
                break

        # Calculate results
        usage_after = self.calculate_resource_usage()
        peaks_after = {r: usage_after[r].max() for r in usage_after.columns}

        peak_reduction = {}
        for r in peaks_before:
            if r in peaks_after:
                reduction = (peaks_before[r] - peaks_after[r]) / peaks_before[r] * 100
                peak_reduction[r] = reduction

        leveled_end = max(t.end_date for t in self.tasks.values())

        if leveled_end > original_end + timedelta(days=max_extension):
            warnings.append(f"Project extended beyond max allowed ({max_extension} days)")

        remaining_over = self.identify_overallocations()

        return LevelingResult(
            success=len(remaining_over) == 0,
            original_end_date=original_end,
            leveled_end_date=leveled_end,
            tasks_moved=tasks_moved,
            peak_reduction=peak_reduction,
            warnings=warnings
        )

    def _find_movable_tasks(self, resource_id: str, on_date: date,
                            protect_critical: bool) -> List[Task]:
        """Find tasks that can be moved to reduce overallocation."""
        candidates = []

        for task in self.tasks.values():
            # Check if task uses this resource on this date
            uses_resource = any(
                a.resource_id == resource_id and
                a.start_date <= on_date <= a.end_date
                for a in task.resource_assignments
            )

            if not uses_resource:
                continue

            # Check if critical path protected
            if protect_critical and task.is_critical:
                continue

            # Check if has float
            if task.total_float > 0:
                candidates.append(task)

        return candidates

    def _calculate_shift(self, task: Task, resource_id: str,
                         max_extension: int) -> Tuple[date, date]:
        """Calculate optimal shift for a task."""
        resource = self.resources[resource_id]

        # Try shifting forward
        for days in range(1, min(task.total_float + 1, max_extension + 1)):
            new_start = task.start_date + timedelta(days=days)
            new_end = task.end_date + timedelta(days=days)

            # Check if this resolves overallocation
            temp_usage = self._calculate_usage_if_moved(task.id, new_start, new_end)

            if temp_usage.get(resource_id, 0) <= resource.max_units:
                return new_start, new_end

        return None, None

    def _calculate_usage_if_moved(self, task_id: str, new_start: date,
                                   new_end: date) -> Dict[str, float]:
        """Calculate resource usage if task were moved."""
        # Simplified: calculate peak on affected dates
        usage = defaultdict(float)

        for assignment in self.assignments:
            if assignment.task_id == task_id:
                # Use new dates
                for d in pd.date_range(new_start, new_end):
                    usage[assignment.resource_id] = max(
                        usage[assignment.resource_id],
                        assignment.units
                    )
            else:
                # Use existing dates
                for d in pd.date_range(assignment.start_date, assignment.end_date):
                    usage[assignment.resource_id] = max(
                        usage[assignment.resource_id],
                        assignment.units
                    )

        return dict(usage)

    def _shift_task(self, task_id: str, new_start: date, new_end: date):
        """Shift a task to new dates."""
        task = self.tasks[task_id]
        delta = new_start - task.start_date

        # Update task
        task.start_date = new_start
        task.end_date = new_end

        # Update assignments
        for assignment in self.assignments:
            if assignment.task_id == task_id:
                assignment.start_date += delta
                assignment.end_date += delta

    def optimize_crew_size(self, resource_id: str,
                            target_utilization: float = 0.85) -> Dict:
        """Recommend optimal crew size for a resource."""
        usage = self.calculate_resource_usage()

        if resource_id not in usage.columns:
            return None

        daily_usage = usage[resource_id]
        resource = self.resources[resource_id]

        # Calculate statistics
        peak = daily_usage.max()
        avg = daily_usage.mean()
        working_days = (daily_usage > 0).sum()

        # Current utilization
        current_util = avg / resource.max_units if resource.max_units > 0 else 0

        # Optimal size for target utilization
        optimal_size = avg / target_utilization

        return {
            'resource_id': resource_id,
            'current_max_units': resource.max_units,
            'peak_usage': peak,
            'average_usage': avg,
            'working_days': int(working_days),
            'current_utilization': current_util,
            'recommended_max_units': round(optimal_size, 1),
            'potential_savings': (resource.max_units - optimal_size) * resource.cost_per_unit * working_days
        }

    def generate_histogram(self, resource_id: str) -> pd.DataFrame:
        """Generate resource histogram data."""
        usage = self.calculate_resource_usage()

        if resource_id not in usage.columns:
            return pd.DataFrame()

        resource = self.resources[resource_id]

        df = pd.DataFrame({
            'date': usage.index,
            'usage': usage[resource_id].values,
            'capacity': resource.max_units,
            'overallocated': usage[resource_id].values > resource.max_units
        })

        return df

    def generate_report(self) -> str:
        """Generate resource leveling report."""
        lines = ["# Resource Leveling Report", ""]
        lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        lines.append(f"**Resources:** {len(self.resources)}")
        lines.append(f"**Tasks:** {len(self.tasks)}")
        lines.append("")

        # Overallocations
        overallocations = self.identify_overallocations()
        if overallocations:
            lines.append("## Overallocations Found")
            for over in overallocations:
                lines.append(f"\n### {over['resource_name']}")
                lines.append(f"- **Max Units:** {over['max_units']}")
                lines.append(f"- **Peak Usage:** {over['peak_usage']}")
                lines.append(f"- **Days Overallocated:** {over['days_overallocated']}")
                lines.append(f"- **Worst Day:** {over['worst_day']}")
        else:
            lines.append("## No Overallocations")
            lines.append("All resources are within capacity.")

        # Resource utilization
        lines.append("\n## Resource Utilization")
        for resource_id in self.resources:
            opt = self.optimize_crew_size(resource_id)
            if opt:
                lines.append(f"\n### {self.resources[resource_id].name}")
                lines.append(f"- **Utilization:** {opt['current_utilization']:.1%}")
                lines.append(f"- **Peak:** {opt['peak_usage']:.1f}")
                lines.append(f"- **Average:** {opt['average_usage']:.1f}")

        return "\n".join(lines)

クイックスタート

from datetime import date

# Initialize leveler
leveler = ConstructionResourceLeveler()

# Add resources
leveler.add_resource(Resource(
    id="CARP",
    name="Carpenters",
    resource_type="labor",
    max_units=10,
    cost_per_unit=75,
    unit="hours"
))

# Add tasks
leveler.add_task(Task(
    id="T1",
    name="Frame Level 1",
    duration=10,
    start_date=date(2026, 3, 1),
    end_date=date(2026, 3, 14),
    predecessors=[],
    total_float=5,
    is_critical=False
))

# Add assignments
leveler.add_assignment(ResourceAssignment(
    task_id="T1",
    resource_id="CARP",
    units=8,
    start_date=date(2026, 3, 1),
    end_date=date(2026, 3, 14)
))

# Check overallocations
overallocations = leveler.identify_overallocations()
for over in overallocations:
    print(f"{over['resource_name']}: {over['peak_usage']} vs {over['max_units']} max")

# Level resources
result = leveler.level_resources(protect_critical_path=True)
print(f"Tasks moved: {result.tasks_moved}")
print(f"End date change: {result.original_end_date} -> {result.leveled_end_date}")

# Generate report
print(leveler.generate_report())

依存関係

pip install pandas numpy