labor-productivity-analyzer

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)