cwicr-crew-optimizer

name	cwicr-crew-optimizer
description	Optimize crew composition using CWICR labor norms. Balance productivity, cost, and skill requirements for construction crews.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🗄️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

CWICR Crew Optimizer

Business Case

Problem Statement

Crew planning challenges:

• Right mix of workers?
• Optimal crew size?
• Balance cost vs productivity?
• Match skills to work?

Solution

Optimize crew composition using CWICR labor productivity data to balance cost, output, and skill requirements.

Business Value

• Optimal productivity - Right-sized crews
• Cost efficiency - No overstaffing
• Skill matching - Proper worker mix
• Schedule support - Meet deadlines

Technical Implementation

import pandas as pd
import numpy as np
from typing import Dict, , , , 
 dataclasses  dataclass, field
 enum  Enum
 datetime  date, timedelta


 ():
    
    FOREMAN = 
    JOURNEYMAN = 
    APPRENTICE = 
    LABORER = 
    OPERATOR = 
    HELPER = 


 ():
    
    CONCRETE = 
    CARPENTRY = 
    MASONRY = 
    STEEL = 
    ELECTRICAL = 
    PLUMBING = 
    HVAC = 
    PAINTING = 
    ROOFING = 
    GENERAL = 



 :
    
    worker_type: WorkerType
    trade: Trade
    hourly_rate: 
    productivity_factor:  = 
    overtime_multiplier:  = 



 :
    
    name: 
    trade: Trade
    workers: [[WorkerType, ]]  
    base_productivity:   
    hourly_cost: 
    daily_output: 



 :
    
    work_item: 
    quantity: 
    unit: 
    recommended_crew: CrewComposition
    alternative_crews: [CrewComposition]
    duration_days: 
    total_labor_cost: 
    cost_per_unit: 



STANDARD_CREWS = {
    : {
        : Trade.CONCRETE,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.LABORER, )],
        : 
    },
    : {
        : Trade.CONCRETE,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.LABORER, ), (WorkerType.OPERATOR, )],
        : 
    },
    : {
        : Trade.MASONRY,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.HELPER, )],
        : 
    },
    : {
        : Trade.CARPENTRY,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.APPRENTICE, )],
        : 
    },
    : {
        : Trade.ELECTRICAL,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.APPRENTICE, )],
        : 
    },
    : {
        : Trade.PLUMBING,
        : [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.APPRENTICE, )],
        : 
    }
}


