Labor Productivity Analytics

Overview

This skill implements data-driven labor productivity analysis for construction projects. Track, measure, and optimize workforce performance to improve project efficiency and reduce costs.

Capabilities:

• Productivity metrics calculation

• Time series analysis

• Crew optimization

• Resource forecasting

• Benchmarking

• Variance analysis

Quick Start

from dataclasses import dataclass, field from datetime import date, datetime, timedelta from typing import List, Dict, Optional, Tuple from enum import Enum import numpy as np

@dataclass class WorkLog: worker_id: str work_date: date activity_code: str hours_worked: float quantity_completed: float unit: str crew_id: str weather: str = "normal" notes: str = ""

@dataclass class ProductivityMetric: activity: str unit_rate: float # units per hour hours_per_unit: float # hours per unit standard_rate: float # benchmark variance_pct: float

def calculate_productivity(logs: List[WorkLog], standard_rates: Dict[str, float]) -> List[ProductivityMetric]: """Calculate productivity metrics from work logs""" # Group by activity by_activity = {} for log in logs: if log.activity_code not in by_activity: by_activity[log.activity_code] = {'hours': 0, 'quantity': 0, 'unit': log.unit} by_activity[log.activity_code]['hours'] += log.hours_worked by_activity[log.activity_code]['quantity'] += log.quantity_completed

metrics = []
for activity, data in by_activity.items():
    if data['hours'] > 0 and data['quantity'] > 0:
        unit_rate = data['quantity'] / data['hours']
        hours_per_unit = data['hours'] / data['quantity']
        standard = standard_rates.get(activity, hours_per_unit)
        variance = (hours_per_unit - standard) / standard * 100

        metrics.append(ProductivityMetric(
            activity=activity,
            unit_rate=unit_rate,
            hours_per_unit=hours_per_unit,
            standard_rate=standard,
            variance_pct=variance
        ))

return metrics

Example

logs = [ WorkLog("W001", date.today(), "CONCRETE", 8, 10, "m³", "C01"), WorkLog("W002", date.today(), "CONCRETE", 8, 12, "m³", "C01"), WorkLog("W003", date.today(), "REBAR", 8, 500, "kg", "C02"), ]

standards = {"CONCRETE": 0.7, "REBAR": 0.015} # hours per unit metrics = calculate_productivity(logs, standards) for m in metrics: print(f"{m.activity}: {m.hours_per_unit:.3f} h/unit (variance: {m.variance_pct:+.1f}%)")

Comprehensive Productivity System

Data Collection and Management

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

class TradeCode(Enum): CARPENTER = "carpenter" IRONWORKER = "ironworker" ELECTRICIAN = "electrician" PLUMBER = "plumber" LABORER = "laborer" OPERATOR = "operator" MASON = "mason" PAINTER = "painter" HVAC = "hvac" WELDER = "welder"

@dataclass class Worker: worker_id: str name: str trade: TradeCode skill_level: int # 1-5 hourly_rate: float certifications: List[str] = field(default_factory=list) hire_date: date = None

@dataclass class Crew: crew_id: str name: str foreman_id: str workers: List[str] = field(default_factory=list) primary_trade: TradeCode = None avg_productivity: float = 1.0

@dataclass class DailyProductionRecord: record_id: str date: date crew_id: str activity_code: str activity_name: str

# Time tracking
start_time: datetime
end_time: datetime
total_hours: float
overtime_hours: float = 0

# Production
quantity_completed: float
unit: str
percent_complete: float = 0

# Conditions
weather: str = "clear"
temperature: float = 20
disruptions: List[str] = field(default_factory=list)

# Calculated
productivity_factor: float = 1.0
unit_rate: float = 0

def calculate_metrics(self):
    """Calculate derived metrics"""
    if self.total_hours > 0 and self.quantity_completed > 0:
        self.unit_rate = self.quantity_completed / self.total_hours

class ProductivityDatabase: """Manage productivity data collection"""

def __init__(self, project_id: str):
    self.project_id = project_id
    self.workers: Dict[str, Worker] = {}
    self.crews: Dict[str, Crew] = {}
    self.records: List[DailyProductionRecord] = []
    self.standards: Dict[str, Dict] = {}

def add_worker(self, worker: Worker):
    self.workers[worker.worker_id] = worker

