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

Quick Start

from datetime import datetime

Load CWICR data

cwicr = pd.read_parquet("ddc_cwicr_en.parquet")

Initialize scheduler

scheduler = CWICRLaborScheduler(cwicr)

Define work items with schedule

items = [ {'work_item_code': 'EXCV-001', 'quantity': 500, 'duration_days': 5, 'start_day': 0}, {'work_item_code': 'CONC-002', 'quantity': 200, 'duration_days': 10, 'start_day': 5}, {'work_item_code': 'REBAR-003', 'quantity': 5000, 'duration_days': 8, 'start_day': 3} ]

Calculate requirements

project_start = datetime(2024, 6, 1) requirements = scheduler.calculate_labor_requirements(items, project_start)

Generate schedule

schedule = scheduler.generate_schedule(requirements)

print(f"Total Labor Hours: {schedule.total_labor_hours:,.0f}") print(f"Peak Workers: {schedule.peak_workers}") print(f"Average Workers: {schedule.average_workers}")

Common Use Cases

1. Resource Leveling

Check if schedule can meet target

leveled = scheduler.level_resources(schedule, target_workers=25)

2. Loading Curve

Get labor loading data for charts

loading_df = scheduler.generate_loading_curve(schedule)

3. Trade Breakdown

See hours by trade

trades = scheduler.get_trade_breakdown(schedule) print(trades)

4. Weekly Schedule Export

gen = WeeklyScheduleGenerator(scheduler) gen.export_to_excel(schedule, "labor_schedule.xlsx")

Resources

• GitHub: OpenConstructionEstimate-DDC-CWICR

• DDC Book: Chapter 3.1 - Labor Resource Planning