open-construction-estimate

name	open-construction-estimate
description	Access and utilize open construction pricing databases. Match BIM elements to standardized work items, calculate costs using public unit price databases with 55,000+ work items.

Open Construction Estimate

Overview

This skill leverages open construction pricing databases for automated cost estimation. Match project elements to standardized work items and calculate costs using publicly available unit prices.

Data Sources:

• OpenConstructionEstimate (55,000+ work items)
• RSMeans Online (subscription)
• Government pricing databases
• Regional cost indexes

"Открытые базы данных расценок содержат более 55,000 позиций работ, что позволяет автоматизировать сметные расчеты для большинства проектов." — DDC LinkedIn

Quick Start

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# Load work items database
work_items = pd.read_csv("open_construction_estimate.csv")
print(f"Loaded {len(work_items)} work items")

# Simple matching function
vectorizer = TfidfVectorizer(ngram_range=(1, 2))
item_vectors = vectorizer.fit_transform(work_items['description'])

def find_matching_items(query, top_n=5):
    query_vec = vectorizer.transform([query])
    similarities = cosine_similarity(query_vec, item_vectors)[0]
    top_indices = similarities.argsort()[-top_n:][::-1]

    return work_items.iloc[top_indices][['code', 'description', 'unit', 'unit_price']]

# Find matches
matches = find_matching_items("reinforced concrete wall 300mm")
print(matches)

Open Database Structure

Database Schema

# Standard work items database structure
WORK_ITEMS_SCHEMA = {
    'code': 'Work item code (e.g., 03.31.13.13)',
    'description': 'Full description of work',
    'short_description': 'Abbreviated description',
    'unit': 'Unit of measure (m³, m², ton, pcs)',
    'unit_price': 'Base unit price',
    'labor_cost': 'Labor component per unit',
    'material_cost': 'Material component per unit',
    'equipment_cost': 'Equipment component per unit',
    'labor_hours': 'Labor hours per unit',
    'crew_size': 'Typical crew size',
    'productivity': 'Units per day',
    'category_l1': 'Primary category (CSI Division)',
    'category_l2': 'Secondary category',
    'category_l3': 'Detailed category',
    'region': 'Geographic region',
    'year': 'Price year',
    'source': 'Data source'
}

# CSI MasterFormat Divisions
CSI_DIVISIONS = {
    '03': 'Concrete',
    '04': 'Masonry',
    '05': 'Metals',
    '06': 'Wood, Plastics, Composites',
    '07': 'Thermal and Moisture Protection',
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : ,
    : 
}

Work Item Matching Engine

Semantic Matching System

import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer
from typing import List, Dict, Optional, Tuple
import re

class WorkItemMatcher:
    """Match BIM elements to standardized work items"""

    def __init__(self, database_path: str, use_embeddings: bool = True):
        self.db = pd.read_csv(database_path)

        # TF-IDF for fast initial filtering
        self.tfidf = TfidfVectorizer(
            ngram_range=(1, 3),
            max_features=10000,
            stop_words='english'
        )
        self.tfidf_matrix = self.tfidf.fit_transform(self.db['description'])

        # Sentence embeddings for semantic matching
        self.use_embeddings = use_embeddings
        if use_embeddings:
            self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
            self.embeddings = self.embedder.encode(
                .db[].tolist(),
                show_progress_bar=
            )

     () -> []:
        
        
         category:
            mask = .db[]..contains(category, =, na=)
            search_db = .db[mask]
            search_matrix = .tfidf_matrix[mask]
        :
            search_db = .db
            search_matrix = .tfidf_matrix

         .use_embeddings:
             ._semantic_match(query, search_db, top_n)
        :
             ._tfidf_match(query, search_db, search_matrix, top_n)

     () -> []:
        
        query_vec = .tfidf.transform([query])
        similarities = cosine_similarity(query_vec, matrix)[]

        top_indices = similarities.argsort()[-top_n:][::-]

        results = []
         idx  top_indices:
            row = db.iloc[idx]
            results.append({
                : row[],
                : row[],
                : row[],
                : row[],
                : (similarities[idx]),
                : row.get(, )
            })

         results

     () -> []:
        
        query_embedding = .embedder.encode([query])

        
        indices = db.index.tolist()
        filtered_embeddings = .embeddings[indices]

        similarities = cosine_similarity(query_embedding, filtered_embeddings)[]
        top_indices = similarities.argsort()[-top_n:][::-]

        results = []
         i, idx  (top_indices):
            row = db.iloc[idx]
            results.append({
                : row[],
                : row[],
                : row[],
                : row[],
                : (similarities[idx]),
                : row.get(, )
            })

         results

     () -> []:
        
        
        query_parts = []

         element.get():
            query_parts.append(element[])
         element.get():
            query_parts.append(element[])
         element.get():
            query_parts.append(element[])

        
         element.get():
            query_parts.append()
         element.get():
            query_parts.append()

        query = .join(query_parts)

        
        category = ._get_category_from_element(element)

         .(query, top_n=, category=category)

     () -> []:
        
        element_mapping = {
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : 
        }

        elem_type = element.get(, )
         element_mapping.get(elem_type)