def add_crew(self, crew: Crew):
    self.crews[crew.crew_id] = crew

def log_production(self, record: DailyProductionRecord):
    """Log daily production"""
    record.calculate_metrics()
    self.records.append(record)

    # Update crew average
    crew = self.crews.get(record.crew_id)
    if crew:
        crew_records = [r for r in self.records if r.crew_id == crew.crew_id]
        if crew_records:
            rates = [r.productivity_factor for r in crew_records if r.productivity_factor > 0]
            crew.avg_productivity = sum(rates) / len(rates) if rates else 1.0

def set_standard(self, activity_code: str, hours_per_unit: float,
                unit: str, trade: TradeCode = None):
    """Set productivity standard"""
    self.standards[activity_code] = {
        'hours_per_unit': hours_per_unit,
        'unit': unit,
        'trade': trade.value if trade else None,
        'source': 'manual'
    }

def get_records_df(self) -> pd.DataFrame:
    """Get records as DataFrame"""
    data = []
    for r in self.records:
        data.append({
            'date': r.date,
            'crew_id': r.crew_id,
            'activity_code': r.activity_code,
            'hours': r.total_hours,
            'quantity': r.quantity_completed,
            'unit': r.unit,
            'unit_rate': r.unit_rate,
            'weather': r.weather,
            'productivity_factor': r.productivity_factor
        })
    return pd.DataFrame(data)

Productivity Analysis Engine

from scipy import stats import numpy as np import pandas as pd

class ProductivityAnalyzer: """Analyze labor productivity data"""

def __init__(self, database: ProductivityDatabase):
    self.db = database
    self.df = database.get_records_df()

def calculate_earned_value_metrics(self) -> Dict:
    """Calculate earned value productivity metrics"""
    if self.df.empty:
        return {}

    total_hours = self.df['hours'].sum()
    total_quantity = self.df['quantity'].sum()

    # Calculate budgeted hours (from standards)
    budgeted_hours = 0
    for _, row in self.df.iterrows():
        standard = self.db.standards.get(row['activity_code'], {})
        if standard:
            budgeted_hours += row['quantity'] * standard.get('hours_per_unit', 1)

    # Productivity Index
    productivity_index = budgeted_hours / total_hours if total_hours > 0 else 0

    return {
        'actual_hours': total_hours,
        'budgeted_hours': budgeted_hours,
        'hours_variance': budgeted_hours - total_hours,
        'productivity_index': productivity_index,
        'interpretation': 'efficient' if productivity_index > 1 else 'needs_improvement'
    }

def analyze_by_activity(self) -> pd.DataFrame:
    """Analyze productivity by activity"""
    if self.df.empty:
        return pd.DataFrame()

    grouped = self.df.groupby('activity_code').agg({
        'hours': 'sum',
        'quantity': 'sum',
        'unit_rate': ['mean', 'std', 'min', 'max']
    })

    grouped.columns = ['total_hours', 'total_quantity', 'avg_rate', 'std_rate', 'min_rate', 'max_rate']
    grouped['hours_per_unit'] = grouped['total_hours'] / grouped['total_quantity']

    # Add standard comparison
    grouped['standard_rate'] = grouped.index.map(
        lambda x: 1 / self.db.standards.get(x, {}).get('hours_per_unit', 1)
    )
    grouped['variance_pct'] = (grouped['avg_rate'] - grouped['standard_rate']) / grouped['standard_rate'] * 100

    return grouped

def analyze_by_crew(self) -> pd.DataFrame:
    """Analyze productivity by crew"""
    if self.df.empty:
        return pd.DataFrame()

    grouped = self.df.groupby('crew_id').agg({
        'hours': 'sum',
        'quantity': 'sum',
        'productivity_factor': 'mean',
        'date': 'nunique'
    }).rename(columns={'date': 'work_days'})

    grouped['avg_daily_hours'] = grouped['hours'] / grouped['work_days']
    grouped['avg_daily_quantity'] = grouped['quantity'] / grouped['work_days']

    return grouped

