change-order-manager
skillManage construction change orders from request to approval. Track costs, schedule impacts, and maintain audit trail for dispute prevention.
apm::install
apm install @datadrivenconstruction/change-order-managerapm::skill.md
---
name: "change-order-manager"
description: "Manage construction change orders from request to approval. Track costs, schedule impacts, and maintain audit trail for dispute prevention."
homepage: "https://datadrivenconstruction.io"
metadata: {"openclaw": {"emoji": "📝", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}
---
# 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
```python
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.markup_rates["overhead"]
co.cost_breakdown.profit = direct_cost * self.markup_rates["profit"]
co.cost_breakdown.bond = direct_cost * self.markup_rates["bond"]
co.proposed_amount = co.cost_breakdown.total
def submit_change_order(self, co_id: str) -> ChangeOrder:
"""Submit change order for review."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
co.status = ChangeOrderStatus.SUBMITTED
co.submitted_date = datetime.now()
self._add_note(co, "Submitted for review")
return co
def approve_change_order(self, co_id: str, approved_amount: float,
approved_time: int = 0) -> ChangeOrder:
"""Approve 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]
co.status = ChangeOrderStatus.APPROVED
co.approved_date = datetime.now()
co.approved_amount = approved_amount
co.approved_time_days = approved_time
self._add_note(co, f"Approved: ${approved_amount:,.2f}, {approved_time} days")
return co
def reject_change_order(self, co_id: str, reason: str) -> ChangeOrder:
"""Reject 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]
co.status = ChangeOrderStatus.REJECTED
self._add_note(co, f"Rejected: {reason}")
return co
def execute_change_order(self, co_id: str) -> ChangeOrder:
"""Mark change order as executed."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
if co.status != ChangeOrderStatus.APPROVED:
raise ValueError("Change order must be approved before execution")
co.status = ChangeOrderStatus.EXECUTED
co.executed_date = datetime.now()
self._add_note(co, "Executed")
return co
def _add_note(self, co: ChangeOrder, text: str):
"""Add note to change order."""
co.notes.append({
"timestamp": datetime.now().isoformat(),
"text": text
})
def add_reference(self, co_id: str, ref_type: str, reference: str):
"""Add reference document 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]
if ref_type == "rfi":
co.rfi_references.append(reference)
elif ref_type == "drawing":
co.drawing_references.append(reference)
elif ref_type == "photo":
co.photo_attachments.append(reference)
elif ref_type == "backup":
co.backup_documents.append(reference)
def get_summary(self) -> Dict:
"""Get change order summary statistics."""
total_approved = sum(
co.approved_amount for co in self.change_orders.values()
if co.status in [ChangeOrderStatus.APPROVED, ChangeOrderStatus.EXECUTED]
)
total_pending = sum(
co.proposed_amount for co in self.change_orders.values()
if co.status in [ChangeOrderStatus.SUBMITTED, ChangeOrderStatus.UNDER_REVIEW,
ChangeOrderStatus.PRICING, ChangeOrderStatus.NEGOTIATING]
)
by_type = {}
for co in self.change_orders.values():
t = co.change_type.value
by_type[t] = by_type.get(t, 0) + (co.approved_amount or co.proposed_amount)
by_status = {}
for co in self.change_orders.values():
s = co.status.value
by_status[s] = by_status.get(s, 0) + 1
return {
"original_contract": self.original_contract,
"total_approved": total_approved,
"total_pending": total_pending,
"revised_contract": self.original_contract + total_approved,
"change_order_count": len(self.change_orders),
"change_percentage": (total_approved / self.original_contract * 100) if self.original_contract else 0,
"by_type": by_type,
"by_status": by_status
}
def generate_log(self) -> str:
"""Generate change order log."""
summary = self.get_summary()
lines = [
"# Change Order Log",
"",
f"**Project:** {self.project_name}",
f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
"",
"## Summary",
"",
f"| Metric | Amount |",
f"|--------|--------|",
f"| Original Contract | ${summary['original_contract']:,.2f} |",
f"| Approved Changes | ${summary['total_approved']:,.2f} |",
f"| Pending Changes | ${summary['total_pending']:,.2f} |",
f"| **Revised Contract** | **${summary['revised_contract']:,.2f}** |",
f"| Change % | {summary['change_percentage']:.1f}% |",
"",
"## Change Orders",
"",
"| # | Title | Type | Status | Proposed | Approved | Time |",
"|---|-------|------|--------|----------|----------|------|"
]
for co in sorted(self.change_orders.values(), key=lambda x: x.number):
lines.append(
f"| {co.number} | {co.title[:30]} | {co.change_type.value} | "
f"{co.status.value} | ${co.proposed_amount:,.0f} | "
f"${co.approved_amount:,.0f} | {co.approved_time_days}d |"
)
return "\n".join(lines)
def generate_co_document(self, co_id: str) -> str:
"""Generate formal change order document."""
if co_id not in self.change_orders:
return "Change order not found"
co = self.change_orders[co_id]
lines = [
f"# CHANGE ORDER NO. {co.number}",
"",
f"**Project:** {self.project_name}",
f"**Change Order ID:** {co.id}",
f"**Date:** {co.submitted_date.strftime('%Y-%m-%d') if co.submitted_date else 'Draft'}",
"",
"---",
"",
f"## Description of Change",
"",
co.description,
"",
f"**Type:** {co.change_type.value.replace('_', ' ').title()}",
"",
]
if co.line_items:
lines.extend([
"## Schedule of Values",
"",
"| Item | Description | Qty | Unit | Unit Price | Total |",
"|------|-------------|-----|------|------------|-------|"
])
for item in co.line_items:
lines.append(
f"| {item.id} | {item.description} | {item.quantity} | "
f"{item.unit} | ${item.unit_price:,.2f} | ${item.total_price:,.2f} |"
)
lines.append("")
lines.extend([
"## Cost Summary",
"",
f"| Category | Amount |",
f"|----------|--------|",
f"| Labor | ${co.cost_breakdown.labor:,.2f} |",
f"| Materials | ${co.cost_breakdown.materials:,.2f} |",
f"| Equipment | ${co.cost_breakdown.equipment:,.2f} |",
f"| Subcontractor | ${co.cost_breakdown.subcontractor:,.2f} |",
f"| Overhead | ${co.cost_breakdown.overhead:,.2f} |",
f"| Profit | ${co.cost_breakdown.profit:,.2f} |",
f"| Bond | ${co.cost_breakdown.bond:,.2f} |",
f"| **Total** | **${co.cost_breakdown.total:,.2f}** |",
"",
f"## Time Impact",
"",
f"Proposed Extension: **{co.proposed_time_days} days**",
f"Critical Path Impact: {'Yes' if co.impacts_critical_path else 'No'}",
"",
])
if co.rfi_references:
lines.extend([
"## References",
"",
f"- RFIs: {', '.join(co.rfi_references)}",
f"- Drawings: {', '.join(co.drawing_references)}" if co.drawing_references else "",
])
return "\n".join(lines)
```
## Quick Start
```python
# Initialize manager
manager = ChangeOrderManager(
project_id="PRJ-001",
project_name="Office Tower",
original_contract=5000000.0
)
# Create change order
co = manager.create_change_order(
title="Additional Structural Steel",
description="Add steel reinforcement at Level 5 per RFI-042",
change_type=ChangeType.DESIGN_ERROR,
created_by="Project Manager"
)
# Add line items
manager.add_line_item(
co.id,
"W12x26 Steel Beam",
quantity=450,
unit="LF",
unit_price=85.00,
csi_code="05 12 00"
)
# Or set cost breakdown directly
manager.set_cost_breakdown(
co.id,
labor=15000,
materials=25000,
equipment=2000,
subcontractor=5000
)
# Add references
manager.add_reference(co.id, "rfi", "RFI-042")
manager.add_reference(co.id, "drawing", "S-501 Rev 2")
# Submit for approval
manager.submit_change_order(co.id)
# Approve (with negotiated amount)
manager.approve_change_order(co.id, approved_amount=50000, approved_time=5)
# Generate documents
print(manager.generate_co_document(co.id))
print(manager.generate_log())
```
## Requirements
```bash
pip install (no external dependencies)
```