Process construction change orders using CWICR data. Calculate cost impact, compare to original estimate, and generate change order documentation.
Change orders require:
Systematic change order processing using CWICR data to calculate fair costs, document changes, and analyze impact on project budget.
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 + profit, 2)
}
def compare_to_budget(self,
co_number: str,
original_budget: float,
approved_changes: float = 0) -> Dict[str, Any]:
"""Compare change order to project budget."""
co = self._change_orders.get(co_number)
if co is None:
return {}
current_budget = original_budget + approved_changes
new_budget = current_budget + co.total_impact
return {
'original_budget': original_budget,
'previously_approved_changes': approved_changes,
'current_budget': current_budget,
'this_change_order': co.total_impact,
'new_budget': round(new_budget, 2),
'change_from_original': round(new_budget - original_budget, 2),
'change_percent': round((new_budget - original_budget) / original_budget * 100, 1)
}
def get_project_changes_summary(self,
project_name: str) -> Dict[str, Any]:
"""Get summary of all changes for a project."""
project_cos = [
co for co in self._change_orders.values()
if co.project_name == project_name
]
total_additions = 0
total_deletions = 0
total_impact = 0
total_schedule_impact = 0
for co in project_cos:
for item in co.items:
if item.change_type == ChangeType.ADDITION:
total_additions += item.cost_impact
elif item.change_type == ChangeType.DELETION:
total_deletions += abs(item.cost_impact)
total_impact += co.total_impact
total_schedule_impact += co.schedule_impact_days
return {
'project': project_name,
'total_change_orders': len(project_cos),
'approved': len([co for co in project_cos if co.status == ChangeStatus.APPROVED]),
'pending': len([co for co in project_cos if co.status == ChangeStatus.PENDING]),
'total_additions': round(total_additions, 2),
'total_deletions': round(total_deletions, 2),
'net_cost_impact': round(total_impact, 2),
'total_schedule_days': total_schedule_impact
}
def export_change_order(self,
co_number: str,
output_path: str) -> str:
"""Export change order to Excel."""
co = self._change_orders.get(co_number)
if co is None:
raise ValueError(f"Change order {co_number} not found")
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Header
header_df = pd.DataFrame([{
'Change Order': co.co_number,
'Project': co.project_name,
'Date': co.date_created,
'Status': co.status.value,
'Description': co.description,
'Schedule Impact (days)': co.schedule_impact_days
}])
header_df.to_excel(writer, sheet_name='Summary', index=False)
# Items
items_df = pd.DataFrame([
{
'#': item.item_number,
'Code': item.work_item_code,
'Description': item.description,
'Type': item.change_type.value,
'Original Qty': item.original_qty,
'Revised Qty': item.revised_qty,
'Unit': item.unit,
'Unit Cost': item.unit_cost,
'Original Cost': item.original_cost,
'Revised Cost': item.revised_cost,
'Impact': item.cost_impact
}
for item in co.items
])
items_df.to_excel(writer, sheet_name='Items', index=False)
# Totals
totals_df = pd.DataFrame([{
'Direct Cost Impact': co.direct_cost_impact,
f'Overhead ({self.overhead_rate:.0%})': co.overhead_markup,
f'Profit ({self.profit_rate:.0%})': co.profit_markup,
'Total Impact': co.total_impact
}])
totals_df.to_excel(writer, sheet_name='Totals', index=False)
return output_path
# Load CWICR data
cwicr = pd.read_parquet("ddc_cwicr_en.parquet")
# Initialize change order processor
co_processor = CWICRChangeOrder(cwicr, overhead_rate=0.12, profit_rate=0.08)
# Create change order
co_processor.create_change_order(
co_number="CO-001",
project_name="Building A",
description="Additional foundation work due to soil conditions"
)
# Add items
co_processor.add_change_item(
co_number="CO-001",
work_item_code="EXCV-002",
change_type=ChangeType.ADDITION,
original_qty=0,
revised_qty=150
)
# Get change order
co = co_processor.get_change_order("CO-001")
print(f"Direct Impact: ${co.direct_cost_impact:,.2f}")
print(f"Total Impact: ${co.total_impact:,.2f}")
changes = [
{'code': 'CONC-001', 'change_type': 'addition', 'original_qty': 0, 'revised_qty': 50},
{'code': 'REBAR-002', 'change_type': 'modification', 'original_qty': 1000, 'revised_qty': 1500}
]
impact = co_processor.calculate_quick_impact(changes)
print(f"Net Impact: ${impact['total_impact']:,.2f}")
comparison = co_processor.compare_to_budget(
co_number="CO-001",
original_budget=5000000,
approved_changes=150000
)
print(f"Budget Change: {comparison['change_percent']}%")
co_processor.export_change_order("CO-001", "change_order_001.xlsx")