DEFAULT_RATES = {
    WorkerType.FOREMAN: ,
    WorkerType.JOURNEYMAN: ,
    WorkerType.APPRENTICE: ,
    WorkerType.LABORER: ,
    WorkerType.OPERATOR: ,
    WorkerType.HELPER: 
}


 :
    

    HOURS_PER_DAY = 

     ():
        .cost_data = cwicr_data
        .rates = custom_rates  DEFAULT_RATES
         cwicr_data   :
            ._index_data()

     ():
        
           .cost_data.columns:
            ._code_index = .cost_data.set_index()
        :
            ._code_index = 

     () -> [, ]:
        
         ._code_index    code   ._code_index.index:
             (, )

        item = ._code_index.loc[code]
        norm = (item.get(, item.get(, ))  )
        unit = (item.get(, ))

         (norm, unit)

     () -> :
        
        total = 
         worker_type, count  workers:
            rate = .rates.get(worker_type, )
            total += rate * count
         total

     () -> CrewComposition:
        

        hourly_cost = .calculate_crew_cost(workers)
        daily_output = base_productivity * .HOURS_PER_DAY

         CrewComposition(
            name=name,
            trade=trade,
            workers=workers,
            base_productivity=base_productivity,
            hourly_cost=hourly_cost,
            daily_output=daily_output
        )

     () -> CrewOptimizationResult:
        

        labor_norm, unit = .get_labor_norm(work_item_code)
        total_hours = quantity * labor_norm

        
        trade = ._detect_trade(work_item_code)

        
        crews = []

        
        small_workers = [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.LABORER, )]
        small_crew = .build_crew(, trade, small_workers, )
        crews.append(small_crew)

        
        med_workers = [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.LABORER, )]
        med_crew = .build_crew(, trade, med_workers, )
        crews.append(med_crew)

        
        large_workers = [(WorkerType.FOREMAN, ), (WorkerType.JOURNEYMAN, ), (WorkerType.LABORER, )]
        large_crew = .build_crew(, trade, large_workers, )
        crews.append(large_crew)

        
        results = []
         crew  crews:
            
            crew_workers = (count  _, count  crew.workers)
            efficiency = ._crew_efficiency(crew_workers)

            effective_productivity = crew.base_productivity * efficiency
            hours_needed = total_hours / effective_productivity
            days_needed = hours_needed / .HOURS_PER_DAY
            labor_cost = hours_needed * crew.hourly_cost
            cost_per_unit = labor_cost / quantity  quantity >   

            results.append({
                : crew,
                : days_needed,
                : labor_cost,
                : cost_per_unit,
                : efficiency
            })

        
         target_days:
            
            valid = [r  r  results  r[] <= target_days]
             valid:
                best = (valid, key= x: x[])
            :
                best = (results, key= x: x[])
        :
            
            best = (results, key= x: x[])

        recommended = best[]
        alternatives = [r[]  r  results  r[] != recommended]

         CrewOptimizationResult(
            work_item=work_item_code,
            quantity=quantity,
            unit=unit,
            recommended_crew=recommended,
            alternative_crews=alternatives,
            duration_days=(best[], ),
            total_labor_cost=(best[], ),
            cost_per_unit=(best[], )
        )

     () -> Trade:
        
        code_lower = code.lower()

        trade_map = {
            : Trade.CONCRETE,
            : Trade.CARPENTRY,
            : Trade.MASONRY,
            : Trade.STEEL,
            : Trade.STEEL,
            : Trade.ELECTRICAL,
            : Trade.PLUMBING,
            : Trade.HVAC,
            : Trade.PAINTING,
            : Trade.ROOFING
        }

         key, trade  trade_map.items():
             key  code_lower:
                 trade

         Trade.GENERAL

     () -> :
        
         crew_size <= :
             
         crew_size <= :
             
         crew_size <= :
             
         crew_size <= :
             
        :
             

     () -> [, ]:
        

         result.duration_days <= available_days:
             {
                : ,
                : result.duration_days,
                : ,
                : ,
                : result.total_labor_cost
            }

        
        regular_hours = available_days * .HOURS_PER_DAY
        total_hours_available = available_days * (.HOURS_PER_DAY + max_overtime_hours)

        labor_norm, _ = .get_labor_norm(result.work_item)
        total_hours_needed = result.quantity * labor_norm / result.recommended_crew.base_productivity

         total_hours_needed > total_hours_available:
            
            overtime_hours = available_days * max_overtime_hours
            shortage = total_hours_needed - total_hours_available
        :
            overtime_hours = total_hours_needed - regular_hours
            shortage = 

        overtime_cost = overtime_hours * result.recommended_crew.hourly_cost * 

         {
            : ,
            : available_days,
            : max_overtime_hours,
            : (overtime_hours, ),
            : (overtime_cost, ),
            : (result.total_labor_cost + overtime_cost, ),
            : (shortage, )  shortage >   ,
            : shortage == 
        }

     () -> :
        

         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary_data = []
             r  results:
                workers_str = .join(  wt, count  r.recommended_crew.workers)
                summary_data.append({
                    : r.work_item,
                    : r.quantity,
                    : r.unit,
                    : r.recommended_crew.name,
                    : workers_str,
                    : r.duration_days,
                    : r.total_labor_cost,
                    : r.cost_per_unit
                })

            summary_df = pd.DataFrame(summary_data)
            summary_df.to_excel(writer, sheet_name=, index=)

            
            totals_df = pd.DataFrame([{
                : (r.duration_days  r  results),
                : (r.total_labor_cost  r  results)
            }])
            totals_df.to_excel(writer, sheet_name=, index=)

         output_path