CWICR Productivity Tracker
Business Case
Problem Statement
Project performance tracking requires:
- Comparing actual vs planned productivity
- Identifying underperforming activities
- Forecasting completion dates
- Learning from historical data
Solution
Track productivity by comparing actual hours/quantities against CWICR norms, generating variance analysis and forecasts.
Business Value
- Performance visibility - Real-time productivity metrics
- Early warning - Identify issues before escalation
- Continuous improvement - Learn from variances
- Accurate forecasting - Data-driven predictions
Technical Implementation
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_nee