Labor Productivity Analyzer
Technical Implementation
import pandas as pd
from datetime import date
from typing import Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
class ProductivityStatus(Enum):
EXCEEDING = "exceeding"
ON_TARGET = "on_target"
BELOW_TARGET = "below_target"
CRITICAL = "critical"
@dataclass
class ProductivityEntry:
entry_id: str
date: date
trade: str
activity_code: str
activity_description: str
location: str
crew_size: int
hours_worked: float
quantity_installed: float
unit: str
target_productivity: float # units per hour
@property
def actual_productivity(self) -> float:
if self.hours_worked == 0:
return 0
return self.quantity_installed / self.hours_worked
@property
def productivity_factor(self) -> float:
if self.target_productivity == 0:
return 0
return self.actual_productivity / self.target_productivity
@property
def status(self) -> ProductivityStatus:
pf = self.productivity_factor
if pf >= 1.1:
return ProductivityStatus.EXCEEDING
elif pf >= 0.9:
return ProductivityStatus.ON_TARGET
elif pf >= 0.7:
return ProductivityStatus.BELOW_TARGET
return ProductivityStatus.CRITICAL
class LaborProductivityAnalyzer:
def __init__(self, project_name: str):
self.project_name = project_name
self.entries: List[ProductivityEntry] = []
self.targets: Dict[str, float] = {} # activity_code: target_productivity
self._counter = 0
def set_target(self, activity_code: str, target_productivity: float):
self.targets[activity_code] = target_productivity
def add_entry(self, entry_date: date, trade: str, activity_code: str,
activity_description: str, location: str, crew_size: int,
hours_worked: float, quantity_installed: float,
unit: str) -> ProductivityEntry:
self._counter += 1
entry_id = f"PROD-{self._counter:05d}"
target = self.targets.get(activity_code, 1.0)
entry = ProductivityEntry(
entry_id=entry_id,
date=entry_date,
trade=trade,
activity_code=activity_code,
activity_description=activity_description,
location=location,
crew_size=crew_size,
hours_worked=hours_worked,
quantity_installed=quantity_installed,
unit=unit,
target_productivity=target
)
self.entries.append(entry)
return entry
def get_productivity_by_trade(self) -> Dict[str, Dict[str, Any]]:
by_trade = {}
for entry in self.entries:
if entry.trade not in by_trade:
by_trade[entry.trade] = {'hours': 0, 'quantity': 0, 'entries': 0}
by_trade[entry.trade]['hours'] += entry.hours_worked
by_trade[entry.trade]['quantity'] += entry.quantity_installed
by_trade[entry.trade]['entries'] += 1
for trade in by_trade:
hours = by_trade[trade]['hours']
qty = by_trade[trade]['quantity']
by_trade[trade]['avg_productivity'] = qty / hours if hours > 0 else 0
return by_trade
def get_productivity_by_activity(self) -> Dict[str, Dict[str, Any]]:
by_activity = {}
for entry in self.entries:
code = entry.activity_code
if code not in by_activity:
by_activity[code] = {
'description': entry.activity_description,
'hours': 0, 'quantity': 0, 'target': entry.target_productivity
}
by_activity[code]['hours'] += entry.hours_worked
by_activity[code]['quantity'] += entry.quantity_installed
for code in by_activity:
hours = by_activity[code]['hours']
qty = by_activity[code]['quantity']
by_activity[code]['actual'] = qty / hours if hours > 0 else 0
by_activity[code]['factor'] = (
by_activity[code]['actual'] / by_activity[code]['target']
if by_activity[code]['target'] > 0 else 0
)
return by_activity
def get_low_performers(self) -> List[ProductivityEntry]:
return [e for e in self.entries
if e.status in [ProductivityStatus.BELOW_TARGET, ProductivityStatus.CRITICAL]]
def get_summary(self) -> Dict[str, Any]:
if not self.entries:
return {'total_entries': 0}
total_hours = sum(e.hours_worked for e in self.entries)
factors = [e.productivity_factor for e in self.entries]
avg_factor = sum(factors) / len(factors)
return {
'total_entries': len(self.entries),
'total_hours': total_hours,
'average_productivity_factor': round(avg_factor, 2),
'exceeding': sum(1 for e in self.entries if e.status == ProductivityStatus.EXCEEDING),
'on_target': sum(1 for e in self.entries if e.status == ProductivityStatus.ON_TARGET),
'below_target': sum(1 for e in self.entries if e.status == ProductivityStatus.BELOW_TARGET),
'critical': sum(1 for e in self.entries if e.status == ProductivityStatus.CRITICAL)
}
def export_report(self, output_path: str):
data = [{
'Date': e.date,
'Trade': e.trade,
'Activity': e.activity_code,
'Location': e.location,
'Crew': e.crew_size,
'Hours': e.hours_worked,
'Quantity': e.quantity_installed,
'Unit': e.unit,
'Target': e.target_productivity,
'Actual': round(e.actual_productivity, 2),
'Factor': round(e.productivity_factor, 2),
'Status': e.status.value
} for e in self.entries]
pd.DataFrame(data).to_excel(output_path, index=False)
Quick Start
analyzer = LaborProductivityAnalyzer("Office Tower")
# Set targets
analyzer.set_target("CONC-001", 2.5) # m3 per hour
# Add entry
entry = analyzer.add_entry(
entry_date=date.today(),
trade="Concrete",
activity_code="CONC-001",
activity_description="Pour concrete slab",
location="Level 3",
crew_size=8,
hours_worked=80,
quantity_installed=180,
unit="m3"
)
print(f"Productivity factor: {entry.productivity_factor:.2f}")
print(f"Status: {entry.status.value}")
Resources
- DDC Book: Chapter 3.2 - Resource Management
