data-quality-check

name	data-quality-check
description	Assess construction data quality using completeness, accuracy, consistency, timeliness, and validity metrics. Automated validation with regex patterns, thresholds, and reporting.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"✔️","os":["win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Data Quality Check for Construction

Overview

Based on DDC methodology (Chapter 2.6), this skill provides comprehensive data quality assessment for construction projects. Poor data quality leads to poor decisions - validate early, validate often.

Book Reference: "Требования к качеству данных и его обеспечение" / "Data Quality Requirements"

"Качество данных определяется пятью ключевыми метриками: полнота, точность, согласованность, своевременность и достоверность." — DDC Book, Chapter 2.6

Quick Start

import pandas as pd

# Load construction data
df = pd.read_excel("bim_export.xlsx")

# Quick quality check
quality_score = {
    'completeness': (1 - df.isnull().sum().sum() / df.size) * 100,
    'unique_ids': df['ElementId'].nunique() == len(df),
    'valid_volumes': (df['Volume_m3'] >= 0).all()
}

print(f"Completeness: {quality_score['completeness']:.1f}%")
print(f"Unique IDs: {quality_score['unique_ids']}")
print(f"Valid volumes: {quality_score['valid_volumes']}")

Data Quality Dimensions

The 5 Quality Metrics

import pandas as pd
import numpy as np
import re
from datetime import datetime, timedelta

class DataQualityChecker:
    """Comprehensive data quality assessment for construction data"""

    def __init__(self, df):
        self.df = df.copy()
        self.results = {}
        self.issues = []

    def check_completeness(self, required_columns=None):
        """Check for missing values (Полнота)"""
        if required_columns is None:
            required_columns = self.df.columns.tolist()

        completeness = {}
        for col in required_columns:
            if col in self.df.columns:
                non_null = self.df[col].notna().sum()
                total = len(self.df)
                completeness[col] = (non_null / total) * 100
            else:
                completeness[col] = 0
                self.issues.append(f"Missing required column: {col}")

        overall = np.mean(list(completeness.values()))

        self.results['completeness'] = {
            : completeness,
            : overall,
            : ,
            : overall >= 
        }

         .results[]

     ():
        
         rules  :
            
            rules = {
                : {: , : },
                : {: , : },
                : {: , : },
                : {: , : }
            }

        accuracy = {}
         col, bounds  rules.items():
             col  .df.columns:
                valid = .df[col].between(
                    bounds.get(, -np.inf),
                    bounds.get(, np.inf)
                ).()
                total = .df[col].notna().()
                accuracy[col] = (valid / total * )  total >   

                
                invalid_count = total - valid
                 invalid_count > :
                    .issues.append(
                        
                    )

        overall = np.mean((accuracy.values()))  accuracy  

        .results[] = {
            : accuracy,
            : overall,
            : ,
            : overall >= 
        }

         .results[]

     ():
        
        consistency = {}

        
         unique_cols  :
            unique_cols = []

         col  unique_cols:
             col  .df.columns:
                is_unique = .df[col].nunique() == (.df)
                consistency[] =   is_unique  \
                    (.df[col].nunique() / (.df) * )

                  is_unique:
                    duplicates = .df[.df[col].duplicated()][col].unique()
                    .issues.append()

        
         relationship_rules  :
            relationship_rules = [
                (, , ),
                (, , )
            ]

         col1, op, col2  relationship_rules:
             col1  .df.columns  col2  .df.columns:
                 op == :
                    valid = (.df[col1] >= .df[col2]).()
                 op == :
                    valid = (.df[col1] > .df[col2]).()
                 op == :
                    valid = (.df[col1] == .df[col2]).()

                total = .df[[col1, col2]].notna().(axis=).()
                consistency[] = (valid / total * )  total >   

        overall = np.mean((consistency.values()))  consistency  

        .results[] = {
            : consistency,
            : overall,
            : ,
            : overall >= 
        }

         .results[]

     ():
        
         date_col   .df.columns:
            .results[] = {
                : ,
                : 
            }
             .results[]

        dates = pd.to_datetime(.df[date_col], errors=)
        cutoff = datetime.now() - timedelta(days=max_age_days)

        recent = (dates >= cutoff).()
        total = dates.notna().()
        timeliness_pct = (recent / total * )  total >   

        oldest = dates.()
        newest = dates.()
        avg_age = (datetime.now() - dates.mean()).days  dates.notna().()  

        .results[] = {
            : timeliness_pct,
            : oldest,
            : newest,
            : avg_age,
            : ,
            : timeliness_pct >= 
        }

         .results[]

     ():
        
         patterns  :
            patterns = {
                : ,  
                : ,
                : ,
                : 
            }

        validity = {}
         col, pattern  patterns.items():
             col  .df.columns:
                non_null = .df[col].dropna()
                 (non_null) > :
                    matches = non_null.astype()..(pattern).()
                    validity[col] = (matches / (non_null) * )

                    invalid = (non_null) - matches
                     invalid > :
                        .issues.append()
                :
                    validity[col] = 

        overall = np.mean((validity.values()))  validity  

        .results[] = {
            : validity,
            : overall,
            : ,
            : overall >= 
        }

         .results[]

     ():
        
        .check_completeness()
        .check_accuracy()
        .check_consistency()
        .check_timeliness()
        .check_validity()

        
        scores = []
         metric  [, , , ]:
             metric  .results  .results[metric].get():
                scores.append(.results[metric][])

        .results[] = np.mean(scores)  scores  
        .results[] = ._calculate_grade(.results[])
        .results[] = .issues

         .results

     ():
        
         score >= :
             
         score >= :
             
         score >= :
             
         score >= :
             
         score >= :
             
        :
             

     ():
        
          .results:
            .run_full_check()

        report = []
        report.append( * )
        report.append()
        report.append( * )
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        report.append( * )

        
         metric  [, , , , ]:
             metric  .results:
                r = .results[metric]
                passed =   r.get(, )  
                overall = r.get(, r.get(, ))
                 (overall, (, )):
                    report.append()
                :
                    report.append()

        report.append( * )

         .issues:
            report.append()
            report.append()
             issue  .issues[:]:  
                report.append()
             (.issues) > :
                report.append()

        report.append()
        report.append( * )

         .join(report)

Validation Rules Builder

Custom Validation Rules

class ValidationRulesBuilder:
    """Build custom validation rules for construction data"""

    def __init__(self):
        self.rules = []

    def add_not_null(self, column):
        """Column must not have null values"""
        self.rules.append({
            'type': 'not_null',
            'column': column,
            'check': lambda df, col=column: df[col].notna().all()
        })
        return self

    def add_unique(self, column):
        """Column must have unique values"""
        self.rules.append({
            'type': 'unique',
            'column': column,
            'check': lambda df, col=column: df[col].nunique() == len(df)
        })
        return self

    def add_range(self, column, min_val=None, max_val=None):
        """Column values must be within range"""
        self.rules.append({
            'type': 'range',
            'column': column,
            'min': min_val,
            'max': max_val,
            :  df, col=column, mn=min_val, mx=max_val:
                df[col].between(mn  -np.inf, mx  np.inf).()
        })
         

     ():
        
        .rules.append({
            : ,
            : column,
            : pattern,
            :  df, col=column, p=pattern:
                df[col].astype()..(p).()
        })
         

     ():
        
        .rules.append({
            : ,
            : column,
            : valid_values,
            :  df, col=column, vals=valid_values:
                df[col].isin(vals).()
        })
         

     ():
        
        .rules.append({
            : ,
            : name,
            : check_func
        })
         

     ():
        
        results = []

         rule  .rules:
            :
                passed = rule[](df)
                results.append({
                    : rule.get(, ),
                    : passed,
                    : rule[]
                })
             Exception  e:
                results.append({
                    : rule.get(, ),
                    : ,
                    : (e)
                })

         results


rules = (ValidationRulesBuilder()
    .add_not_null()
    .add_unique()
    .add_range(, min_val=)
    .add_range(, min_val=)
    .add_in_list(, [, , , , ])
    .add_regex(, )
)

results = rules.validate(df)
 r  results:
    status =   r[]  
    ()

Automated Quality Pipeline

class DataQualityPipeline:
    """Automated data quality pipeline"""

    def __init__(self, config=None):
        self.config = config or self._default_config()
        self.history = []

    def _default_config(self):
        return {
            'required_columns': ['ElementId', 'Category', 'Volume_m3'],
            'unique_columns': ['ElementId'],
            'numeric_ranges': {
                'Volume_m3': (0, 10000),
                'Area_m2': (0, 100000),
                'Cost': (0, 100000000)
            },
            'valid_categories': ['Wall', 'Floor', 'Column', 'Beam', 'Slab',
                                 'Foundation', 'Roof', 'Stair', 'Door', 'Window'],
            'min_quality_score': 90
        }

    def run(self, df, source_name='unknown'):
        """Run quality pipeline"""
        checker = DataQualityChecker(df)

        
        checker.check_completeness(.config[])
        checker.check_accuracy({
            col: {: r[], : r[]}
             col, r  .config[].items()
        })
        checker.check_consistency(.config[])
        checker.check_validity()

        results = checker.run_full_check()

        
        .history.append({
            : datetime.now(),
            : source_name,
            : (df),
            : results[],
            : results[],
            : (results[])
        })

        
        passed = results[] >= .config[]

         {
            : passed,
            : results[],
            : results[],
            : results,
            : checker.generate_report()
        }

     ():
        
          .history:
             

        df_history = pd.DataFrame(.history)
         {
            : (.history),
            : df_history[].mean(),
            : df_history[].(),
            : df_history[].(),
            : .history[-]
        }

Quality Reporting

Export Quality Report

def export_quality_report(df, output_path, include_details=True):
    """Export comprehensive quality report to Excel"""
    checker = DataQualityChecker(df)
    results = checker.run_full_check()

    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary sheet
        summary = pd.DataFrame({
            'Metric': ['Overall Score', 'Grade', 'Records', 'Columns', 'Issues'],
            'Value': [
                f"{results['overall_score']:.1f}%",
                results['grade'],
                len(df),
                len(df.columns),
                len(results['issues'])
            ]
        })
        summary.to_excel(writer, sheet_name='Summary', index=False)

        # Completeness details
        if 'completeness' in results:
            comp_df = pd.DataFrame.from_dict(
                results['completeness']['by_column'],
                orient='index',
                columns=['Completeness_%']
            )
            comp_df.to_excel(writer, sheet_name='Completeness')

        # Issues list
        if results['issues']:
            issues_df = pd.DataFrame({'Issue': results['issues']})
            issues_df.to_excel(writer, sheet_name='Issues', index=)

        
         include_details:
            missing = df.isnull().()
            missing_df = pd.DataFrame({
                : missing.index,
                : missing.values,
                : (missing.values / (df) * ).()
            })
            missing_df.to_excel(writer, sheet_name=, index=)

     output_path

Quick Reference

Metric	Description	Threshold
Completeness	% non-null values	≥ 95%
Accuracy	Values within valid range	≥ 98%
Consistency	Unique IDs, valid relationships	≥ 99%
Validity	Match expected patterns	≥ 95%
Timeliness	Records updated recently	≥ 80%

Common Validation Patterns

# Construction-specific regex patterns
PATTERNS = {
    'element_id': r'^[A-Z]{1,3}\d{3,8}$',
    'revit_id': r'^\d{5,8}$',
    'ifc_guid': r'^[A-Za-z0-9_$]{22}$',
    'level': r'^(Level|L|Уровень)\s*[-]?\d+$',
    'grid': r'^[A-Z]{1,2}[-/]?\d{0,3}$',
    'date_iso': r'^\d{4}-\d{2}-\d{2}$',
    'cost_code': r'^\d{2,3}[.-]\d{2,4}[.-]?\d{0,4}$'
}

Resources

• Book: "Data-Driven Construction" by Artem Boiko, Chapter 2.6
• Website: https://datadrivenconstruction.io
• Great Expectations: https://greatexpectations.io

Next Steps

• See bim-validation-pipeline for BIM-specific validation
• See etl-pipeline for data processing pipelines
• See data-visualization for quality dashboards