cwicr-data-loader

name	cwicr-data-loader
description	Load and parse DDC CWICR construction cost database from multiple formats: Parquet, Excel, CSV, Qdrant snapshots. Foundation for all CWICR operations.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🗄️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"],"env":["QDRANT_URL"]},"primaryEnv":"QDRANT_URL"}}

CWICR Data Loader

Business Case

Problem Statement

DDC CWICR database is distributed in multiple formats:

• Apache Parquet (optimized for analytics)
• Excel workbooks (human-readable)
• CSV files (universal exchange)
• Qdrant snapshots (vector search)

Applications need unified data access regardless of source format.

Solution

Universal data loader supporting all CWICR formats with automatic schema detection, validation, and pandas DataFrame conversion.

Business Value

• Format agnostic - Load from any CWICR distribution
• Validated data - Automatic schema validation
• Memory efficient - Lazy loading for large datasets
• Type-safe - Proper data types preserved

Technical Implementation

Prerequisites

pip install pandas pyarrow openpyxl qdrant-client

Python Implementation

import pandas as pd
import pyarrow.parquet as pq
from pathlib import Path
from typing import Optional, Dict, Any, List, Union
from dataclasses import dataclass, field
from enum import Enum
import json


class CWICRFormat(Enum):
    """Supported CWICR data formats."""
    PARQUET = "parquet"
    EXCEL = "excel"
    CSV = "csv"
    QDRANT = "qdrant"
    JSON = "json"


class CWICRLanguage(Enum):
    """Supported languages in CWICR database."""
    ARABIC = "ar"
    CHINESE = "zh"
    GERMAN = "de"
    ENGLISH = "en"
    SPANISH = "es"
    FRENCH = "fr"
    HINDI = "hi"
    PORTUGUESE = "pt"
    RUSSIAN = "ru"


