import pandas as pd
import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
from openai import OpenAI
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
@dataclass
class WorkItem:
"""Matched work item from CWICR"""
cwicr_code: str
description: str
unit: str
unit_price: float
labor_cost: float
material_cost: float
equipment_cost: float
productivity: float
currency: str
confidence: float
@dataclass
class CostLineItem:
"""Single line item in estimate"""
bim_type: str
work_item: WorkItem
quantity: float
quantity_unit: str
total_cost: float
labor_cost: float
material_cost: float
equipment_cost: float
phase: str
trade: str
class BIMCostEstimator:
"""BIM to cost estimation using DDC CWICR"""
def __init__(
self,
qdrant_url: str,
qdrant_api_key: str = None,
openai_api_key: str = None,
language: str = "EN"
):
self.qdrant = QdrantClient(url=qdrant_url, api_key=qdrant_api_key)
self.openai = OpenAI(api_key=openai_api_key)
self.language = language
self.collection = f"ddc_cwicr_{language.lower()}"
def get_embedding(self, text: str) -> List[float]:
"""Generate embedding for text"""
response = self.openai.embeddings.create(
model="text-embedding-3-large",
input=text,
dimensions=3072
)
return response.data[0].embedding
def search_cwicr(
self,
query: str,
limit: int = 5,
category_filter: str = None
) -> List[WorkItem]:
"""Search CWICR database for matching work items"""
query_vector = self.get_embedding(query)
query_filter = None
if category_filter:
query_filter = Filter(
must=[
FieldCondition(
key="category",
match=MatchValue(value=category_filter)
)
]
)
results = self.qdrant.search(
collection_name=self.collection,
query_vector=query_vector,
query_filter=query_filter,
limit=limit
)
work_items = []
for r in results:
payload = r.payload
work_items.append(WorkItem(
cwicr_code=payload.get('code', ''),
description=payload.get('description', ''),
unit=payload.get('unit', ''),
unit_price=float(payload.get('unit_price', 0)),
labor_cost=float(payload.get('labor_cost', 0)),
material_cost=float(payload.get('material_cost', 0)),
equipment_cost=float(payload.get('equipment_cost', 0)),
productivity=float(payload.get('productivity', 1)),
currency=payload.get('currency', 'USD'),
confidence=r.score
))
return work_items
def decompose_bim_type(
self,
bim_type: str,
category: str
) -> List[str]:
"""Use LLM to decompose BIM type into work items"""
prompt = f"""
Decompose this BIM element type into construction work items:
BIM Type: {bim_type}
Category: {category}
List the individual work activities needed to construct this element.
For example, "Brick Wall 240mm" decomposes into:
- Masonry: Brick laying
- Mortar: Cement mortar for joints
- Plaster: Internal plaster finish
- Paint: Wall painting
Return a JSON array of work item descriptions.
Example: ["Brick masonry laying", "Cement mortar for brick joints", "Internal cement plaster 15mm"]
"""
response = self.openai.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
try:
result = json.loads(response.choices[0].message.content)
return result.get('work_items', [bim_type])
except:
return [bim_type]
def estimate_element(
self,
bim_type: str,
category: str,
quantity: float,
quantity_unit: str,
phase: str = "Construction"
) -> List[CostLineItem]:
"""Estimate cost for single BIM element type"""
work_descriptions = self.decompose_bim_type(bim_type, category)
line_items = []
for work_desc in work_descriptions:
matches = self.search_cwicr(work_desc, limit=1)
if not matches:
continue
best_match = matches[0]
adjusted_qty = self._convert_units(
quantity, quantity_unit, best_match.unit
)
total = adjusted_qty * best_match.unit_price
labor = adjusted_qty * best_match.labor_cost
material = adjusted_qty * best_match.material_cost
equipment = adjusted_qty * best_match.equipment_cost
line_items.append(CostLineItem(
bim_type=bim_type,
work_item=best_match,
quantity=adjusted_qty,
quantity_unit=best_match.unit,
total_cost=total,
labor_cost=labor,
material_cost=material,
equipment_cost=equipment,
phase=phase,
trade=self._get_trade(category)
))
return line_items
def estimate_from_qto(
self,
qto_data: pd.DataFrame,
type_column: str = "Type Name",
category_column: str = "Category",
quantity_column: str = "Volume"
) -> List[CostLineItem]:
"""Generate estimate from QTO DataFrame"""
all_line_items = []
grouped = qto_data.groupby([category_column, type_column]).agg({
quantity_column: 'sum'
}).reset_index()
for _, row in grouped.iterrows():
items = self.estimate_element(
bim_type=row[type_column],
category=row[category_column],
quantity=row[quantity_column],
quantity_unit="m³"
)
all_line_items.extend(items)
return all_line_items
def _convert_units(
self,
value: float,
from_unit: str,
to_unit: str
) -> float:
"""Convert between units"""
conversions = {
('m³', 'm³'): 1.0,
('m²', 'm²'): 1.0,
('m', 'm'): 1.0,
('ft³', 'm³'): 0.0283168,
('ft²', 'm²'): 0.092903,
('ft', 'm'): 0.3048,
}
key = (from_unit.lower(), to_unit.lower())
factor = conversions.get(key, 1.0)
return value * factor
def _get_trade(self, category: str) -> str:
"""Map BIM category to trade"""
trade_map = {
'Walls': 'Masonry',
'Floors': 'Concrete',
'Structural Columns': 'Concrete',
'Structural Framing': 'Steel',
'Doors': 'Carpentry',
'Windows': 'Glazing',
'Plumbing Fixtures': 'Plumbing',
'Electrical Equipment': 'Electrical',
'Mechanical Equipment': 'HVAC'
}
return trade_map.get(category, 'General')
def generate_estimate_report(
self,
line_items: List[CostLineItem],
project_name: str,
output_path: str
) -> dict:
"""Generate comprehensive estimate report"""
records = []
for item in line_items:
records.append({
'BIM Type': item.bim_type,
'Work Item': item.work_item.description,
'CWICR Code': item.work_item.cwicr_code,
'Quantity': round(item.quantity, 2),
'Unit': item.quantity_unit,
'Unit Price': round(item.work_item.unit_price, 2),
'Labor': round(item.labor_cost, 2),
'Material': round(item.material_cost, 2),
'Equipment': round(item.equipment_cost, 2),
'Total': round(item.total_cost, 2),
'Phase': item.phase,
'Trade': item.trade,
'Currency': item.work_item.currency,
'Confidence': round(item.work_item.confidence, 2)
})
df = pd.DataFrame(records)
total_cost = df['Total'].sum()
total_labor = df['Labor'].sum()
total_material = df['Material'].sum()
total_equipment = df['Equipment'].sum()
by_trade = df.groupby('Trade')['Total'].sum().sort_values(ascending=False)
excel_path = f"{output_path}/{project_name}_Estimate.xlsx"
with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
summary_data = {
'Metric': ['Total Cost', 'Labor Cost', 'Material Cost', 'Equipment Cost'],
'Value': [total_cost, total_labor, total_material, total_equipment]
}
pd.DataFrame(summary_data).to_excel(writer, sheet_name='Summary', index=False)
by_trade.to_frame().to_excel(writer, sheet_name='By Trade')
df.to_excel(writer, sheet_name='Detail', index=False)
return {
'excel_path': excel_path,
'total_cost': total_cost,
'total_labor': total_labor,
'total_material': total_material,
'total_equipment': total_equipment,
'by_trade': by_trade.to_dict(),
'line_items': len(df),
'currency': line_items[0].work_item.currency if line_items else 'USD'
}
def estimate_from_bim_model(
model_path: str,
qdrant_url: str,
language: str = "EN",
output_dir: str = "."
) -> dict:
"""Complete BIM to cost estimation workflow"""
import subprocess
from pathlib import Path
print("Converting BIM model...")
subprocess.run([
r"C:\DDC\RvtExporter.exe",
model_path,
"complete", "bbox"
])
xlsx_path = Path(model_path).with_suffix('.xlsx')
print("Loading quantity data...")
df = pd.read_excel(xlsx_path)
estimator = BIMCostEstimator(
qdrant_url=qdrant_url,
language=language
)
print("Generating cost estimate...")
line_items = estimator.estimate_from_qto(df)
project_name = Path(model_path).stem
result = estimator.generate_estimate_report(
line_items=line_items,
project_name=project_name,
output_path=output_dir
)
print(f"\nEstimate Complete!")
print(f"Total Cost: {result['currency']} {result['total_cost']:,.2f}")
print(f"Excel Report: {result['excel_path']}")
return result
if __name__ == "__main__":
result = estimate_from_bim_model(
model_path=r"C:\Projects\Building.rvt",
qdrant_url="https://your-qdrant-instance.io",
language="DE",
output_dir=r"C:\Projects\Estimates"
)