def analyze_trends(self, window_days: int = 7) -> Dict:
    """Analyze productivity trends over time"""
    if self.df.empty:
        return {}

    self.df['date'] = pd.to_datetime(self.df['date'])
    daily = self.df.groupby('date').agg({
        'hours': 'sum',
        'quantity': 'sum',
        'productivity_factor': 'mean'
    })

    # Rolling average
    daily['productivity_ma'] = daily['productivity_factor'].rolling(window=window_days).mean()

    # Trend analysis
    if len(daily) > window_days:
        x = np.arange(len(daily))
        slope, intercept, r_value, p_value, std_err = stats.linregress(
            x, daily['productivity_factor'].values
        )

        trend = 'improving' if slope > 0.001 else 'declining' if slope < -0.001 else 'stable'
    else:
        slope, trend = 0, 'insufficient_data'

    return {
        'daily_data': daily.to_dict(),
        'trend': trend,
        'trend_slope': slope,
        'avg_productivity': daily['productivity_factor'].mean(),
        'productivity_std': daily['productivity_factor'].std()
    }

def analyze_weather_impact(self) -> Dict:
    """Analyze weather impact on productivity"""
    if self.df.empty:
        return {}

    weather_impact = self.df.groupby('weather').agg({
        'productivity_factor': ['mean', 'count'],
        'hours': 'sum'
    })

    weather_impact.columns = ['avg_productivity', 'record_count', 'total_hours']

    baseline = weather_impact.loc['clear']['avg_productivity'] if 'clear' in weather_impact.index else 1.0

    impact = {}
    for weather in weather_impact.index:
        productivity = weather_impact.loc[weather]['avg_productivity']
        impact[weather] = {
            'productivity': productivity,
            'impact_pct': (productivity - baseline) / baseline * 100,
            'records': weather_impact.loc[weather]['record_count']
        }

    return impact

def identify_inefficiencies(self) -> List[Dict]:
    """Identify areas of inefficiency"""
    issues = []

    # Low performing crews
    crew_analysis = self.analyze_by_crew()
    for crew_id, row in crew_analysis.iterrows():
        if row['productivity_factor'] < 0.8:
            issues.append({
                'type': 'low_crew_productivity',
                'crew_id': crew_id,
                'productivity': row['productivity_factor'],
                'recommendation': 'Review crew composition and training needs'
            })

    # High variance activities
    activity_analysis = self.analyze_by_activity()
    for activity, row in activity_analysis.iterrows():
        if row['std_rate'] > row['avg_rate'] * 0.5:
            issues.append({
                'type': 'high_variance',
                'activity': activity,
                'std_rate': row['std_rate'],
                'recommendation': 'Standardize work methods and improve planning'
            })

    # Weather impact
    weather_impact = self.analyze_weather_impact()
    for weather, impact in weather_impact.items():
        if impact['impact_pct'] < -20:
            issues.append({
                'type': 'weather_impact',
                'weather': weather,
                'impact_pct': impact['impact_pct'],
                'recommendation': 'Plan weather-sensitive activities for better conditions'
            })

    return issues

Resource Forecasting

from sklearn.linear_model import LinearRegression from sklearn.ensemble import RandomForestRegressor import numpy as np import pandas as pd

class ResourceForecaster: """Forecast labor resource needs"""

def __init__(self, database: ProductivityDatabase):
    self.db = database
    self.models = {}

def forecast_hours(self, activity_code: str, remaining_quantity: float,
                  crew_id: str = None) -> Dict:
    """Forecast hours needed to complete remaining work"""
    records = [r for r in self.db.records if r.activity_code == activity_code]

    if not records:
        # Use standard
        standard = self.db.standards.get(activity_code, {})
        hours_per_unit = standard.get('hours_per_unit', 1)
        return {
            'method': 'standard',
            'forecasted_hours': remaining_quantity * hours_per_unit,
            'confidence': 'low'
        }

    # Calculate from historical data
    if crew_id:
        crew_records = [r for r in records if r.crew_id == crew_id]
        if crew_records:
            records = crew_records

    total_hours = sum(r.total_hours for r in records)
    total_quantity = sum(r.quantity_completed for r in records)
    avg_rate = total_hours / total_quantity if total_quantity > 0 else 1

    # Consider trend
    if len(records) >= 5:
        recent_records = sorted(records, key=lambda x: x.date)[-5:]
        recent_rate = sum(r.total_hours for r in recent_records) / sum(r.quantity_completed for r in recent_records)
        # Weighted average (recent performance weighted more)
        adjusted_rate = avg_rate * 0.3 + recent_rate * 0.7
    else:
        adjusted_rate = avg_rate

    return {
        'method': 'historical',
        'forecasted_hours': remaining_quantity * adjusted_rate,
        'avg_rate': avg_rate,
        'adjusted_rate': adjusted_rate,
        'sample_size': len(records),
        'confidence': 'high' if len(records) >= 10 else 'medium' if len(records) >= 5 else 'low'
    }

