labor-productivity-analytics

労働生産性分析

概要

このスキルは、建設プロジェクト向けにデータに基づく労働生産性分析を実現します。作業員のパフォーマンスを追跡、測定、最適化して、プロジェクトの効率を高め、コストを削減します。

機能：

• 生産性指標の算出
• 時系列分析
• 作業班の最適化
• リソース予測
• ベンチマーク
• 変動分析

クイックスタート

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}%)")

包括的な生産性システム

データの収集と管理

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)

生産性分析エンジン

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': 'sum',
            'quantity': 'sum',
            'unit_rate': ['mean', 'std', 'min', 'max']
        })

        grouped.columns = ['total_hours', 'total_quantity', 'avg_rate', 'std_rate', 'min_rate', 'max_rate']
        grouped['hours_per_unit'] = grouped['total_hours'] / grouped['total_quantity']

        # Add standard comparison
        grouped['standard_rate'] = grouped.index.map(
            lambda x: 1 / self.db.standards.get(x, {}).get('hours_per_unit', 1)
        )
        grouped['variance_pct'] = (grouped['avg_rate'] - grouped['standard_rate']) / grouped['standard_rate'] * 100

        return grouped

    def analyze_by_crew(self) -> pd.DataFrame:
        """Analyze productivity by crew"""
        if self.df.empty:
            return pd.DataFrame()

        grouped = self.df.groupby('crew_id').agg({
            'hours': 'sum',
            'quantity': 'sum',
            'productivity_factor': 'mean',
            'date': 'nunique'
        }).rename(columns={'date': 'work_days'})

        grouped['avg_daily_hours'] = grouped['hours'] / grouped['work_days']
        grouped['avg_daily_quantity'] = grouped['quantity'] / grouped['work_days']

        return grouped

    def analyze_trends(self, window_days: int = 7) -> Dict:
        """Analyze productivity trends over time"""
        if self.df.empty:
            return {}

        self.df['date'] = pd.to_datetime(self.df['date'])
        daily = self.df.groupby('date').agg({
            'hours': 'sum',
            'quantity': 'sum',
            'productivity_factor': 'mean'
        })

        # Rolling average
        daily['productivity_ma'] = daily['productivity_factor'].rolling(window=window_days).mean()

        # Trend analysis
        if len(daily) > window_days:
            x = np.arange(len(daily))
            slope, intercept, r_value, p_value, std_err = stats.linregress(
                x, daily['productivity_factor'].values
            )

            trend = 'improving' if slope > 0.001 else 'declining' if slope < -0.001 else 'stable'
        else:
            slope, trend = 0, 'insufficient_data'

        return {
            'daily_data': daily.to_dict(),
            'trend': trend,
            'trend_slope': slope,
            'avg_productivity': daily['productivity_factor'].mean(),
            'productivity_std': daily['productivity_factor'].std()
        }

    def analyze_weather_impact(self) -> Dict:
        """Analyze weather impact on productivity"""
        if self.df.empty:
            return {}

        weather_impact = self.df.groupby('weather').agg({
            'productivity_factor': ['mean', 'count'],
            'hours': 'sum'
        })

        weather_impact.columns = ['avg_productivity', 'record_count', 'total_hours']

        baseline = weather_impact.loc['clear']['avg_productivity'] if 'clear' in weather_impact.index else 1.0

        impact = {}
        for weather in weather_impact.index:
            productivity = weather_impact.loc[weather]['avg_productivity']
            impact[weather] = {
                'productivity': productivity,
                'impact_pct': (productivity - baseline) / baseline * 100,
                'records': weather_impact.loc[weather]['record_count']
            }

        return impact

    def identify_inefficiencies(self) -> List[Dict]:
        """Identify areas of inefficiency"""
        issues = []

        # Low performing crews
        crew_analysis = self.analyze_by_crew()
        for crew_id, row in crew_analysis.iterrows():
            if row['productivity_factor'] < 0.8:
                issues.append({
                    'type': 'low_crew_productivity',
                    'crew_id': crew_id,
                    'productivity': row['productivity_factor'],
                    'recommendation': 'Review crew composition and training needs'
                })

        # High variance activities
        activity_analysis = self.analyze_by_activity()
        for activity, row in activity_analysis.iterrows():
            if row['std_rate'] > row['avg_rate'] * 0.5:
                issues.append({
                    'type': 'high_variance',
                    'activity': activity,
                    'std_rate': row['std_rate'],
                    'recommendation': 'Standardize work methods and improve planning'
                })

        # Weather impact
        weather_impact = self.analyze_weather_impact()
        for weather, impact in weather_impact.items():
            if impact['impact_pct'] < -20:
                issues.append({
                    'type': 'weather_impact',
                    'weather': weather,
                    'impact_pct': impact['impact_pct'],
                    'recommendation': 'Plan weather-sensitive activities for better conditions'
                })

        return issues

リソース予測

from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
import numpy as np
import pandas as pd

