labor-productivity-analytics

What this skill does

# 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

```python
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

```python
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

```python
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