def forecast_crew_needs(self, planned_work: List[Dict],
                       available_hours_per_day: float = 8) -> Dict:
    """Forecast crew requirements for planned work

    planned_work: List of {activity_code, quantity, deadline_days}
    """
    crew_needs = []
    total_hours = 0

    for work in planned_work:
        forecast = self.forecast_hours(work['activity_code'], work['quantity'])
        hours_needed = forecast['forecasted_hours']
        total_hours += hours_needed

        workers_per_day = hours_needed / (work['deadline_days'] * available_hours_per_day)

        crew_needs.append({
            'activity': work['activity_code'],
            'hours_needed': hours_needed,
            'deadline_days': work['deadline_days'],
            'workers_needed': int(np.ceil(workers_per_day)),
            'daily_production_target': work['quantity'] / work['deadline_days']
        })

    return {
        'crew_needs': crew_needs,
        'total_hours': total_hours,
        'peak_workers': max(c['workers_needed'] for c in crew_needs),
        'avg_workers': sum(c['workers_needed'] for c in crew_needs) / len(crew_needs)
    }

def optimize_crew_allocation(self, crews: List[Crew],
                            work_items: List[Dict]) -> Dict:
    """Optimize crew allocation to work items"""
    # Simple greedy allocation based on productivity
    allocations = []
    remaining_work = list(work_items)

    for crew in sorted(crews, key=lambda c: c.avg_productivity, reverse=True):
        if not remaining_work:
            break

        # Find best matching work for this crew
        crew_trade = crew.primary_trade

        best_work = None
        best_score = -1

        for work in remaining_work:
            activity = work['activity_code']
            standard = self.db.standards.get(activity, {})
            work_trade = standard.get('trade')

            # Score based on trade match and productivity
            if work_trade == crew_trade.value if crew_trade else True:
                score = crew.avg_productivity
                if score > best_score:
                    best_score = score
                    best_work = work

        if best_work:
            allocations.append({
                'crew_id': crew.crew_id,
                'activity': best_work['activity_code'],
                'expected_productivity': crew.avg_productivity,
                'quantity': best_work['quantity']
            })
            remaining_work.remove(best_work)

    return {
        'allocations': allocations,
        'unassigned_work': remaining_work
    }

Reporting and Visualization

def generate_productivity_report(analyzer: ProductivityAnalyzer, forecaster: ResourceForecaster, output_path: str) -> str: """Generate comprehensive productivity report""" with pd.ExcelWriter(output_path, engine='openpyxl') as writer: # Summary ev_metrics = analyzer.calculate_earned_value_metrics() summary = pd.DataFrame([ev_metrics]) summary.to_excel(writer, sheet_name='Summary', index=False)

    # By Activity
    activity_analysis = analyzer.analyze_by_activity()
    activity_analysis.to_excel(writer, sheet_name='By_Activity')

    # By Crew
    crew_analysis = analyzer.analyze_by_crew()
    crew_analysis.to_excel(writer, sheet_name='By_Crew')

    # Issues
    issues = analyzer.identify_inefficiencies()
    pd.DataFrame(issues).to_excel(writer, sheet_name='Issues', index=False)

    # Weather Impact
    weather = analyzer.analyze_weather_impact()
    pd.DataFrame(weather).T.to_excel(writer, sheet_name='Weather_Impact')

return output_path

Quick Reference

Trade Typical Unit Rate Unit Notes

Carpenter 0.5-1.0 m²/hr Formwork

Ironworker 15-25 kg/hr Rebar tying

Mason 0.3-0.5 m²/hr Brickwork

Electrician 2-4 points/hr Rough-in

Plumber 2-3 fixtures/hr Install

Painter 3-5 m²/hr Interior

Resources

• RS Means Labor Rates: Industry benchmarks

• AGC Productivity Standards: https://www.agc.org

• DDC Website: https://datadrivenconstruction.io

Next Steps

• See cost-prediction for labor cost forecasting

• See 4d-simulation for schedule integration

• See risk-assessment-ml for productivity risk