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': 'su