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

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={},
                by_trade={}
            )

        hours_per_day = self.HOURS_PER_SHIFT[shift_type]

        # Find date range
        start_date = min(r.start_date for r in requirements)
        end_date = max(r.end_date for r in requirements)

        # Build daily labor loading
        daily_loading = defaultdict(lambda: defaultdict(float))

        for req in requirements:
            current = req.start_date
            while current < req.end_date:
                date_key = current.strftime('%Y-%m-%d')
                daily_loading[date_key][req.trade] += req.daily_hours
                current += timedelta(days=1)

        # Convert to crew assignments
        assignments = []
        daily_totals = {}
        by_trade = defaultdict(float)

        for date_key, trades in daily_loading.items():
            date = datetime.strptime(date_key, '%Y-%m-%d')
            day_total = 0

            for trade, hours in trades.items():
                workers = int(np.ceil(hours / hours_per_day))
                workers = min(workers, max_workers_per_trade)

                assignments.append(CrewAssignment(
                    date=date,
                    trade=trade,
                    skill_level=SkillLevel.SKILLED,
                    workers_needed=workers,
                    hours_per_worker=hours_per_day,
                    total_hours=hours,
                    work_items=[]
                ))

                day_total += workers
                by_trade[trade] += hours

            daily_totals[date_key] = day_total

        # Statistics
        total_hours = sum(r.total_hours for r in requirements)
        peak_workers = max(daily_totals.values()) if daily_totals else 0
        avg_workers = sum(daily_totals.values()) / len(daily_totals) if daily_totals else 0

        return LaborSchedule(
            project_name="Project",
            start_date=start_date,
            end_date=end_date,
            total_labor_hours=total_hours,
            peak_workers=peak_workers,
            average_workers=round(avg_workers, 1),
            assignments=assignments,
            daily_loading=dict(daily_totals),
            by_trade=dict(by_trade)
        )

    def level_resources(self,
                       schedule: LaborSchedule,
                       target_workers: int) -> LaborSchedule:
        """Level resources to target workforce size."""

        # Resource leveling algorithm
        # Shifts work to reduce peaks while maintaining total hours

        daily_loads = schedule.daily_loading.copy()

        # Find days exceeding target
        over_days = {d: w for d, w in daily_loads.items() if w > target_workers}
        under_days = {d: w for d, w in daily_loads.items() if w < target_workers}

        # Simple leveling: can't easily shift without changing durations
        # Return schedule with analysis

        leveling_analysis = {
            'days_over_target': len(over_days),
            'days_under_target': len(under_days),
            'max_over': max(over_days.values()) - target_workers if over_days else 0,
            'leveling_possible': len(over_days) == 0
        }

        return schedule

    def generate_loading_curve(self,
                               schedule: LaborSchedule) -> pd.DataFrame:
        """Generate labor loading curve data."""

        data = []
        for date_str, workers in sorted(schedule.daily_loading.items()):
            data.append({
                'date': date_str,
                'workers': workers,
                'cumulative_hours': 0  # Would need to calculate
            })

        df = pd.DataFrame(data)

        # Add cumulative hours
        if not df.empty:
            hours_per_worker = 8  # Assuming single shift
            df['daily_hours'] = df['workers'] * hours_per_worker
            df['cumulative_hours'] = df['daily_hours'].cumsum()

        return df

    def get_trade_breakdown(self,
                           schedule: LaborSchedule) -> pd.DataFrame:
        """Get labor breakdown by trade."""

        trade_data = []
        for trade, hours in schedule.by_trade.items():
            trade_data.append({
                'trade': trade,
                'total_hours': round(hours, 1),
                'worker_days': round(hours / 8, 1),
                'percentage': round(hours / schedule.total_labor_hours * 100, 1) if schedule.total_labor_hours > 0 else 0
            })

        return pd.DataFrame(trade_data).sort_values('total_hours', ascending=False)

    def optimize_crew_composition(self,
                                  requirements: List[LaborRequirement],
                                  available_workers: Dict[str, int]) -> Dict[str, Any]:
        """Optimize crew composition based on availability."""

        required_by_trade = defaultdict(float)
        for req in requirements:
            required_by_trade[req.trade] += req.total_hours

        analysis = {
            'sufficient': True,
            'shortages': {},
            'surplus': {},
            'recommendations': []
        }

        for trade, hours_needed in required_by_trade.items():
            workers_needed = int(np.ceil(hours_needed / 8))  # Per day
            available = available_workers.get(trade, 0)

            if workers_needed > available:
                analysis['sufficient'] = False
                analysis['shortages'][trade] = workers_needed - available
                analysis['recommendations'].append(
                    f"Hire {workers_needed - available} additional {trade} workers"
                )
            elif available > workers_needed * 1.5:
                analysis['surplus'][trade] = available - workers_needed

        return analysis


class WeeklyScheduleGenerator:
    """Generate weekly labor schedules."""

    def __init__(self, scheduler: CWICRLaborScheduler):
        self.scheduler = scheduler

    def generate_weekly_schedule(self,
                                 schedule: LaborSchedule,
                                 week_start: datetime) -> pd.DataFrame:
        """Generate schedule for specific week."""

        week_end = week_start + timedelta(days=7)

        weekly_assignments = [
            a for a in schedule.assignments
            if week_start <= a.date < week_end
        ]

        # Pivot by day and trade
        data = []
        for a in weekly_assignments:
            data.append({
                'date': a.date.strftime('%Y-%m-%d'),
                'day': a.date.strftime('%A'),
                'trade': a.trade,
                'workers': a.workers_needed,
                'hours': a.total_hours
            })

        return pd.DataFrame(data)

    def export_to_excel(self,
                       schedule: LaborSchedule,
                       output_path: str) -> str:
        """Export schedule to Excel."""

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Loading curve
            loading = self.scheduler.generate_loading_curve(schedule)
            loading.to_excel(writer, sheet_name='Loading Curve', index=False)

            # Trade breakdown
            trades = self.scheduler.get_trade_breakdown(schedule)
            trades.to_excel(writer, sheet_name='By Trade', index=False)

            # Summary
            summary = pd.DataFrame([{
                'Total Labor Hours': schedule.total_labor_hours,
                'Peak Workers': schedule.peak_workers,
                'Average Workers': schedule.average_workers,
                'Project Duration (days)': (schedule.end_date - schedule.start_date).days
            }])
            summary.to_excel(writer, sheet_name='Summary', index=False)

        return output_path