productivity-analyzer

生産性アナライザー

ビジネスケース

問題の説明

生産性を理解するには、以下が必要です。

• 実際の生産量を追跡する
• 計画値と比較する
• 問題のある領域を特定する
• プロジェクトの完了を予測する

ソリューション

労働生産性のデータを分析し、傾向を特定してベンチマークと比較し、実行可能なインサイトを提供します。

技術的な実装

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import date, timedelta
from enum import Enum


class ProductivityStatus(Enum):
    EXCELLENT = "excellent"    # >110% of planned
    ON_TARGET = "on_target"    # 90-110%
    BELOW = "below"            # 70-90%
    CRITICAL = "critical"      # <70%


@dataclass
class ProductivityRecord:
    date: date
    activity_code: str
    description: str
    planned_output: float
    actual_output: float
    unit: str
    manhours: float
    crew_size: int
    conditions: str  # weather, access issues


@dataclass
class ProductivityAnalysis:
    activity_code: str
    description: str
    total_planned: float
    total_actual: float
    total_manhours: float
    planned_rate: float  # unit per manhour
    actual_rate: float
    efficiency: float  # percentage
    status: ProductivityStatus
    trend: str  # improving, declining, stable


class ProductivityAnalyzer:
    """Analyze construction productivity data."""

    # Industry benchmark rates (unit per manhour)
    BENCHMARKS = {
        'concrete_pour': 0.5,      # m3/MH
        'rebar_install': 15,       # kg/MH
        'formwork': 0.8,           # m2/MH
        'brick_laying': 35,        # bricks/MH
        'drywall': 1.5,            # m2/MH
        'painting': 3.0,           # m2/MH
        'conduit': 8,              # m/MH
        'pipe': 3,                 # m/MH
        'excavation': 2.5,         # m3/MH
        'backfill': 3.0,           # m3/MH
    }

    def __init__(self):
        self.records: List[ProductivityRecord] = []

    def add_record(self,
                   date: date,
                   activity_code: str,
                   description: str,
                   planned_output: float,
                   actual_output: float,
                   unit: str,
                   manhours: float,
                   crew_size: int,
                   conditions: str = "normal"):
        """Add productivity record."""

        self.records.append(ProductivityRecord(
            date=date,
            activity_code=activity_code,
            description=description,
            planned_output=planned_output,
            actual_output=actual_output,
            unit=unit,
            manhours=manhours,
            crew_size=crew_size,
            conditions=conditions
        ))

    def import_from_dataframe(self, df: pd.DataFrame):
        """Import records from DataFrame."""
        for _, row in df.iterrows():
            self.add_record(
                date=pd.to_datetime(row['date']).date(),
                activity_code=row['activity_code'],
                description=row.get('description', ''),
                planned_output=float(row['planned_output']),
                actual_output=float(row['actual_output']),
                unit=row.get('unit', 'unit'),
                manhours=float(row['manhours']),
                crew_size=int(row.get('crew_size', 1)),
                conditions=row.get('conditions', 'normal')
            )

    def _get_status(self, efficiency: float) -> ProductivityStatus:
        """Determine productivity status."""
        if efficiency >= 110:
            return ProductivityStatus.EXCELLENT
        elif efficiency >= 90:
            return ProductivityStatus.ON_TARGET
        elif efficiency >= 70:
            return ProductivityStatus.BELOW
        else:
            return ProductivityStatus.CRITICAL

    def _calculate_trend(self, records: List[ProductivityRecord]) -> str:
        """Calculate productivity trend."""
        if len(records) < 3:
            return "insufficient_data"

        # Sort by date
        sorted_records = sorted(records, key=lambda x: x.date)

        # Calculate efficiency for first and last third
        n = len(sorted_records)
        third = n // 3

        early_efficiency = []
        late_efficiency = []

        for i, r in enumerate(sorted_records):
            if r.manhours > 0:
                eff = (r.actual_output / r.planned_output * 100) if r.planned_output > 0 else 0
                if i < third:
                    early_efficiency.append(eff)
                elif i >= n - third:
                    late_efficiency.append(eff)

        if not early_efficiency or not late_efficiency:
            return "stable"

        early_avg = np.mean(early_efficiency)
        late_avg = np.mean(late_efficiency)

        if late_avg > early_avg * 1.05:
            return "improving"
        elif late_avg < early_avg * 0.95:
            return "declining"
        else:
            return "stable"

    def analyze_activity(self, activity_code: str) -> Optional[ProductivityAnalysis]:
        """Analyze productivity for specific activity."""

        activity_records = [r for r in self.records if r.activity_code == activity_code]

        if not activity_records:
            return None

        total_planned = sum(r.planned_output for r in activity_records)
        total_actual = sum(r.actual_output for r in activity_records)
        total_manhours = sum(r.manhours for r in activity_records)

        planned_rate = total_planned / total_manhours if total_manhours > 0 else 0
        actual_rate = total_actual / total_manhours if total_manhours > 0 else 0
        efficiency = (total_actual / total_planned * 100) if total_planned > 0 else 0

        return ProductivityAnalysis(
            activity_code=activity_code,
            description=activity_records[0].description,
            total_planned=round(total_planned, 2),
            total_actual=round(total_actual, 2),
            total_manhours=round(total_manhours, 1),
            planned_rate=round(planned_rate, 3),
            actual_rate=round(actual_rate, 3),
            efficiency=round(efficiency, 1),
            status=self._get_status(efficiency),
            trend=self._calculate_trend(activity_records)
        )

    def analyze_all_activities(self) -> List[ProductivityAnalysis]:
        """Analyze all activities."""
        activities = set(r.activity_code for r in self.records)
        return [self.analyze_activity(code) for code in activities if self.analyze_activity(code)]

    def compare_to_benchmark(self, activity_code: str) -> Dict[str, Any]:
        """Compare activity to industry benchmark."""

        analysis = self.analyze_activity(activity_code)
        if not analysis:
            return {}

        # Find matching benchmark
        benchmark = None
        for key, value in self.BENCHMARKS.items():
            if key in activity_code.lower():
                benchmark = value
                break

        if benchmark is None:
            return {
                'activity': activity_code,
                'actual_rate': analysis.actual_rate,
                'benchmark': 'Not available',
                'vs_benchmark': 'N/A'
            }

        vs_benchmark = (analysis.actual_rate / benchmark * 100) if benchmark > 0 else 0

        return {
            'activity': activity_code,
            'actual_rate': analysis.actual_rate,
            'benchmark_rate': benchmark,
            'vs_benchmark_pct': round(vs_benchmark, 1),
            'recommendation': 'Above benchmark' if vs_benchmark >= 100 else 'Below benchmark - investigate'
        }

    def identify_problem_areas(self) -> List[Dict[str, Any]]:
        """Identify activities with productivity issues."""

        problems = []

        for analysis in self.analyze_all_activities():
            if analysis.status in [ProductivityStatus.BELOW, ProductivityStatus.CRITICAL]:
                problems.append({
                    'activity': analysis.activity_code,
                    'efficiency': analysis.efficiency,
                    'status': analysis.status.value,
                    'trend': analysis.trend,
                    'manhours_impacted': analysis.total_manhours,
                    'priority': 'HIGH' if analysis.status == ProductivityStatus.CRITICAL else 'MEDIUM'
                })

        return sorted(problems, key=lambda x: x['efficiency'])

    def forecast_completion(self,
                            activity_code: str,
                            remaining_quantity: float) -> Dict[str, Any]:
        """Forecast completion based on current productivity."""

        analysis = self.analyze_activity(activity_code)
        if not analysis or analysis.actual_rate == 0:
            return {}

        # Manhours needed at current rate
        manhours_needed = remaining_quantity / analysis.actual_rate

        # Average daily manhours
        activity_records = [r for r in self.records if r.activity_code == activity_code]
        avg_daily_mh = np.mean([r.manhours for r in activity_records]) if activity_records else 8

        days_needed = manhours_needed / avg_daily_mh if avg_daily_mh > 0 else 0

        return {
            'activity': activity_code,
            'remaining_qty': remaining_quantity,
            'current_rate': analysis.actual_rate,
            'manhours_needed': round(manhours_needed, 1),
            'days_needed': round(days_needed, 1),
            'estimated_completion': date.today() + timedelta(days=int(days_needed))
        }

    def export_analysis(self, output_path: str) -> str:
        """Export analysis to Excel."""

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Summary
            analyses = self.analyze_all_activities()
            summary_df = pd.DataFrame([
                {
                    'Activity': a.activity_code,
                    'Description': a.description,
                    'Planned': a.total_planned,
                    'Actual': a.total_actual,
                    'Manhours': a.total_manhours,
                    'Efficiency %': a.efficiency,
                    'Status': a.status.value,
                    'Trend': a.trend
                }
                for a in analyses
            ])
            summary_df.to_excel(writer, sheet_name='Summary', index=False)

            # Problems
            problems = self.identify_problem_areas()
            if problems:
                problems_df = pd.DataFrame(problems)
                problems_df.to_excel(writer, sheet_name='Problem Areas', index=False)

            # Raw data
            records_df = pd.DataFrame([
                {
                    'Date': r.date,
                    'Activity': r.activity_code,
                    'Planned': r.planned_output,
                    'Actual': r.actual_output,
                    'Unit': r.unit,
                    'Manhours': r.manhours,
                    'Crew': r.crew_size,
                    'Conditions': r.conditions
                }
                for r in self.records
            ])
            records_df.to_excel(writer, sheet_name='Raw Data', index=False)

        return output_path

