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