CWICR Assembly Builder
Business Case
Problem Statement
Estimating repetitive elements requires:
- Consistent item groupings
- Reusable templates
- Standard assemblies
- Quick application
Solution
Build and manage assemblies of CWICR work items that can be applied as templates to speed up estimating and ensure completeness.
Business Value
- Speed - Apply complete assemblies quickly
- Consistency - Standard item groupings
- Completeness - No missed items
- Reusability - Template library
Technical Implementation
import pandas as pd
import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
class AssemblyType(Enum):
"""Types of assemblies."""
STRUCTURAL = "structural"
ARCHITECTURAL = "architectural"
MECHANICAL = "mechanical"
ELECTRICAL = "electrical"
SITEWORK = "sitework"
GENERAL = "general"
@dataclass
class AssemblyItem:
"""Single item in assembly."""
work_item_code: str
description: str
quantity_per_unit: float # Quantity per assembly unit
unit: str
unit_cost: float
total_cost: float
notes: str = ""
@dataclass
class Assembly:
"""Complete assembly definition."""
assembly_code: str
name: str
description: str
assembly_type: AssemblyType
unit: str # Assembly unit (e.g., "m2", "each", "LF")
items: List[AssemblyItem]
total_cost_per_unit: float
labor_hours_per_unit: float
created_date: datetime
version: int = 1
class CWICRAssemblyBuilder:
"""Build and manage assemblies from CWICR data."""
def __init__(self, cwicr_data: pd.DataFrame):
self.cwicr = cwicr_data
self._index_cwicr()
self._assemblies: Dict[str, Assembly] = {}
def _index_cwicr(self):
"""Index CWICR data."""
if 'work_item_code' in self.cwicr.columns:
self._cwicr_index = self.cwicr.set_index('work_item_code')
else:
self._cwicr_index = None
def _get_item_cost(self, code: str) -> Tuple[float, float, str]:
"""Get item unit cost and labor hours."""
if self._cwicr_index is None or code not in self._cwicr_index.index:
return (0, 0, 'unit')
item = self._cwicr_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)
labor_hours = float(item.get('labor_norm', item.get('labor_hours', 0)) or 0)
unit = str(item.get('unit', 'unit'))
return (labor + material + equipment, labor_hours, unit)
def create_assembly(self,
assembly_code: str,
name: str,
description: str,
assembly_type: AssemblyType,
unit: str,
items: List[Dict[str, Any]]) -> Assembly:
"""Create new assembly from work items."""
assembly_items = []
total_cost = 0
total_hours = 0
for item_def in items:
code = item_def.get('work_item_code', item_def.get('code'))
qty_per_unit = item_def.get('quantity_per_unit', 1)
notes = item_def.get('notes', '')
unit_cost, labor_hours, item_unit = self._get_item_cost(code)
# Get description from CWICR
if self._cwicr_index is not None and code in self._cwicr_index.index:
desc = str(self._cwicr_index.loc[code].get('description', code))
else:
desc = item_def.get('description', code)
item_total = unit_cost * qty_per_unit
assembly_items.append(AssemblyItem(
work_item_code=code,
description=desc,
quantity_per_unit=qty_per_unit,
unit=item_unit,
unit_cost=round(unit_cost, 2),
total_cost=round(item_total, 2),
notes=notes
))
total_cost += item_total
total_hours += labor_hours * qty_per_unit
assembly = Assembly(
assembly_code=assembly_code,
name=name,
description=description,
assembly_type=assembly_type,
unit=unit,
items=assembly_items,
total_cost_per_unit=round(total_cost, 2),
labor_hours_per_unit=round(total_hours, 2),
created_date=datetime.now(),
version=1
)
self._assemblies[assembly_code] = assembly
return assembly
def apply_assembly(self,
assembly_code: str,
quantity: float,
location_factor: float = 1.0) -> Dict[str, Any]:
"""Apply assembly to get estimate."""
assembly = self._assemblies.get(assembly_code)
if assembly is None:
return {'error': f"Assembly {assembly_code} not found"}
items = []
total_cost = 0
total_hours = 0
for item in assembly.items:
qty = item.quantity_per_unit * quantity
cost = item.total_cost * quantity * location_factor
hours = qty * (item.unit_cost / 50 if item.unit_cost > 0 else 0) # Approximate labor hours
items.append({
'work_item_code': item.work_item_code,
'description': item.description,
'quantity': round(qty, 2),
'unit': item.unit,
'cost': round(cost, 2)
})
total_cost += cost
total_hours += hours
return {
'assembly_code': assembly_code,
'assembly_name': assembly.name,
'quantity': quantity,
'unit': assembly.unit,
'location_factor': location_factor,
'items': items,
'total_cost': round(total_cost, 2),
'total_labor_hours': round(total_hours, 2),
'cost_per_unit': round(total_cost / quantity, 2) if quantity > 0 else 0
}
def get_assembly(self, assembly_code: str) -> Optional[Assembly]:
"""Get assembly by code."""
return self._assemblies.get(assembly_code)
def list_assemblies(self, assembly_type: AssemblyType = None) -> List[Dict[str, Any]]:
"""List all assemblies."""
assemblies = self._assemblies.values()
if assembly_type:
assemblies = [a for a in assemblies if a.assembly_type == assembly_type]
return [
{
'code': a.assembly_code,
'name': a.name,
'type': a.assembly_type.value,
'unit': a.unit,
'cost_per_unit': a.total_cost_per_unit,
'item_count': len(a.items)
}
for a in assemblies
]
def clone_assembly(self,
source_code: str,
new_code: str,
new_name: str = None) -> Optional[Assembly]:
"""Clone existing assembly."""
source = self._assemblies.get(source_code)
if source is None:
return None
new_assembly = Assembly(
assembly_code=new_code,
name=new_name or f"{source.name} (Copy)",
description=source.description,
assembly_type=source.assembly_type,
unit=source.unit,
items=source.items.copy(),
total_cost_per_unit=source.total_cost_per_unit,
labor_hours_per_unit=source.labor_hours_per_unit,
created_date=datetime.now(),
version=1
)
self._assemblies[new_code] = new_assembly
return new_assembly
def compare_assemblies(self,
codes: List[str],
quantity: float = 1) -> pd.DataFrame:
"""Compare mu