クイックスタート

from datetime import date, timedelta

# Initialize analyzer
analyzer = ProductivityAnalyzer()

# Add records
for i in range(10):
    analyzer.add_record(
        date=date.today() - timedelta(days=i),
        activity_code="concrete_pour",
        description="Slab pour Level 3",
        planned_output=20,
        actual_output=18 + (i * 0.3),  # improving
        unit="m3",
        manhours=40,
        crew_size=5
    )

# Analyze
analysis = analyzer.analyze_activity("concrete_pour")
print(f"Efficiency: {analysis.efficiency}%")
print(f"Status: {analysis.status.value}")
print(f"Trend: {analysis.trend}")

一般的なユースケース

1. 問題を特定する

problems = analyzer.identify_problem_areas()
for p in problems:
    print(f"{p['activity']}: {p['efficiency']}% - {p['priority']}")

2. 完了を予測する

forecast = analyzer.forecast_completion("concrete_pour", remaining_quantity=500)
print(f"Days needed: {forecast['days_needed']}")
print(f"Completion: {forecast['estimated_completion']}")

3. ベンチマークと比較する

comparison = analyzer.compare_to_benchmark("concrete_pour")
print(f"vs Benchmark: {comparison['vs_benchmark_pct']}%")

リソース

• DDC Book：第4.1章 - 生産性管理