cwicr-productivity-tracker

CWICR 生産性トラッカー

ビジネスケース

課題

プロジェクトのパフォーマンスを追跡するには、次のことが必要です。

• 実績と計画上の生産性を比較する
• 生産性が低い作業を特定する
• 完了日を予測する
• 過去のデータから学ぶ

ソリューション

実績の時間数・数量を CWICR 基準と比較し、差異分析と予測を生成して生産性を追跡します。

ビジネス価値

• パフォーマンスの可視化 - 生産性指標をリアルタイムで確認
• 早期警告 - 問題が深刻化する前に特定
• 継続的な改善 - 差異から学習
• 正確な予測 - データに基づく予測

技術的な実装

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import defaultdict


class PerformanceStatus(Enum):
    """Performance status categories."""
    EXCELLENT = "excellent"      # >110% productivity
    ON_TARGET = "on_target"      # 90-110%
    BELOW_TARGET = "below_target"  # 70-90%
    CRITICAL = "critical"        # <70%


@dataclass
class ProductivityRecord:
    """Single productivity record."""
    work_item_code: str
    description: str
    date: datetime
    planned_hours: float
    actual_hours: float
    planned_quantity: float
    actual_quantity: float
    productivity_rate: float  # Percentage
    status: PerformanceStatus
    variance_hours: float
    labor_cost_variance: float


@dataclass
class ProductivitySummary:
    """Productivity summary for period/project."""
    period_start: datetime
    period_end: datetime
    total_planned_hours: float
    total_actual_hours: float
    overall_productivity: float
    hours_variance: float
    cost_variance: float
    records: List[ProductivityRecord]
    by_status: Dict[str, int]
    by_category: Dict[str, float]
    trend: List[float]  # Daily/weekly productivity trend