Cost Estimation Engine

Automated Estimator

class OpenConstructionEstimator:
    """Generate cost estimates using open databases"""

    def __init__(self, matcher: WorkItemMatcher, region: str = 'default'):
        self.matcher = matcher
        self.region = region
        self.regional_factors = self._load_regional_factors()
        self.estimates = []

    def _load_regional_factors(self) -> Dict[str, float]:
        """Load regional cost adjustment factors"""
        return {
            'default': 1.0,
            'northeast_us': 1.15,
            'southeast_us': 0.92,
            'midwest_us': 0.95,
            'west_us': 1.08,
            'moscow': 1.20,
            'spb': 1.10,
            'regions_ru': 0.85
        }

    def estimate_element(self, element: Dict) -> Dict:
        """Estimate cost for a single element"""
        # Get matching work items
        matches = self.matcher.match_bim_element(element)

        if  matches:
             {
                : element.get(),
                : ,
                : 
            }

        best_match = matches[]
        quantity = element.get(, )
        unit_price = best_match[]

        
        regional_factor = .regional_factors.get(.region, )
        adjusted_price = unit_price * regional_factor

        
        total_cost = adjusted_price * quantity

        estimate = {
            : element.get(),
            : element.get(),
            : element.get(, ),
            : best_match[],
            : best_match[],
            : best_match[],
            : best_match[],
            : quantity,
            : unit_price,
            : regional_factor,
            : adjusted_price,
            : total_cost
        }

        .estimates.append(estimate)
         estimate

     () -> :
        
         element  elements:
            .estimate_element(element)

        df = pd.DataFrame(.estimates)

        
          df.empty:
            summary = df.groupby().agg({
                : ,
                : ,
                : 
            }).rename(columns={: })
        :
            summary = pd.DataFrame()

        total = df[].()   df.empty  

         {
            : total,
            : (elements),
            : (df[df[] > ])   df.empty  ,
            : summary.to_dict()   summary.empty  {},
            : .estimates
        }

     () -> :
        
        df = pd.DataFrame(.estimates)

         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary = pd.DataFrame({
                : [, , , ],
                : [
                    df[].()   df.empty  ,
                    (df),
                    (df[df[] > ])   df.empty  ,
                    df[].mean()   df.empty  
                ]
            })
            summary.to_excel(writer, sheet_name=, index=)

            
              df.empty:
                df.to_excel(writer, sheet_name=, index=)

                
                by_type = df.groupby()[].()
                by_type.to_excel(writer, sheet_name=)

         output_path

     () -> []:
        
        df = pd.DataFrame(.estimates)
         df.empty:
             []

        low_confidence = df[df[] < ]
         low_confidence.to_dict()

Database Management

Creating and Updating Database

class OpenDatabaseManager:
    """Manage open construction pricing database"""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.db = self._load_or_create()

    def _load_or_create(self) -> pd.DataFrame:
        """Load existing or create new database"""
        try:
            return pd.read_csv(self.db_path)
        except FileNotFoundError:
            return pd.DataFrame(columns=list(WORK_ITEMS_SCHEMA.keys()))

    def add_items(self, items: List[Dict]):
        """Add new work items"""
        new_df = pd.DataFrame(items)
        self.db = pd.concat([self.db, new_df], ignore_index=True)
        self.db.drop_duplicates(subset=['code'], keep='last', inplace=True)

    def update_prices(self, updates: pd.DataFrame, year: int):
        """Update prices with new data"""
        for _, row in updates.iterrows():
            mask = self.db['code'] == row['code']
            if mask.():
                .db.loc[mask, ] = row[]
                .db.loc[mask, ] = year

     ():
        
        .db[] = .db[] * ( + rate)

     ():
        
        subset = .db[
            .db[]..contains(category, =, na=)
        ]
        subset.to_csv(output_path, index=)

     ():
        
        .db.to_csv(.db_path, index=)

     () -> :
        
         {
            : (.db),
            : .db[].nunique(),
            : .db[].mean(),
            : (.db[].(), .db[].()),
            : .db[].()    .db  
        }

Quick Reference

Category	CSI Division	Typical Items
Concrete	03	Walls, slabs, columns, beams
Masonry	04	Brick, block, stone
Metals	05	Structural steel, misc metals
Finishes	09	Drywall, paint, flooring
MEP	21-26	Plumbing, HVAC, electrical
Sitework	31-33	Excavation, paving, utilities

Resources

• OpenConstructionEstimate: Open database initiative
• CSI MasterFormat: https://www.csiresources.org/standards/masterformat
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See vector-search for semantic item matching
• See cost-prediction for ML-based estimation
• See qto-report for quantity extraction