CWICR Change Order Processor
Business Case
Problem Statement
Change orders require:
- Quick cost impact analysis
- Comparison to original scope
- Fair pricing for added work
- Documentation for disputes
Solution
Systematic change order processing using CWICR data to calculate fair costs, document changes, and analyze impact on project budget.
Business Value
- Fair pricing - Based on validated norms
- Quick turnaround - Rapid cost analysis
- Documentation - Clear change records
- Budget tracking - Cumulative impact
Technical Implementation
import pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, date
from enum import Enum
class ChangeType(Enum):
"""Types of changes."""
ADDITION = "addition" # New work added
DELETION = "deletion" # Work removed
MODIFICATION = "modification" # Changed scope
SUBSTITUTION = "substitution" # Material/method change
class ChangeStatus(Enum):
"""Change order status."""
DRAFT = "draft"
SUBMITTED = "submitted"
APPROVED = "approved"
REJECTED = "rejected"
PENDING = "pending"
@dataclass
class ChangeItem:
"""Single change item."""
item_number: int
work_item_code: str
description: str
change_type: ChangeType
original_qty: float
revised_qty: float
unit: str
unit_cost: float
original_cost: float
revised_cost: float
cost_impact: float
@dataclass
class ChangeOrder:
"""Complete change order."""
co_number: str
project_name: str
date_created: date
status: ChangeStatus
description: str
items: List[ChangeItem]
direct_cost_impact: float
overhead_markup: float
profit_markup: float
total_impact: float
schedule_impact_days: int
justification: str
class CWICRChangeOrder:
"""Process change orders using CWICR data."""
def __init__(self,
cwicr_data: pd.DataFrame,
overhead_rate: float = 0.12,
profit_rate: float = 0.08):
self.cost_data = cwicr_data
self.overhead_rate = overhead_rate
self.profit_rate = profit_rate
self._index_data()
self._change_orders: Dict[str, ChangeOrder] = {}
def _index_data(self):
"""Index cost data."""
if 'work_item_code' in self.cost_data.columns:
self._code_index = self.cost_data.set_index('work_item_code')
else:
self._code_index = None
def get_unit_cost(self, code: str) -> Tuple[float, str]:
"""Get unit cost from CWICR."""
if self._code_index is None or code not in self._code_index.index:
return (0, 'unit')
item = self._code_index.loc[code]
labor = float(item.get('labor_cost', 0) or 0)
material = float(item.get('material_cost', 0) or 0)
equipment = float(item.get('equipment_cost', 0) or 0)
unit = str(item.get('unit', 'unit'))
return (labor + material + equipment, unit)
def create_change_order(self,
co_number: str,
project_name: str,
description: str,
justification: str = "") -> str:
"""Create new change order."""
co = ChangeOrder(
co_number=co_number,
project_name=project_name,
date_created=date.today(),
status=ChangeStatus.DRAFT,
description=description,
items=[],
direct_cost_impact=0,
overhead_markup=0,
profit_markup=0,
total_impact=0,
schedule_impact_days=0,
justification=justification
)
self._change_orders[co_number] = co
return co_number
def add_change_item(self,
co_number: str,
work_item_code: str,
change_type: ChangeType,
original_qty: float,
revised_qty: float,
description: str = None) -> ChangeItem:
"""Add item to change order."""
co = self._change_orders.get(co_number)
if co is None:
raise ValueError(f"Change order {co_number} not found")
unit_cost, unit = self.get_unit_cost(work_item_code)
if description is None:
if self._code_index is not None and work_item_code in self._code_index.index:
description = str(self._code_index.loc[work_item_code].get('description', work_item_code))
else:
description = work_item_code
original_cost = original_qty * unit_cost
revised_cost = revised_qty * unit_cost
cost_impact = revised_cost - original_cost
item = ChangeItem(
item_number=len(co.items) + 1,
work_item_code=work_item_code,
description=description,
change_type=change_type,
original_qty=original_qty,
revised_qty=revised_qty,
unit=unit,
unit_cost=unit_cost,
original_cost=round(original_cost, 2),
revised_cost=round(revised_cost, 2),
cost_impact=round(cost_impact, 2)
)
co.items.append(item)
self._recalculate_totals(co_number)
return item
def _recalculate_totals(self, co_number: str):
"""Recalculate change order totals."""
co = self._change_orders.get(co_number)
if co is None:
return
direct_impact = sum(item.cost_impact for item in co.items)
overhead = direct_impact * self.overhead_rate
profit = (direct_impact + overhead) * self.profit_rate
co.direct_cost_impact = round(direct_impact, 2)
co.overhead_markup = round(overhead, 2)
co.profit_markup = round(profit, 2)
co.total_impact = round(direct_impact + overhead + profit, 2)
def set_schedule_impact(self, co_number: str, days: int):
"""Set schedule impact for change order."""
co = self._change_orders.get(co_number)
if co:
co.schedule_impact_days = days
def update_status(self, co_number: str, status: ChangeStatus):
"""Update change order status."""
co = self._change_orders.get(co_number)
if co:
co.status = status
def get_change_order(self, co_number: str) -> Optional[ChangeOrder]:
"""Get change order by number."""
return self._change_orders.get(co_number)
def calculate_quick_impact(self,
changes: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Quick impact calculation without creating CO."""
additions = 0
deletions = 0
modifications = 0
for change in changes:
code = change.get('work_item_code', change.get('code'))
change_type = change.get('change_type', 'modification')
original = change.get('original_qty', 0)
revised = change.get('revised_qty', 0)
unit_cost, _ = self.get_unit_cost(code)
if change_type == 'addition':
additions += revised * unit_cost
elif change_type == 'deletion':
deletions += original * unit_cost
else:
modifications += (revised - original) * unit_cost
net_direct = additions - deletions + modifications
overhead = net_direct * self.overhead_rate
profit = (net_direct + overhead) * self.profit_rate
return {
'additions': round(additions, 2),
'deletions': round(deletions, 2),
'modifications': round(modifications, 2),
'net_direct_impact': round(net_direct, 2),
'overhead': round(overhead, 2),
'profit': round(profit, 2),
'total_impact': round(net_direct + overhead + pr