class ResourceForecaster:
    """Forecast labor resource needs"""

    def __init__(self, database: ProductivityDatabase):
        self.db = database
        self.models = {}

    def forecast_hours(self, activity_code: str, remaining_quantity: float,
                      crew_id: str = None) -> Dict:
        """Forecast hours needed to complete remaining work"""
        records = [r for r in self.db.records if r.activity_code == activity_code]

        if not records:
            # Use standard
            standard = self.db.standards.get(activity_code, {})
            hours_per_unit = standard.get('hours_per_unit', 1)
            return {
                'method': 'standard',
                'forecasted_hours': remaining_quantity * hours_per_unit,
                'confidence': 'low'
            }

        # Calculate from historical data
        if crew_id:
            crew_records = [r for r in records if r.crew_id == crew_id]
            if crew_records:
                records = crew_records

        total_hours = sum(r.total_hours for r in records)
        total_quantity = sum(r.quantity_completed for r in records)
        avg_rate = total_hours / total_quantity if total_quantity > 0 else 1

        # Consider trend
        if len(records) >= 5:
            recent_records = sorted(records, key=lambda x: x.date)[-5:]
            recent_rate = sum(r.total_hours for r in recent_records) / sum(r.quantity_completed for r in recent_records)
            # Weighted average (recent performance weighted more)
            adjusted_rate = avg_rate * 0.3 + recent_rate * 0.7
        else:
            adjusted_rate = avg_rate

        return {
            'method': 'historical',
            'forecasted_hours': remaining_quantity * adjusted_rate,
            'avg_rate': avg_rate,
            'adjusted_rate': adjusted_rate,
            'sample_size': len(records),
            'confidence': 'high' if len(records) >= 10 else 'medium' if len(records) >= 5 else 'low'
        }

    def forecast_crew_needs(self, planned_work: List[Dict],
                           available_hours_per_day: float = 8) -> Dict:
        """Forecast crew requirements for planned work

        planned_work: List of {activity_code, quantity, deadline_days}
        """
        crew_needs = []
        total_hours = 0

        for work in planned_work:
            forecast = self.forecast_hours(work['activity_code'], work['quantity'])
            hours_needed = forecast['forecasted_hours']
            total_hours += hours_needed

            workers_per_day = hours_needed / (work['deadline_days'] * available_hours_per_day)

            crew_needs.append({
                'activity': work['activity_code'],
                'hours_needed': hours_needed,
                'deadline_days': work['deadline_days'],
                'workers_needed': int(np.ceil(workers_per_day)),
                'daily_production_target': work['quantity'] / work['deadline_days']
            })

        return {
            'crew_needs': crew_needs,
            'total_hours': total_hours,
            'peak_workers': max(c['workers_needed'] for c in crew_needs),
            'avg_workers': sum(c['workers_needed'] for c in crew_needs) / len(crew_needs)
        }

    def optimize_crew_allocation(self, crews: List[Crew],
                                work_items: List[Dict]) -> Dict:
        """Optimize crew allocation to work items"""
        # Simple greedy allocation based on productivity
        allocations = []
        remaining_work = list(work_items)

        for crew in sorted(crews, key=lambda c: c.avg_productivity, reverse=True):
            if not remaining_work:
                break

            # Find best matching work for this crew
            crew_trade = crew.primary_trade

            best_work = None
            best_score = -1

            for work in remaining_work:
                activity = work['activity_code']
                standard = self.db.standards.get(activity, {})
                work_trade = standard.get('trade')

                # Score based on trade match and productivity
                if work_trade == crew_trade.value if crew_trade else True:
                    score = crew.avg_productivity
                    if score > best_score:
                        best_score = score
                        best_work = work

            if best_work:
                allocations.append({
                    'crew_id': crew.crew_id,
                    'activity': best_work['activity_code'],
                    'expected_productivity': crew.avg_productivity,
                    'quantity': best_work['quantity']
                })
                remaining_work.remove(best_work)

        return {
            'allocations': allocations,
            'unassigned_work': remaining_work
        }

レポートと可視化

def generate_productivity_report(analyzer: ProductivityAnalyzer,
                                forecaster: ResourceForecaster,
                                output_path: str) -> str:
    """Generate comprehensive productivity report"""
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary
        ev_metrics = analyzer.calculate_earned_value_metrics()
        summary = pd.DataFrame([ev_metrics])
        summary.to_excel(writer, sheet_name='Summary', index=False)

        # By Activity
        activity_analysis = analyzer.analyze_by_activity()
        activity_analysis.to_excel(writer, sheet_name='By_Activity')

        # By Crew
        crew_analysis = analyzer.analyze_by_crew()
        crew_analysis.to_excel(writer, sheet_name='By_Crew')

        # Issues
        issues = analyzer.identify_inefficiencies()
        pd.DataFrame(issues).to_excel(writer, sheet_name='Issues', index=False)

        # Weather Impact
        weather = analyzer.analyze_weather_impact()
        pd.DataFrame(weather).T.to_excel(writer, sheet_name='Weather_Impact')

    return output_path

クイックリファレンス

職種	一般的な単位作業量	単位	備考
大工	0.5-1.0	m²/hr	型枠
鉄筋工	15-25	kg/hr	鉄筋結束
左官工	0.3-0.5	m²/hr	レンガ積み
電気工	2-4	points/hr	粗配線
配管工	2-3	fixtures/hr	設置
塗装工	3-5	m²/hr	内装

リソース

• RS Means 労務単価：業界ベンチマーク
• AGC 生産性基準：https://www.agc.org
• DDC ウェブサイト：https://datadrivenconstruction.io

次のステップ

• 労務費の予測については cost-prediction を参照してください
• スケジュールとの連携については 4d-simulation を参照してください
• 生産性リスクについては risk-assessment-ml を参照してください