class CWICRProductivityTracker:
    """Track productivity against CWICR norms."""

    def __init__(self, cwicr_data: pd.DataFrame,
                 labor_rate: float = 35.0):
        self.work_items = cwicr_data
        self.labor_rate = labor_rate
        self._index_data()

    def _index_data(self):
        """Index work items for fast lookup."""
        if 'work_item_code' in self.work_items.columns:
            self._work_index = self.work_items.set_index('work_item_code')
        else:
            self._work_index = None

    def _get_status(self, productivity_rate: float) -> PerformanceStatus:
        """Determine performance status from productivity rate."""
        if productivity_rate >= 110:
            return PerformanceStatus.EXCELLENT
        elif productivity_rate >= 90:
            return PerformanceStatus.ON_TARGET
        elif productivity_rate >= 70:
            return PerformanceStatus.BELOW_TARGET
        else:
            return PerformanceStatus.CRITICAL

    def calculate_productivity(self,
                               work_item_code: str,
                               actual_hours: float,
                               actual_quantity: float,
                               date: datetime = None) -> ProductivityRecord:
        """Calculate productivity for single work item."""

        if date is None:
            date = datetime.now()

        if self._work_index is not None and work_item_code in self._work_index.index:
            work_item = self._work_index.loc[work_item_code]
            labor_norm = float(work_item.get('labor_norm', 0) or 0)
            planned_hours = labor_norm * actual_quantity

            # Productivity rate (planned/actual * 100)
            productivity_rate = (planned_hours / actual_hours * 100) if actual_hours > 0 else 0

            # Variances
            hours_variance = planned_hours - actual_hours
            cost_variance = hours_variance * self.labor_rate

            return ProductivityRecord(
                work_item_code=work_item_code,
                description=str(work_item.get('description', '')),
                date=date,
                planned_hours=round(planned_hours, 2),
                actual_hours=actual_hours,
                planned_quantity=actual_quantity,  # Using actual as target
                actual_quantity=actual_quantity,
                productivity_rate=round(productivity_rate, 1),
                status=self._get_status(productivity_rate),
                variance_hours=round(hours_variance, 2),
                labor_cost_variance=round(cost_variance, 2)
            )
        else:
            return ProductivityRecord(
                work_item_code=work_item_code,
                description="NOT FOUND",
                date=date,
                planned_hours=0,
                actual_hours=actual_hours,
                planned_quantity=actual_quantity,
                actual_quantity=actual_quantity,
                productivity_rate=0,
                status=PerformanceStatus.CRITICAL,
                variance_hours=0,
                labor_cost_variance=0
            )

    def track_daily_production(self,
                                records: List[Dict[str, Any]]) -> ProductivitySummary:
        """Track daily production from multiple records."""

        productivity_records = []

        for record in records:
            prod = self.calculate_productivity(
                work_item_code=record.get('work_item_code', record.get('code')),
                actual_hours=record.get('actual_hours', 0),
                actual_quantity=record.get('actual_quantity', 0),
                date=record.get('date', datetime.now())
            )
            productivity_records.append(prod)

        # Aggregate
        total_planned = sum(r.planned_hours for r in productivity_records)
        total_actual = sum(r.actual_hours for r in productivity_records)

        overall_productivity = (total_planned / total_actual * 100) if total_actual > 0 else 0

        # By status
        by_status = defaultdict(int)
        for r in productivity_records:
            by_status[r.status.value] += 1

        # Get date range
        dates = [r.date for r in productivity_records if r.date]
        period_start = min(dates) if dates else datetime.now()
        period_end = max(dates) if dates else datetime.now()

        return ProductivitySummary(
            period_start=period_start,
            period_end=period_end,
            total_planned_hours=round(total_planned, 2),
            total_actual_hours=round(total_actual, 2),
            overall_productivity=round(overall_productivity, 1),
            hours_variance=round(total_planned - total_actual, 2),
            cost_variance=round((total_planned - total_actual) * self.labor_rate, 2),
            records=productivity_records,
            by_status=dict(by_status),
            by_category={},
            trend=[]
        )

    def forecast_completion(self,
                            remaining_work: List[Dict[str, Any]],
                            current_productivity: float,
                            available_hours_per_day: float = 80) -> Dict[str, Any]:
        """Forecast completion based on current productivity."""

        # Calculate remaining planned hours
        total_planned = 0
        for item in remaining_work:
            code = item.get('work_item_code', item.get('code'))
            qty = item.get('quantity', 0)

            if self._work_index is not None and code in self._work_index.index:
                work_item = self._work_index.loc[code]
                labor_norm = float(work_item.get('labor_norm', 0) or 0)
                total_planned += labor_norm * qty

        # Adjust for productivity
        if current_productivity > 0:
            actual_hours_needed = total_planned / (current_productivity / 100)
        else:
            actual_hours_needed = total_planned

        # Days to complete
        days_to_complete = actual_hours_needed / available_hours_per_day if available_hours_per_day > 0 else 0

        return {
            'remaining_planned_hours': round(total_planned, 1),
            'estimated_actual_hours': round(actual_hours_needed, 1),
            'current_productivity': current_productivity,
            'days_to_complete': int(np.ceil(days_to_complete)),
            'forecasted_completion': datetime.now() + timedelta(days=int(np.ceil(days_to_complete))),
            'productivity_impact': round(actual_hours_needed - total_planned, 1)
        }

    def analyze_variance(self,
                         summary: ProductivitySummary) -> Dict[str, Any]:
        """Analyze productivity variances in detail."""

        # Get critical items
        critical = [r for r in summary.records if r.status == PerformanceStatus.CRITICAL]
        below_target = [r for r in summary.records if r.status == PerformanceStatus.BELOW_TARGET]

        # Top impact items (by cost variance)
        sorted_by_impact = sorted(summary.records, key=lambda x: x.labor_cost_variance)
        top_negative = [r for r in sorted_by_impact[:5] if r.labor_cost_variance < 0]
        top_positive = [r for r in sorted_by_impact[-5:] if r.labor_cost_variance > 0]

        return {
            'overall_productivity': summary.overall_productivity,
            'total_hours_variance': summary.hours_variance,
            'total_cost_variance': summary.cost_variance,
            'critical_items_count': len(critical),
            'below_target_count': len(below_target),
            'critical_items': [
                {'code': r.work_item_code, 'productivity': r.productivity_rate, 'variance': r.labor_cost_variance}
                for r in critical
            ],
            'top_negative_impact': [
                {'code': r.work_item_code, 'variance': r.labor_cost_variance}
                for r in top_negative
            ],
            'top_positive_impact': [
                {'code': r.work_item_code, 'variance': r.labor_cost_variance}
                for r in top_positive
            ],
            'recommendations': self._generate_recommendations(critical, below_target)
        }

    def _generate_recommendations(self,
                                   critical: List[ProductivityRecord],
                                   below_target: List[ProductivityRecord]) -> List[str]:
        """Generate improvement recommendations."""
        recommendations = []

        if len(critical) > 0:
            recommendations.append(
                f"Immediate attention needed for {len(critical)} critical items"
            )

        if len(below_target) > 3:
            recommendations.append(
                "Consider crew training or method review for underperforming activities"
            )

        # Check for patterns
        critical_codes = [r.work_item_code for r in critical]
        if any('CONC' in code for code in critical_codes):
            recommendations.append("Review concrete work methods and crew composition")
        if any('EXCV' in code for code in critical_codes):
            recommendations.append("Check equipment availability and operator skills for excavation")

        return recommendations

    def export_report(self,
                      summary: ProductivitySummary,
                      output_path: str) -> str:
        """Export productivity report to Excel."""

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Details
            details_df = pd.DataFrame([
                {
                    'Work Item': r.work_item_code,
                    'Description': r.description,
                    'Date': r.date.strftime('%Y-%m-%d'),
                    'Planned Hours': r.planned_hours,
                    'Actual Hours': r.actual_hours,
                    'Productivity %': r.productivity_rate,
                    'Status': r.status.value,
                    'Hours Variance': r.variance_hours,
                    'Cost Variance': r.labor_cost_variance
                }
                for r in summary.records
            ])
            details_df.to_excel(writer, sheet_name='Details', index=False)

            # Summary
            summary_df = pd.DataFrame([{
                'Period Start': summary.period_start.strftime('%Y-%m-%d'),
                'Period End': summary.period_end.strftime('%Y-%m-%d'),
                'Total Planned Hours': summary.total_planned_hours,
                'Total Actual Hours': summary.total_actual_hours,
                'Overall Productivity %': summary.overall_productivity,
                'Hours Variance': summary.hours_variance,
                'Cost Variance': summary.cost_variance
            }])
            summary_df.to_excel(writer, sheet_name='Summary', index=False)

            # By Status
            status_df = pd.DataFrame([
                {'Status': status, 'Count': count}
                for status, count in summary.by_status.items()
            ])
            status_df.to_excel(writer, sheet_name='By Status', index=False)

        return output_path


