| 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":["python3"]}}} |
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
auditor = DataSourceAuditor()
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"]
))
auditor.add_flow(DataFlow(
source="Procore",
target="Sage 300",
flow_type="push",
frequency="daily",
entities=["change_orders", "budget_changes"],
transformation=,
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:
[, , , , ]
Resources
- DAMA DMBOK: Data Management Body of Knowledge
- Data Governance Frameworks: DCAM, EDM Council
- Integration Patterns: Enterprise Integration Patterns book
