data-source-audit

name	data-source-audit
description	Comprehensive audit of all construction data sources and systems. Map data flows, identify silos, assess quality, and create integration roadmap.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🔗","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Data Source Audit for Construction

Overview

Perform comprehensive audits of construction data sources to identify silos, map data flows, assess quality, and plan integration strategies. Essential for digital transformation and data-driven construction initiatives.

Business Case

Construction organizations typically have 10-50+ data sources:

• Project management systems
• Estimating software
• Scheduling tools
• Accounting/ERP systems
• BIM platforms
• Document management systems
• Field apps
• Spreadsheets

Note: This skill is vendor-agnostic and works with any data source. Product names mentioned elsewhere in examples are trademarks of their respective owners.

This skill helps:

• Discover all data sources
• Map data flows and dependencies
• Identify integration opportunities
• Prioritize data improvement efforts

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Set
from enum import Enum
from datetime import datetime
import pandas as pd
import json

class DataSourceType(Enum):
    DATABASE = "database"
    API = "api"
    FILE_SHARE = "file_share"
    CLOUD_APP = "cloud_app"
    SPREADSHEET = "spreadsheet"
    LEGACY_SYSTEM = "legacy_system"
    IOT_SENSOR = "iot_sensor"
    MANUAL_ENTRY = "manual_entry"

class DataDomain(Enum):
    COST = "cost"
    SCHEDULE = "schedule"
    BIM = "bim"
    DOCUMENT = "document"
    FIELD = "field"
    SAFETY = "safety"
    QUALITY = "quality"
    HR = "hr"
    ACCOUNTING = "accounting"
    PROCUREMENT = "procurement"

@dataclass
class DataSource:
    name: str
    source_type: DataSourceType
    domains: List[DataDomain]
    owner: str
    department: str
    description: str
    
    technology: 
    location:   
    access_method:   
    
    update_frequency:   
    data_volume:   
    retention_period: 
    
    completeness_score:  = 
    accuracy_score:  = 
    timeliness_score:  = 
    
    integrations: [] = field(default_factory=)
    is_master:  =   
    master_for: [] = field(default_factory=)
    
    known_issues: [] = field(default_factory=)
    
    last_audit_date: [datetime] = 
    audit_notes:  = 


 :
    source: 
    target: 
    flow_type:   
    frequency: 
    entities: []  
    transformation:   
    status:   


 :
    name: 
    sources: []
    impact:   
    description: 
    resolution_options: []

 :
    

     ():
        .sources: [, DataSource] = {}
        .flows: [DataFlow] = []
        .silos: [DataSilo] = []

     ():
        
        .sources[source.name] = source

     ():
        
        .flows.append(flow)

     () -> [DataSource]:
        
        sources = []

         response  survey_responses:
            source = DataSource(
                name=response[],
                source_type=DataSourceType(response[]),
                domains=[DataDomain(d)  d  response[]],
                owner=response[],
                department=response[],
                description=response[],
                technology=response[],
                location=response[],
                access_method=response[],
                update_frequency=response[],
                data_volume=response[],
                retention_period=response[],
            )
            sources.append(source)
            .add_source(source)

         sources

     () -> [DataSilo]:
        
        silos = []

        
        isolated_sources = [
            name  name, source  .sources.items()
              source.integrations  source.source_type != DataSourceType.MANUAL_ENTRY
        ]

         isolated_sources:
            silos.append(DataSilo(
                name=,
                sources=isolated_sources,
                impact=,
                description=,
                resolution_options=[
                    ,
                    ,
                    
                ]
            ))

        
        domain_sources: [DataDomain, []] = {}
         name, source  .sources.items():
             domain  source.domains:
                 domain   domain_sources:
                    domain_sources[domain] = []
                domain_sources[domain].append(name)

         domain, sources  domain_sources.items():
             (sources) > :
                
                masters = [s  s  sources  .sources[s].is_master]
                  masters:
                    silos.append(DataSilo(
                        name=,
                        sources=sources,
                        impact=,
                        description=,
                        resolution_options=[
                            ,
                            ,
                            
                        ]
                    ))

        
        flow_pairs = {}
         flow  .flows:
            key = (([flow.source, flow.target]))
             key   flow_pairs:
                flow_pairs[key] = []
            flow_pairs[key].append(flow)

         (s1, s2), flows  flow_pairs.items():
             (flows) ==   flows[].flow_type != :
                
                s1_domains = (.sources[s1].domains)
                s2_domains = (.sources[s2].domains)
                 s1_domains & s2_domains:  
                    silos.append(DataSilo(
                        name=,
                        sources=[s1, s2],
                        impact=,
                        description=,
                        resolution_options=[
                            ,
                            
                        ]
                    ))

        .silos = silos
         silos

     () -> [, ]:
        
         source_name   .sources:
             ValueError()

        scores = {}

        
        completeness =  - (sample_data.isnull().().() / sample_data.size)
        scores[] = completeness

        
         (sample_data) > :
            uniqueness = (sample_data.drop_duplicates()) / (sample_data)
        :
            uniqueness = 
        scores[] = uniqueness

        
        validity_checks = 
        total_checks = 

         col  sample_data.columns:
               col.lower():
                total_checks += 
                :
                    pd.to_datetime(sample_data[col], errors=)
                    validity_checks += 
                :
                    
               col.lower():
                total_checks += 
                valid_emails = sample_data[col]..contains(, na=).()
                 valid_emails / (sample_data) > :
                    validity_checks += 

        scores[] = validity_checks / total_checks  total_checks >   

        
        .sources[source_name].completeness_score = scores[]
        .sources[source_name].accuracy_score = scores[]

         scores

     () -> pd.DataFrame:
        
        catalog_entries = []

         name, source  .sources.items():
            entry = {
                : name,
                : source.source_type.value,
                : .join(d.value  d  source.domains),
                : source.owner,
                : source.department,
                : source.technology,
                : source.location,
                : source.access_method,
                : source.update_frequency,
                : source.data_volume,
                : (source.integrations),
                :   source.is_master  ,
                : (source.completeness_score + source.accuracy_score) / ,
                : (source.known_issues),
            }
            catalog_entries.append(entry)

         pd.DataFrame(catalog_entries)

     () -> pd.DataFrame:
        
        source_names = (.sources.keys())
        matrix = pd.DataFrame(
            index=source_names,
            columns=source_names,
            data=
        )

         flow  .flows:
             flow.source  source_names  flow.target  source_names:
                current = matrix.loc[flow.source, flow.target]
                symbol =   flow.flow_type ==     flow.flow_type ==   
                matrix.loc[flow.source, flow.target] =   current  symbol

         matrix

     () -> [, ]:
        
          .sources:
             {: }

        scores = {}

        
        integrated = (  s  .sources.values()  s.integrations)
        scores[] = integrated / (.sources)

        
        domains_with_master = ()
         source  .sources.values():
             source.is_master:
                domains_with_master.update(source.master_for)

        all_domains = ()
         source  .sources.values():
            all_domains.update(d.value  d  source.domains)

        scores[] = (domains_with_master) / (all_domains)  all_domains  

        
        quality_scores = [
            (s.completeness_score + s.accuracy_score) / 
             s  .sources.values()
             s.completeness_score >   s.accuracy_score > 
        ]
        scores[] = (quality_scores) / (quality_scores)  quality_scores  

        
        high_impact_silos = (  s  .silos  s.impact == )
        scores[] =  - (high_impact_silos * )  

        
        scores[] = (
            scores[] *  +
            scores[] *  +
            scores[] *  +
            scores[] * 
        )

         scores

     () -> :
        
        report = [, ]
        report.append()
        report.append()
        report.append()
        report.append()

        
        scores = .calculate_integration_score()
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()

        
        report.append()
        by_type = {}
         source  .sources.values():
            t = source.source_type.value
            by_type[t] = by_type.get(t, ) + 
         t, count  (by_type.items(), key= x: -x[]):
            report.append()
        report.append()

        
        report.append()
         .silos:
             silo  .silos:
                report.append()
                report.append()
                report.append()
                report.append()
                report.append()
                 opt  silo.resolution_options:
                    report.append()
        :
            report.append()
        report.append()

        
        report.append()
        recommendations = ._generate_recommendations()
         i, rec  (recommendations, ):
            report.append()

         .join(report)

     () -> []:
        
        recommendations = []

        scores = .calculate_integration_score()

         scores[] < :
            recommendations.append(
                
                
            )

         scores[] < :
            recommendations.append(
                
                
            )

         scores[] < :
            recommendations.append(
                
                
            )

        
        spreadsheets = [s  s  .sources.values()
                        s.source_type == DataSourceType.SPREADSHEET]
         (spreadsheets) > :
            recommendations.append(
                
                
            )

        
        legacy = [s  s  .sources.values()
                  s.source_type == DataSourceType.LEGACY_SYSTEM]
         legacy:
            recommendations.append(
                
                
            )

         recommendations