class ProductivityDashboard:
    """Generate productivity dashboard data."""

    def __init__(self, tracker: CWICRProductivityTracker):
        self.tracker = tracker

    def get_kpis(self, summary: ProductivitySummary) -> Dict[str, Any]:
        """Get key performance indicators."""
        return {
            'overall_productivity': summary.overall_productivity,
            'productivity_status': 'Good' if summary.overall_productivity >= 90 else 'Needs Attention',
            'hours_saved': max(0, summary.hours_variance),
            'hours_over': abs(min(0, summary.hours_variance)),
            'cost_impact': summary.cost_variance,
            'items_on_target': summary.by_status.get('on_target', 0) + summary.by_status.get('excellent', 0),
            'items_below_target': summary.by_status.get('below_target', 0) + summary.by_status.get('critical', 0)
        }

    def get_trend_data(self,
                       historical_summaries: List[ProductivitySummary]) -> pd.DataFrame:
        """Get productivity trend data for charting."""
        data = []
        for s in historical_summaries:
            data.append({
                'date': s.period_end.strftime('%Y-%m-%d'),
                'productivity': s.overall_productivity,
                'planned_hours': s.total_planned_hours,
                'actual_hours': s.total_actual_hours
            })
        return pd.DataFrame(data)

クイックスタート

# Load CWICR data
cwicr = pd.read_parquet("TR_workitems_costs_resources_DDC_CWICR.parquet")

# Initialize tracker
tracker = CWICRProductivityTracker(cwicr, labor_rate=35.0)

# Track daily production
records = [
    {'work_item_code': 'CONC-001', 'actual_hours': 45, 'actual_quantity': 50},
    {'work_item_code': 'REBAR-002', 'actual_hours': 32, 'actual_quantity': 2000},
    {'work_item_code': 'EXCV-003', 'actual_hours': 28, 'actual_quantity': 100}
]

summary = tracker.track_daily_production(records)

print(f"Overall Productivity: {summary.overall_productivity}%")
print(f"Hours Variance: {summary.hours_variance}")
print(f"Cost Variance: ${summary.cost_variance:,.2f}")

一般的なユースケース

1. 差異分析

analysis = tracker.analyze_variance(summary)
for rec in analysis['recommendations']:
    print(rec)

2. 完了予測

remaining = [
    {'work_item_code': 'CONC-001', 'quantity': 100},
    {'work_item_code': 'REBAR-002', 'quantity': 5000}
]
forecast = tracker.forecast_completion(remaining, current_productivity=85.0)
print(f"Days to Complete: {forecast['days_to_complete']}")

3. レポートのエクスポート

tracker.export_report(summary, "productivity_report.xlsx")

リソース

• GitHub：OpenConstructionEstimate-DDC-CWICR
• DDC の書籍：第 3.1 章 - 生産性管理