@dataclass
class CWICRSchema:
    """CWICR database schema definition."""

    # Core fields
    work_item_code: str = "work_item_code"
    description: str = "description"
    unit:  = 
    category:  = 

    
    unit_price:  = 
    labor_cost:  = 
    material_cost:  = 
    equipment_cost:  = 
    overhead_cost:  = 

    
    labor_norm:  = 
    material_norm:  = 
    equipment_norm:  = 

    
    language:  = 
    region:  = 
    currency:  = 
    last_updated:  = 

    
    embedding:  = 



 :
    
    work_item_code: 
    description: 
    unit: 
    category: 

    unit_price:  = 
    labor_cost:  = 
    material_cost:  = 
    equipment_cost:  = 
    overhead_cost:  = 

    labor_norm:  = 
    labor_unit:  = 

    resources: [[, ]] = field(default_factory=)

    language:  = 
    region:  = 
    currency:  = 



 :
    
    resource_code: 
    description: 
    unit: 
    unit_price: 
    resource_type:   
    category:  = 


 :
    

    REQUIRED_COLUMNS = [, , ]
    NUMERIC_COLUMNS = [, , ,
                       , ]

     ():
        .schema = CWICRSchema()
        ._cache: [, pd.DataFrame] = {}

     () -> pd.DataFrame:
        

        cache_key = 
         use_cache  cache_key  ._cache:
             ._cache[cache_key]

        
           :
             = ._detect_format(source)

        
          == CWICRFormat.PARQUET:
            df = ._load_parquet(source)
          == CWICRFormat.EXCEL:
            df = ._load_excel(source)
          == CWICRFormat.CSV:
            df = ._load_csv(source)
          == CWICRFormat.JSON:
            df = ._load_json(source)
        :
             ValueError()

        
        df = ._validate_schema(df)
        df = ._normalize_types(df)

        
         language    df.columns:
            df = df[df[] == language.value]

        
         use_cache:
            ._cache[cache_key] = df

         df

     () -> CWICRFormat:
        
        path = Path(source)

         path.suffix.lower() == :
             CWICRFormat.PARQUET
         path.suffix.lower()  [, ]:
             CWICRFormat.EXCEL
         path.suffix.lower() == :
             CWICRFormat.CSV
         path.suffix.lower() == :
             CWICRFormat.JSON
        :
             ValueError()

     () -> pd.DataFrame:
        
         pd.read_parquet(source)

     () -> pd.DataFrame:
        
        :
             pd.read_excel(source, sheet_name=sheet_name)
        :
            
             pd.read_excel(source, sheet_name=)

     () -> pd.DataFrame:
        
        
         encoding  [, , ]:
            :
                 pd.read_csv(source, encoding=encoding)
             UnicodeDecodeError:
                
         ValueError()

     () -> pd.DataFrame:
        
         (source, , encoding=)  f:
            data = json.load(f)

         (data, ):
             pd.DataFrame(data)
         (data, )    data:
             pd.DataFrame(data[])
        :
             pd.DataFrame([data])

     () -> pd.DataFrame:
        
        
        missing = (.REQUIRED_COLUMNS) - (df.columns)
         missing:
             ValueError()

         df

     () -> pd.DataFrame:
        
         col  .NUMERIC_COLUMNS:
             col  df.columns:
                df[col] = pd.to_numeric(df[col], errors=).fillna()

        
         col  [, , , ]:
             col  df.columns:
                df[col] = df[col].astype()

         df

     () -> pd.DataFrame:
        
           :
             = ._detect_format(source)

          == CWICRFormat.EXCEL:
            :
                 pd.read_excel(source, sheet_name=)
            :
                 pd.DataFrame()
        :
             .load(source, )

     () -> [CWICRWorkItem]:
        
        item = df[df[] == code]
         item.empty:
             

        row = item.iloc[]
         CWICRWorkItem(
            work_item_code=row[],
            description=row.get(, ),
            unit=row.get(, ),
            category=row.get(, ),
            unit_price=row.get(, ),
            labor_cost=row.get(, ),
            material_cost=row.get(, ),
            equipment_cost=row.get(, ),
            labor_norm=row.get(, ),
            language=row.get(, ),
            region=row.get(, ),
            currency=row.get(, )
        )

     () -> []:
        
            df.columns:
             []
         df[].dropna().unique().tolist()

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

     () -> pd.DataFrame:
        
         case_sensitive:
             df[df[]..contains(keyword, na=)]
         df[df[]..contains(keyword, =, na=)]

     () -> [, ]:
        
        stats = {
            : (df),
            : df[].nunique()    df.columns  ,
            : df[].unique().tolist()    df.columns  []
        }

           df.columns:
            stats[] = {
                : df[].(),
                : df[].(),
                : df[].mean()
            }

         stats

     ():
        
          == CWICRFormat.PARQUET:
            df.to_parquet(output_path, index=)
          == CWICRFormat.EXCEL:
            df.to_excel(output_path, index=)
          == CWICRFormat.CSV:
            df.to_csv(output_path, index=)
          == CWICRFormat.JSON:
            df.to_json(output_path, orient=, indent=)


 :
    

     ():
        .loader = CWICRDataLoader()

     () -> pd.DataFrame:
        
        dfs = []
         source  sources:
            :
                df = .loader.load(source)
                dfs.append(df)
             Exception  e:
                ()

          dfs:
             pd.DataFrame()

         pd.concat(dfs, ignore_index=)

     () -> pd.DataFrame:
        
        path = Path(base_path)
        dfs = []

         lang  CWICRLanguage:
            
            patterns = [
                ,
                ,
                
            ]

             pattern  patterns:
                files = (path.glob(pattern))
                 file  files:
                    :
                        df = .loader.load((file), language=lang)
                        dfs.append(df)
                     Exception  e:
                        

          dfs:
             pd.DataFrame()

         pd.concat(dfs, ignore_index=)



 () -> pd.DataFrame:
    
    loader = CWICRDataLoader()
    lang = CWICRLanguage(language)  language  
     loader.load(source, language=lang)


 () -> [, ]:
    
    loader = CWICRDataLoader()
    df = loader.load(source)
     loader.get_statistics(df)

Quick Start

# Load from Parquet (fastest)
loader = CWICRDataLoader()
df = loader.load("TR_workitems_costs_resources_DDC_CWICR.parquet")
print(f"Loaded {len(df)} work items")

# Load from Excel
df = loader.load("cwicr_database.xlsx")

# Get specific work item
item = loader.get_work_item(df, "CONC-001")
print(f"{item.description}: ${item.unit_price} per {item.unit}")

# Get all categories
categories = loader.get_categories(df)
print(f"Categories: {categories}")

Common Use Cases

1. Multi-Language Loading

batch = CWICRBatchLoader()
all_languages = batch.load_all_languages("C:/CWICR/")
print(f"Total items across all languages: {len(all_languages)}")

2. Category Filtering

loader = CWICRDataLoader()
df = loader.load("cwicr.parquet")

# Get concrete work items
concrete = loader.filter_by_category(df, "Concrete")
print(f"Concrete items: {len(concrete)}")

3. Keyword Search

# Find all masonry-related items
masonry = loader.search_by_description(df, "masonry")
print(masonry[['work_item_code', 'description', 'unit_price']])

Database Statistics

stats = loader.get_statistics(df)
print(f"Total items: {stats['total_work_items']}")
print(f"Categories: {stats['categories']}")
print(f"Price range: ${stats['price_range']['min']:.2f} - ${stats['price_range']['max']:.2f}")

Resources

• GitHub: OpenConstructionEstimate-DDC-CWICR
• Downloads: CWICR Releases
• Formats: Parquet, Excel, CSV, Qdrant snapshots