Loading...
Loading...
Calculate construction costs using resource-based method. Estimate project costs from work items, physical resource norms, and current prices.
npx skill4agent add datadrivenconstruction/ddc_skills_for_ai_agents_in_construction cost-estimation-resourceimport pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class ResourceType(Enum):
LABOR = "labor"
MATERIAL = "material"
EQUIPMENT = "equipment"
SUBCONTRACTOR = "subcontractor"
@dataclass
class Resource:
code: str
name: str
resource_type: ResourceType
unit: str
unit_price: float
currency: str = "USD"
@dataclass
class ResourceNorm:
resource_code: str
consumption: float # Units per work item unit
waste_factor: float = 1.0 # 1.1 = 10% waste
@dataclass
class WorkItem:
code: str
name: str
unit: str
resources: List[ResourceNorm] = field(default_factory=list)
@dataclass
class CostLineItem:
work_item_code: str
work_item_name: str
quantity: float
unit: str
labor_cost: float
material_cost: float
equipment_cost: float
subcontractor_cost: float
total_cost: float
class ResourceBasedEstimator:
"""Calculate costs using resource-based method."""
def __init__(self):
self.resources: Dict[str, Resource] = {}
self.work_items: Dict[str, WorkItem] = {}
self.overhead_rate: float = 0.15
self.profit_rate: float = 0.10
def add_resource(self, resource: Resource):
"""Add resource to database."""
self.resources[resource.code] = resource
def add_work_item(self, work_item: WorkItem):
"""Add work item with resource norms."""
self.work_items[work_item.code] = work_item
def load_resources_from_df(self, df: pd.DataFrame):
"""Load resources from DataFrame."""
for _, row in df.iterrows():
resource = Resource(
code=row['code'],
name=row['name'],
resource_type=ResourceType(row['type'].lower()),
unit=row['unit'],
unit_price=float(row['unit_price']),
currency=row.get('currency', 'USD')
)
self.add_resource(resource)
def load_work_items_from_df(self, items_df: pd.DataFrame, norms_df: pd.DataFrame):
"""Load work items and norms from DataFrames."""
# Group norms by work item
norms_grouped = norms_df.groupby('work_item_code')
for _, row in items_df.iterrows():
code = row['code']
resources = []
if code in norms_grouped.groups:
item_norms = norms_grouped.get_group(code)
for _, norm_row in item_norms.iterrows():
resources.append(ResourceNorm(
resource_code=norm_row['resource_code'],
consumption=float(norm_row['consumption']),
waste_factor=float(norm_row.get('waste_factor', 1.0))
))
work_item = WorkItem(
code=code,
name=row['name'],
unit=row['unit'],
resources=resources
)
self.add_work_item(work_item)
def calculate_work_item_cost(self, work_item_code: str, quantity: float) -> CostLineItem:
"""Calculate cost for a work item quantity."""
if work_item_code not in self.work_items:
raise ValueError(f"Work item {work_item_code} not found")
work_item = self.work_items[work_item_code]
labor_cost = 0.0
material_cost = 0.0
equipment_cost = 0.0
subcontractor_cost = 0.0
for norm in work_item.resources:
if norm.resource_code not in self.resources:
continue
resource = self.resources[norm.resource_code]
resource_qty = quantity * norm.consumption * norm.waste_factor
resource_cost = resource_qty * resource.unit_price
if resource.resource_type == ResourceType.LABOR:
labor_cost += resource_cost
elif resource.resource_type == ResourceType.MATERIAL:
material_cost += resource_cost
elif resource.resource_type == ResourceType.EQUIPMENT:
equipment_cost += resource_cost
elif resource.resource_type == ResourceType.SUBCONTRACTOR:
subcontractor_cost += resource_cost
total = labor_cost + material_cost + equipment_cost + subcontractor_cost
return CostLineItem(
work_item_code=work_item_code,
work_item_name=work_item.name,
quantity=quantity,
unit=work_item.unit,
labor_cost=round(labor_cost, 2),
material_cost=round(material_cost, 2),
equipment_cost=round(equipment_cost, 2),
subcontractor_cost=round(subcontractor_cost, 2),
total_cost=round(total, 2)
)
def calculate_estimate(self, items: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Calculate full estimate from list of items."""
line_items = []
totals = {
'labor': 0.0,
'material': 0.0,
'equipment': 0.0,
'subcontractor': 0.0,
'direct': 0.0
}
for item in items:
code = item['work_item_code']
qty = float(item['quantity'])
line = self.calculate_work_item_cost(code, qty)
line_items.append(line)
totals['labor'] += line.labor_cost
totals['material'] += line.material_cost
totals['equipment'] += line.equipment_cost
totals['subcontractor'] += line.subcontractor_cost
totals['direct'] += line.total_cost
# Calculate overhead and profit
overhead = totals['direct'] * self.overhead_rate
subtotal = totals['direct'] + overhead
profit = subtotal * self.profit_rate
grand_total = subtotal + profit
return {
'line_items': line_items,
'totals': {
'labor': round(totals['labor'], 2),
'material': round(totals['material'], 2),
'equipment': round(totals['equipment'], 2),
'subcontractor': round(totals['subcontractor'], 2),
'direct_cost': round(totals['direct'], 2),
'overhead': round(overhead, 2),
'overhead_rate': self.overhead_rate,
'subtotal': round(subtotal, 2),
'profit': round(profit, 2),
'profit_rate': self.profit_rate,
'grand_total': round(grand_total, 2)
},
'summary': {
'item_count': len(line_items),
'labor_pct': round(totals['labor'] / totals['direct'] * 100, 1) if totals['direct'] > 0 else 0,
'material_pct': round(totals['material'] / totals['direct'] * 100, 1) if totals['direct'] > 0 else 0,
'equipment_pct': round(totals['equipment'] / totals['direct'] * 100, 1) if totals['direct'] > 0 else 0
}
}
def adjust_prices(self, factor: float, resource_type: ResourceType = None):
"""Adjust resource prices by factor."""
for code, resource in self.resources.items():
if resource_type is None or resource.resource_type == resource_type:
resource.unit_price *= factor
def apply_regional_factor(self, factor: float):
"""Apply regional cost factor to all resources."""
self.adjust_prices(factor)
def get_resource_breakdown(self, work_item_code: str, quantity: float) -> pd.DataFrame:
"""Get detailed resource breakdown for work item."""
if work_item_code not in self.work_items:
return pd.DataFrame()
work_item = self.work_items[work_item_code]
data = []
for norm in work_item.resources:
if norm.resource_code not in self.resources:
continue
resource = self.resources[norm.resource_code]
resource_qty = quantity * norm.consumption * norm.waste_factor
resource_cost = resource_qty * resource.unit_price
data.append({
'Resource Code': resource.code,
'Resource Name': resource.name,
'Type': resource.resource_type.value,
'Unit': resource.unit,
'Consumption': norm.consumption,
'Waste Factor': norm.waste_factor,
'Total Qty': round(resource_qty, 3),
'Unit Price': resource.unit_price,
'Total Cost': round(resource_cost, 2)
})
return pd.DataFrame(data)
def export_to_excel(self, estimate: Dict[str, Any], output_path: str) -> str:
"""Export estimate to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary_df = pd.DataFrame([estimate['totals']])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Line items
items_data = [{
'Code': item.work_item_code,
'Description': item.work_item_name,
'Quantity': item.quantity,
'Unit': item.unit,
'Labor': item.labor_cost,
'Material': item.material_cost,
'Equipment': item.equipment_cost,
'Subcontractor': item.subcontractor_cost,
'Total': item.total_cost
} for item in estimate['line_items']]
items_df = pd.DataFrame(items_data)
items_df.to_excel(writer, sheet_name='Line Items', index=False)
return output_path# Initialize estimator
estimator = ResourceBasedEstimator()
# Add resources
estimator.add_resource(Resource("L001", "Carpenter", ResourceType.LABOR, "MH", 55.00))
estimator.add_resource(Resource("L002", "Laborer", ResourceType.LABOR, "MH", 35.00))
estimator.add_resource(Resource("M001", "Concrete C30", ResourceType.MATERIAL, "CY", 150.00))
estimator.add_resource(Resource("M002", "Rebar #4", ResourceType.MATERIAL, "TON", 1200.00))
estimator.add_resource(Resource("E001", "Concrete Pump", ResourceType.EQUIPMENT, "HR", 250.00))
# Add work item with resource norms
estimator.add_work_item(WorkItem(
code="03.01.01",
name="Cast-in-place Concrete Foundation",
unit="CY",
resources=[
ResourceNorm("L001", 1.5), # 1.5 carpenter hours per CY
ResourceNorm("L002", 2.0), # 2.0 laborer hours per CY
ResourceNorm("M001", 1.0, 1.05),# 1.0 CY concrete with 5% waste
ResourceNorm("M002", 0.08), # 0.08 ton rebar per CY
ResourceNorm("E001", 0.25) # 0.25 pump hours per CY
]
))
# Calculate estimate
estimate = estimator.calculate_estimate([
{"work_item_code": "03.01.01", "quantity": 100}
])
print(f"Direct Cost: ${estimate['totals']['direct_cost']:,.2f}")
print(f"Grand Total: ${estimate['totals']['grand_total']:,.2f}")breakdown = estimator.get_resource_breakdown("03.01.01", quantity=100)
print(breakdown)# Apply 15% regional factor
estimator.apply_regional_factor(1.15)# Increase labor costs by 10%
estimator.adjust_prices(1.10, ResourceType.LABOR)