CWICR Material Substitution
Business Case
Problem Statement
Material substitution challenges:
- Supply chain issues
- Cost optimization
- Specification compliance
- Equivalent performance
Solution
Systematic material substitution using CWICR data to find functionally equivalent alternatives with cost and performance analysis.
Business Value
- Supply flexibility - Alternative sources
- Cost savings - Lower-cost equivalents
- Compliance - Specification matching
- Quick decisions - Rapid alternative search
Technical Implementation
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
from difflib import SequenceMatcher
class SubstitutionType(Enum):
"""Types of substitution."""
DIRECT = "direct" # Drop-in replacement
EQUIVALENT = "equivalent" # Same function, different material
UPGRADE = "upgrade" # Better performance
DOWNGRADE = "downgrade" # Lower performance (cost saving)
class CompatibilityLevel(Enum):
"""Compatibility levels."""
EXACT = "exact" # Identical specs
HIGH = "high" # Minor differences
MEDIUM = "medium" # Requires review
LOW = "low" # Significant differences
@dataclass
class MaterialSubstitute:
"""Material substitution option."""
original_code: str
original_description: str
substitute_code: str
substitute_description: str
substitution_type: SubstitutionType
compatibility: CompatibilityLevel
original_cost: float
substitute_cost: float
cost_difference: float
cost_difference_pct: float
notes: str
# Material compatibility groups
MATERIAL_GROUPS = {
'concrete': ['cement', 'beton', 'concrete', 'C20', 'C25', 'C30', 'C35', 'C40'],
'steel': ['steel', 'rebar', 'reinforcement', 'S235', 'S275', 'S355'],
'lumber': ['wood', 'timber', 'lumber', 'plywood', 'OSB'],
'masonry': ['brick', 'block', 'CMU', 'masonry'],
'insulation': ['insulation', 'rockwool', 'glasswool', 'EPS', 'XPS', 'PIR'],
'pipe': ['pipe', 'PVC', 'HDPE', 'copper', 'steel pipe'],
'electrical': ['wire', 'cable', 'conduit'],
'finishing': ['paint', 'plaster', 'drywall', 'gypsum'],
'flooring': ['tile', 'vinyl', 'laminate', 'carpet', 'hardwood'],
'roofing': ['shingle', 'membrane', 'metal roof', 'tile roof']
}
class CWICRMaterialSubstitution:
"""Find material substitutions using CWICR data."""
def __init__(self, cwicr_data: pd.DataFrame):
self.materials = cwicr_data
self._index_data()
def _index_data(self):
"""Index material data."""
if 'work_item_code' in self.materials.columns:
self._code_index = self.materials.set_index('work_item_code')
elif 'material_code' in self.materials.columns:
self._code_index = self.materials.set_index('material_code')
else:
self._code_index = None
def _similarity(self, a: str, b: str) -> float:
"""Calculate string similarity."""
return SequenceMatcher(None, a.lower(), b.lower()).ratio()
def _get_material_group(self, description: str) -> Optional[str]:
"""Identify material group from description."""
desc_lower = description.lower()
for group, keywords in MATERIAL_GROUPS.items():
if any(kw.lower() in desc_lower for kw in keywords):
return group
return None
def _get_cost(self, code: str) -> Tuple[float, str]:
"""Get material cost."""
if self._code_index is None or code not in self._code_index.index:
return (0, 'unit')
item = self._code_index.loc[code]
cost = float(item.get('material_cost', item.get('total_cost', 0)) or 0)
unit = str(item.get('unit', 'unit'))
return (cost, unit)
def find_substitutes(self,
material_code: str,
max_results: int = 10,
max_cost_increase: float = 0.20,
include_upgrades: bool = True) -> List[MaterialSubstitute]:
"""Find substitute materials."""
if self._code_index is None or material_code not in self._code_index.index:
return []
original = self._code_index.loc[material_code]
original_desc = str(original.get('description', material_code))
original_cost, original_unit = self._get_cost(material_code)
group = self._get_material_group(original_desc)
substitutes = []
for code, row in self._code_index.iterrows():
if code == material_code:
continue
sub_desc = str(row.get('description', code))
sub_group = self._get_material_group(sub_desc)
# Check if same group or similar description
if group and sub_group == group:
similarity = 0.7
else:
similarity = self._similarity(original_desc, sub_desc)
if similarity < 0.3:
continue
sub_cost, sub_unit = self._get_cost(code)
if sub_unit != original_unit:
continue
cost_diff = sub_cost - original_cost
cost_diff_pct = (cost_diff / original_cost * 100) if original_cost > 0 else 0
# Filter by cost increase limit
if not include_upgrades and cost_diff_pct > max_cost_increase * 100:
continue
# Determine substitution type
if cost_diff_pct < -10:
sub_type = SubstitutionType.DOWNGRADE
elif cost_diff_pct > 10:
sub_type = SubstitutionType.UPGRADE
elif similarity > 0.8:
sub_type = SubstitutionType.DIRECT
else:
sub_type = SubstitutionType.EQUIVALENT
# Determine compatibility
if similarity > 0.9:
compat = CompatibilityLevel.EXACT
elif similarity > 0.7:
compat = CompatibilityLevel.HIGH
elif similarity > 0.5:
compat = CompatibilityLevel.MEDIUM
else:
compat = CompatibilityLevel.LOW
substitutes.append(MaterialSubstitute(
original_code=material_code,
original_description=original_desc,
substitute_code=code,
substitute_description=sub_desc,
substitution_type=sub_type,
compatibility=compat,
original_cost=round(original_cost, 2),
substitute_cost=round(sub_cost, 2),
cost_difference=round(cost_diff, 2),
cost_difference_pct=round(cost_diff_pct, 1),
notes=f"Similarity: {similarity:.0%}"
))
# Sort by compatibility then cost
substitutes.sort(key=lambda x: (
list(CompatibilityLevel).index(x.compatibility),
x.cost_difference
))
return substitutes[:max_results]
def find_cost_saving_alternatives(self,
material_code: str,
min_savings_pct: float = 5.0) -> List[MaterialSubstitute]:
"""Find lower-cost alternatives."""
subs = self.find_substitutes(material_code, max_results=20)
cost_saving = [
s for s in subs
if s.cost_difference_pct <= -min_savings_pct
]
return sorted(cost_saving, key=lambda x: x.cost_difference)
def find_by_group(self,
group_name: str,
max_results: int = 20) -> List[Dict[str, Any]]:
"""Find all materials in a group."""
if self._code_index is None:
return []
results = []
for code, row in self._code_index.iterrows():
desc = str(row.get('description', code))