cwicr-change-order

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 + 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