Change Order Manager

Overview

Manage the complete change order lifecycle from potential change identification through approval and payment. Track cost and schedule impacts, maintain documentation, and provide analytics for project control.

Change Order Workflow

┌─────────────────────────────────────────────────────────────────┐
│                  CHANGE ORDER WORKFLOW                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Identify  →  Document  →  Price  →  Negotiate  →  Execute     │
│  ────────     ────────     ─────     ─────────     ───────     │
│  📋 PCO       📝 RFP       💰 Quote  🤝 Review     ✅ Approve   │
│  🔍 Review    📸 Photos    ⏰ Time   📧 Submit     📄 Sign      │
│  📧 Notify    📄 Backup    📊 Impact 💬 Discuss    💵 Pay       │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum
import json

class ChangeOrderStatus(Enum):
    DRAFT = "draft"
    SUBMITTED = "submitted"
    UNDER_REVIEW = "under_review"
    PRICING = "pricing"
    NEGOTIATING = "negotiating"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXECUTED = "executed"
    VOID = "void"

class ChangeType(Enum):
    OWNER_DIRECTED = "owner_directed"
    DESIGN_ERROR = "design_error"
    FIELD_CONDITION = "field_condition"
    CODE_CHANGE = "code_change"
    VALUE_ENGINEERING = "value_engineering"
    SCHEDULE_ACCELERATION = "schedule_acceleration"
    SCOPE_REDUCTION = "scope_reduction"

class PricingMethod(Enum):
    LUMP_SUM = "lump_sum"
    UNIT_PRICE = "unit_price"
    TIME_AND_MATERIALS = "time_and_materials"
    COST_PLUS = "cost_plus"

@dataclass
class CostBreakdown:
    labor: float = 0.0
    materials: float = 0.0
    equipment: float = 0.0
    subcontractor: float = 0.0
    overhead: float = 0.0
    profit: float = 0.0
    bond: float = 0.0

    @property
    def direct_cost(self) -> float:
        return self.labor + self.materials + self.equipment + self.subcontractor

    @property
    def total(self) -> float:
        return self.direct_cost + self.overhead + self.profit + self.bond

@dataclass
class ChangeOrderItem:
    id: str
    description: str
    quantity: float
    unit: str
    unit_price: float
    total_price: float
    spec_section: str = ""
    csi_code: str = ""

@dataclass
class ChangeOrder:
    id: str
    number: int
    title: str
    description: str
    change_type: ChangeType
    status: ChangeOrderStatus

    # Dates
    identified_date: datetime
    submitted_date: Optional[datetime] = None
    approved_date: Optional[datetime] = None
    executed_date: Optional[datetime] = None

    # Pricing
    pricing_method: PricingMethod = PricingMethod.LUMP_SUM
    proposed_amount: float = 0.0
    approved_amount: float = 0.0
    cost_breakdown: CostBreakdown = field(default_factory=CostBreakdown)
    line_items: List[ChangeOrderItem] = field(default_factory=list)

    # Schedule
    proposed_time_days: int = 0
    approved_time_days: int = 0
    impacts_critical_path: bool = False

    # Documentation
    rfi_references: List[str] = field(default_factory=list)
    drawing_references: List[str] = field(default_factory=list)
    photo_attachments: List[str] = field(default_factory=list)
    backup_documents: List[str] = field(default_factory=list)

    # Tracking
    created_by: str = ""
    assigned_to: str = ""
    notes: List[Dict] = field(default_factory=list)

@dataclass
class ChangeOrderLog:
    project_id: str
    project_name: str
    original_contract: float
    change_orders: List[ChangeOrder]
    total_approved: float
    total_pending: float
    revised_contract: float

class ChangeOrderManager:
    """Manage construction change orders."""

    # Default markup rates
    DEFAULT_MARKUPS = {
        "overhead": 0.10,  # 10%
        "profit": 0.10,    # 10%
        "bond": 0.01,      # 1%
    }

    def __init__(self, project_id: str, project_name: str,
                 original_contract: float):
        self.project_id = project_id
        self.project_name = project_name
        self.original_contract = original_contract
        self.change_orders: Dict[str, ChangeOrder] = {}
        self.next_number = 1
        self.markup_rates = dict(self.DEFAULT_MARKUPS)

    def set_markup_rates(self, overhead: float = None, profit: float = None,
                        bond: float = None):
        """Set markup rates for cost calculations."""
        if overhead is not None:
            self.markup_rates["overhead"] = overhead
        if profit is not None:
            self.markup_rates["profit"] = profit
        if bond is not None:
            self.markup_rates["bond"] = bond

    def create_change_order(self, title: str, description: str,
                           change_type: ChangeType,
                           created_by: str = "") -> ChangeOrder:
        """Create new change order."""
        co_id = f"CO-{self.project_id}-{self.next_number:04d}"

        co = ChangeOrder(
            id=co_id,
            number=self.next_number,
            title=title,
            description=description,
            change_type=change_type,
            status=ChangeOrderStatus.DRAFT,
            identified_date=datetime.now(),
            created_by=created_by
        )

        self.change_orders[co_id] = co
        self.next_number += 1

        return co

    def add_line_item(self, co_id: str, description: str,
                     quantity: float, unit: str, unit_price: float,
                     spec_section: str = "", csi_code: str = "") -> ChangeOrderItem:
        """Add line item to change order."""
        if co_id not in self.change_orders:
            raise ValueError(f"Change order {co_id} not found")

        co = self.change_orders[co_id]

        item_id = f"{co_id}-{len(co.line_items)+1:03d}"
        item = ChangeOrderItem(
            id=item_id,
            description=description,
            quantity=quantity,
            unit=unit,
            unit_price=unit_price,
            total_price=quantity * unit_price,
            spec_section=spec_section,
            csi_code=csi_code
        )

        co.line_items.append(item)

        # Update totals
        self._recalculate_totals(co)

        return item

    def set_cost_breakdown(self, co_id: str, labor: float = 0,
                          materials: float = 0, equipment: float = 0,
                          subcontractor: float = 0) -> CostBreakdown:
        """Set cost breakdown and calculate markups."""
        if co_id not in self.change_orders:
            raise ValueError(f"Change order {co_id} not found")

        co = self.change_orders[co_id]

        direct = labor + materials + equipment + subcontractor

        co.cost_breakdown = CostBreakdown(
            labor=labor,
            materials=materials,
            equipment=equipment,
            subcontractor=subcontractor,
            overhead=direct * self.markup_rates["overhead"],
            profit=direct * self.markup_rates["profit"],
            bond=direct * self.markup_rates["bond"]
        )

        co.proposed_amount = co.cost_breakdown.total

        return co.cost_breakdown

    def _recalculate_totals(self, co: ChangeOrder):
        """Recalculate change order totals from line items."""
        if co.line_items:
            direct_cost = sum(item.total_price for item in co.line_items)

            co.cost_breakdown.labor = direct_cost * 0.4  # Estimate
            co.cost_breakdown.materials = direct_cost * 0.4
            co.cost_breakdown.equipment = direct_cost * 0.1
            co.cost_breakdown.subcontractor = direct_cost * 0.1

            co.cost_breakdown.overhead = direct_cost * self