Quick Start

# Initialize auditor
auditor = DataSourceAuditor()

# Add known sources
auditor.add_source(DataSource(
    name="Procore",
    source_type=DataSourceType.CLOUD_APP,
    domains=[DataDomain.DOCUMENT, DataDomain.FIELD, DataDomain.SCHEDULE],
    owner="Project Controls",
    department="Operations",
    description="Primary project management platform",
    technology="SaaS",
    location="cloud",
    access_method="API",
    update_frequency="real-time",
    data_volume="large",
    retention_period="7 years",
    integrations=["Sage 300", "Primavera P6"],
    is_master=True,
    master_for=["projects", "documents"]
))

auditor.add_source(DataSource(
    name="Sage 300",
    source_type=DataSourceType.DATABASE,
    domains=[DataDomain.COST, DataDomain.ACCOUNTING],
    owner="Finance",
    department="Accounting",
    description="ERP and job costing system",
    technology="SQL Server",
    location="on-prem",
    access_method="ODBC",
    update_frequency="daily",
    data_volume="medium",
    retention_period="10 years",
    is_master=True,
    master_for=["costs", "vendors", "invoices"]
))

# Add data flows
auditor.add_flow(DataFlow(
    source="Procore",
    target="Sage 300",
    flow_type="push",
    frequency="daily",
    entities=["change_orders", "budget_changes"],
    transformation="simple",
    status=
))


silos = auditor.identify_silos()


report = auditor.generate_audit_report()
(report)


catalog = auditor.create_data_catalog()
catalog.to_excel(, index=)

Survey Template

Use this survey to discover data sources across the organization:

System Survey:
  - system_name: "What is the name of this system?"
  - type: "What type of system is it?"
    options: [database, api, file_share, cloud_app, spreadsheet, legacy_system]
  - domains: "What types of data does it contain?"
    options: [cost, schedule, bim, document, field, safety, quality, hr, accounting]
  - owner: "Who is the system owner?"
  - department: "Which department uses this system?"
  - technology: "What technology/platform is it built on?"
  - location: "Where is the system hosted?"
    options: [cloud, on-prem, hybrid]
  - access_method: "How can data be accessed?"
    options: [api, odbc, file_export, manual]
  - update_frequency: "How often is data updated?"
     [, , , , ]
    

Resources

• DAMA DMBOK: Data Management Body of Knowledge
• Data Governance Frameworks: DCAM, EDM Council
• Integration Patterns: Enterprise Integration Patterns book