cwicr-labor-scheduler

# CWICR Labor Scheduler

## Business Case

### Problem Statement
Project managers need to plan labor allocation:
- How many workers per day?
- What skills are needed when?
- How to balance workload across project phases?
- How to avoid resource conflicts?

### Solution
Data-driven labor scheduling using CWICR labor norms to generate crew schedules, loading curves, and skill requirement timelines.

### Business Value
- **Accurate planning** - Based on validated labor norms
- **Resource leveling** - Smooth workload distribution
- **Skill matching** - Right workers at right time
- **Cost control** - Optimize labor costs

## Technical Implementation

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


class ShiftType(Enum):
    """Work shift types."""
    SINGLE = "single"      # 8 hours
    DOUBLE = "double"      # 16 hours (2 shifts)
    TRIPLE = "triple"      # 24 hours (3 shifts)
    EXTENDED = "extended"  # 10 hours


class SkillLevel(Enum):
    """Worker skill levels."""
    UNSKILLED = 1
    SEMI_SKILLED = 2
    SKILLED = 3
    FOREMAN = 4
    SPECIALIST = 5


@dataclass
class LaborRequirement:
    """Labor requirement for a work item."""
    work_item_code: str
    description: str
    total_hours: float
    skill_level: SkillLevel
    trade: str
    start_date: datetime
    end_date: datetime
    daily_hours: float = 0.0


@dataclass
class CrewAssignment:
    """Crew assignment for a period."""
    date: datetime
    trade: str
    skill_level: SkillLevel
    workers_needed: int
    hours_per_worker: float
    total_hours: float
    work_items: List[str]


@dataclass
class LaborSchedule:
    """Complete labor schedule."""
    project_name: str
    start_date: datetime
    end_date: datetime
    total_labor_hours: float
    peak_workers: int
    average_workers: float
    assignments: List[CrewAssignment]
    daily_loading: Dict[str, int]
    by_trade: Dict[str, float]


class CWICRLaborScheduler:
    """Schedule labor based on CWICR norms."""

    HOURS_PER_SHIFT = {
        ShiftType.SINGLE: 8,
        ShiftType.DOUBLE: 16,
        ShiftType.TRIPLE: 24,
        ShiftType.EXTENDED: 10
    }

    def __init__(self, cwicr_data: pd.DataFrame):
        self.data = cwicr_data
        self._index_data()

    def _index_data(self):
        """Index work items for fast lookup."""
        if 'work_item_code' in self.data.columns:
            self._code_index = self.data.set_index('work_item_code')
        else:
            self._code_index = None

    def calculate_labor_requirements(self,
                                     items: List[Dict[str, Any]],
                                     project_start: datetime) -> List[LaborRequirement]:
        """Calculate labor requirements from work items."""

        requirements = []

        for item in items:
            code = item.get('work_item_code', item.get('code'))
            qty = item.get('quantity', 0)
            duration_days = item.get('duration_days', 1)
            start_offset = item.get('start_day', 0)

            if self._code_index is not None and code in self._code_index.index:
                work_item = self._code_index.loc[code]
                labor_norm = float(work_item.get('labor_norm', 0) or 0)
                total_hours = labor_norm * qty

                # Determine trade from category
                trade = self._get_trade(work_item.get('category', 'General'))
                skill_level = self._get_skill_level(work_item)

                start_date = project_start + timedelta(days=start_offset)
                end_date = start_date + timedelta(days=duration_days)

                daily_hours = total_hours / duration_days if duration_days > 0 else total_hours

                requirements.append(LaborRequirement(
                    work_item_code=code,
                    description=str(work_item.get('description', '')),
                    total_hours=total_hours,
                    skill_level=skill_level,
                    trade=trade,
                    start_date=start_date,
                    end_date=end_date,
                    daily_hours=daily_hours
                ))

        return requirements

    def _get_trade(self, category: str) -> str:
        """Map category to trade."""
        trade_mapping = {
            'concrete': 'Concrete',
            'masonry': 'Masonry',
            'steel': 'Steel',
            'carpentry': 'Carpentry',
            'plumbing': 'Plumbing',
            'electrical': 'Electrical',
            'hvac': 'HVAC',
            'painting': 'Painting',
            'excavation': 'Earthwork',
            'roofing': 'Roofing'
        }

        cat_lower = str(category).lower()
        for key, trade in trade_mapping.items():
            if key in cat_lower:
                return trade
        return 'General'

    def _get_skill_level(self, work_item) -> SkillLevel:
        """Determine skill level from work item."""
        # Based on complexity or explicit field
        if 'skill_level' in work_item.index:
            level = int(work_item.get('skill_level', 3))
            return SkillLevel(min(max(level, 1), 5))
        return SkillLevel.SKILLED

    def generate_schedule(self,
                         requirements: List[LaborRequirement],
                         shift_type: ShiftType = ShiftType.SINGLE,
                         max_workers_per_trade: int = 50) -> LaborSchedule:
        """Generate labor schedule from requirements."""

        if not requirements:
            return LaborSchedule(
                project_name="",
                start_date=datetime.now(),
                end_date=datetime.now(),
                total_labor_hours=0,
                peak_workers=0,
                average_workers=0,
                assignments=[],
                daily_loading={