luokai0 Data Profiler for Construction
Overview
Analyze construction data to understand its characteristics, distributions, quality, and patterns. Essential for data quality assessment, ETL planning, and identifying data issues before they impact projects.
Business Case
Before using any construction data, you need to understand:
- What data types are present
- Distribution of values
- Missing data patterns
- Anomalies and outliers
- Referential integrity issues
This skill profiles data to answer these questions and provides actionable insights.
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
import numpy as np
from datetime import datetime
import json
@dataclass
class ColumnProfile:
name: str
data_type: str
inferred_type: str # More specific: project_id, cost, date, csi_code, etc.
total_count: int
null_count: int
null_percentage: float
unique_count: int
uniqueness_ratio: float
# For numeric columns
min_value: Optional[float] = None
max_value: Optional[float] = None
mean_value: Optional[float] = None
median_value: Optional[float] = None
std_dev: Optional[float] = None
# For string columns
min_length: Optional[int] = None
max_length: Optional[int] = None
avg_length: Optional[float] = None
# Top values
top_values: List[Tuple[Any, int]] = field(default_factory=list)
# Patterns
common_patterns: List[str] = field(default_factory=list)
# Quality flags
quality_issues: List[str] = field(default_factory=list)
@dataclass
class DataProfile:
source_name: str
row_count: int
column_count: int
columns: List[ColumnProfile]
duplicate_rows: int
memory_usage: str
profiled_at: datetime
quality_score: float
recommendations: List[str]
class ConstructionDataProfiler:
"""Profile construction data for quality and characteristics."""
# Known construction data patterns
CONSTRUCTION_PATTERNS = {
'csi_code': r'^\d{2}\s?\d{2}\s?\d{2}$',
'project_id': r'^[A-Z]{2,4}[-_]?\d{3,6}$',
'cost_code': r'^\d{2}[-.]?\d{2,4}$',
'wbs': r'^[\d.]+$',
'phone': r'^\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}$',
'email': r'^[\w.-]+@[\w.-]+\.\w+$',
'date_iso': r'^\d{4}-\d{2}-\d{2}',
'date_us': r'^\d{1,2}/\d{1,2}/\d{2,4}$',
'currency': r'^\$?[\d,]+\.?\d{0,2}$',
'percentage': r'^\d+\.?\d*%?$',
}
# Construction-specific column name patterns
COLUMN_TYPE_HINTS = {
'project': ['project_id', 'project_name', 'proj', 'job'],
'cost': ['cost', 'amount', 'price', 'total', 'budget', 'actual'],
'date': ['date', 'start', 'finish', 'end', 'created', 'modified'],
'quantity': ['qty', 'quantity', 'count', 'units'],
'csi': ['csi', 'division', 'masterformat', 'spec'],
'location': ['location', 'area', 'zone', 'floor', 'level'],
'person': ['owner', 'manager', 'superintendent', 'foreman', 'contact'],
}
def __init__(self):
self.profiles: Dict[str, DataProfile] = {}
def profile_dataframe(self, df: pd.DataFrame, source_name: str) -> DataProfile:
"""Profile a pandas DataFrame."""
columns = []
for col in df.columns:
col_profile = self._profile_column(df[col], col)
columns.append(col_profile)
# Calculate duplicates
duplicate_rows = len(df) - len(df.drop_duplicates())
# Calculate memory usage
memory_bytes = df.memory_usage(deep=True).sum()
if memory_bytes < 1024:
memory_usage = f"{memory_bytes} B"
elif memory_bytes < 1024**2:
memory_usage = f"{memory_bytes/1024:.1f} KB"
else:
memory_usage = f"{memory_bytes/1024**2:.1f} MB"
# Calculate overall quality score
quality_score = self._calculate_quality_score(columns)
# Generate recommendations
recommendations = self._generate_recommendations(columns, df)
profile = DataProfile(
source_name=source_name,
row_count=len(df),
column_count=len(df.columns),
columns=columns,
duplicate_rows=duplicate_rows,
memory_usage=memory_usage,
profiled_at=datetime.now(),
quality_score=quality_score,
recommendations=recommendations
)
self.profiles[source_name] = profile
return profile
def _profile_column(self, series: pd.Series, name: str) -> ColumnProfile:
"""Profile a single column."""
total_count = len(series)
null_count = series.isnull().sum()
null_percentage = (null_count / total_count * 100) if total_count > 0 else 0
# Get non-null values for analysis
non_null = series.dropna()
unique_count = non_null.nunique()
uniqueness_ratio = unique_count / len(non_null) if len(non_null) > 0 else 0
profile = ColumnProfile(
name=name,
data_type=str(series.dtype),
inferred_type=self._infer_construction_type(series, name),
total_count=total_count,
null_count=null_count,
null_percentage=round(null_percentage, 2),
unique_count=unique_count,
uniqueness_ratio=round(uniqueness_ratio, 4)
)
# Numeric analysis
if pd.api.types.is_numeric_dtype(series):
profile.min_value = float(non_null.min()) if len(non_null) > 0 else None
profile.max_value = float(non_null.max()) if len(non_null) > 0 else None
profile.mean_value = float(non_null.mean()) if len(non_null) > 0 else None
profile.median_value = float(non_null.median()) if len(non_null) > 0 else None
profile.std_dev = float(non_null.std()) if len(non_null) > 1 else None
# Check for outliers
if len(non_null) > 10 and profile.std_dev:
outliers = non_null[abs(non_null - profile.mean_value) > 3 * profile.std_dev]
if len(outliers) > 0:
profile.quality_issues.append(f"{len(outliers)} potential outliers detected")
# Check for negative costs
if any(hint in name.lower() for hint in ['cost', 'amount', 'price', 'total']):
negatives = (non_null < 0).sum()
if negatives > 0:
profile.quality_issues.append(f"{negatives} negative values in cost column")
# String analysis
elif pd.api.types.is_object_dtype(series) or pd.api.types.is_string_dtype(series):
str_series = non_null.astype(str)
lengths = str_series.str.len()
profile.min_length = int(lengths.min()) if len(lengths) > 0 else None
profile.max_length = int(lengths.max()) if len(lengths) > 0 else None
profile.avg_length = float(lengths.mean()) if len(lengths) > 0 else None
# Detect patterns
profile.common_patterns = self._detect_patterns(str_series)
# Top values
if len(non_null) > 0:
value_counts = non_null.value_counts().head(5)
profile.top_values = list(zip(value_counts.index.tolist(), value_counts.values.tolist()))
# Quality checks
if null_percentage > 50:
profile.quality_issues.append("High null rate (>50%)")
if uniqueness_ratio == 1.0 and total_count > 100:
profile.quality_issues.append("All unique values - possible ID column")
if uniqueness_ratio < 0.01 and unique_count > 1:
profile.quality_issues.append("Low cardinality - possible category")
return profile
def _infer_construction_type(self, series: pd.Series, name: str) -> str:
"""Infer construction-specific data type."""
name_lower = name.lower()
# Check column name hints
for type_name, hints in self.COLUMN_TYPE_HINTS.items():
if any(hint in name_lower for hint in hints):
return type_name
# Check data patterns
non_null = series.dropna().astype(str)
if len(non_null) == 0:
return "unknown"
sample = non_null.head(100)
for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
matches = sample.str.match(pattern, na=False).sum()
if matches / len(sample) > 0.8:
return pattern_name
# Default to pandas dtype
if pd.api.types.is_numeric_dtype(series):
return "numeric"
elif pd.api.types.is_datetime64_any_dtype(series):
return "datetime"
else:
return "text"
def _detect_patterns(self, str_series: pd.Series) -> List[str]:
"""Detect common patterns in string data."""
patterns_found = []
sample = str_series.head(1000)
for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
matches = sample.str.match(pattern, na=False).sum()
if matches / len(sample) > 0.1:
patterns_found.append(f"{pattern_name} ({matches/len(sample):.0%})")
return patterns_found[:3]
def _calculate_quality_score(self, columns: List[ColumnProfile]) -> float:
"""Calculate overall data quality score (0-100)."""
if not columns:
return 0.0
scores = []
for col in columns:
col_score = 100
# Penalize for nulls
col_score -= min(col.null_percentage, 50)
# Penalize for quality issues
col_score -= len(col.quality_issues) * 10
scores.append(max(col_score, 0))
return round(sum(scores) / len(scores), 1)
def _generate_recommendations(self, columns: List[ColumnProfile], df: pd.DataFrame) -> List[str]:
"""Generate recommendations based on profile."""
recommendations = []
# High null columns
high_null = [c for c in columns if c.null_percentage > 30]
if high_null:
recommendations.append(
f"Review {len(high_null)} columns with >30% null values: "
f"{', '.join(c.name for c in high_null[:3])}"
)
# Potential ID columns without uniqueness
for col in columns:
if 'id' in col.name.lower() and col.uniqueness_ratio < 1.0:
recommendations.append(
f"Column '{col.name}' appears to be an ID but has duplicate values"
)
# Date columns that should be datetime
for col in columns:
if col.inferred_type in ['date_iso', 'date_us'] and col.data_type == 'object':
recommendations.append(
f"Convert '{col.name}' to datetime type for better analysis"
)
# Cost columns that are strings
for col in columns:
if col.inferred_type == 'currency' and col.data_type == 'object':
recommendations.append(
f"Convert '{col.name}' to numeric type (remove $ and commas)"
)
return recommendations
def profile_to_dict(self, profile: DataProfile) -> Dict:
"""Convert profile to dictionary for JSON export."""
return {
'source_name': profile.source_name,
'row_count': profile.row_count,
'column_count': profile.column_count,
'duplicate_rows': profile.duplicate_rows,
'memory_usage': profile.memory_usage,
'profiled_at': profile.profiled_at.isoformat(),
'quality_score': profile.quality_score,
'recommendations': profile.recommendations,
'columns': [
{
'name': c.name,
'data_type': c.data_type,
'inferred_type': c.inferred_type,
'null_percentage': c.null_percentage,
'unique_count': c.unique_count,
'quality_issues': c.quality_issues,
'top_values': c.top_values[:3]
}
for c in profile.columns
]
}
def generate_profile_report(self, profile: DataProfile) -> str:
"""Generate markdown profile report."""
report = [f"# Data Profile: {profile.source_name}", ""]
report.append(f"**Profiled At:** {profile.profiled_at.strftime('%Y-%m-%d %H:%M')}")
report.append(f"**Quality Score:** {profile.quality_score}/100")
report.append("")
# Summary
report.append("## Summary")
report.append(f"- **Rows:** {profile.row_count:,}")
report.append(f"- **Columns:** {profile.column_count}")
report.append(f"- **Duplicate Rows:** {profile.duplicate_rows:,}")
report.append(f"- **Memory Usage:** {profile.memory_usage}")
report.append("")
# Recommendations
if profile.recommendations:
report.append("## Recommendations")
for rec in profile.recommendations:
report.append(f"- {rec}")
report.append("")
# Column Details
report.append("## Column Details")
report.append("")
report.append("| Column | Type | Inferred | Nulls | Unique | Issues |")
report.append("|--------|------|----------|-------|--------|--------|")
for col in profile.columns:
issues = len(col.quality_issues)
report.append(
f"| {col.name} | {col.data_type} | {col.inferred_type} | "
f"{col.null_percentage:.1f}% | {col.unique_count:,} | {issues} |"
)
# Detailed column profiles
report.append("")
report.append("## Detailed Column Profiles")
for col in profile.columns:
report.append(f"\n### {col.name}")
report.append(f"- **Type:** {col.data_type} (inferred: {col.inferred_type})")
report.append(f"- **Nulls:** {col.null_count:,} ({col.null_percentage:.1f}%)")
report.append(f"- **Unique Values:** {col.unique_count:,} ({col.uniqueness_ratio:.1%})")
if col.min_value is not None:
report.append(f"- **Range:** {col.min_value:,.2f} to {col.max_value:,.2f}")
report.append(f"- **Mean:** {col.mean_value:,.2f}, Median: {col.median_value:,.2f}")
if col.min_length is not None:
report.append(f"- **Length:** {col.min_length} to {col.max_length} (avg: {col.avg_length:.1f})")
if col.top_values:
report.append(f"- **Top Values:** {col.top_values[:3]}")
if col.common_patterns:
report.append(f"- **Patterns:** {', '.join(col.common_patterns)}")
if col.quality_issues:
report.append(f"- **Issues:** {', '.join(col.quality_issues)}")
return "\n".join(report)
def compare_profiles(self, profile1: DataProfile, profile2: DataProfile) -> Dict:
"""Compare two profiles to detect schema changes or data drift."""
comparison = {
'profiles': [profile1.source_name, profile2.source_name],
'row_count_change': profile2.row_count - profile1.row_count,
'quality_change': profile2.quality_score - profile1.quality_score,
'new_columns': [],
'removed_columns': [],
'type_changes': [],
'null_rate_changes': []
}
cols1 = {c.name: c for c in profile1.columns}
cols2 = {c.name: c for c in profile2.columns}
# Find new/removed columns
comparison['new_columns'] = [n for n in cols2 if n not in cols1]
comparison['removed_columns'] = [n for n in cols1 if n not in cols2]
# Compare common columns
for name in cols1:
if name in cols2:
c1, c2 = cols1[name], cols2[name]
if c1.data_type != c2.data_type:
comparison['type_changes'].append({
'column': name,
'from': c1.data_type,
'to': c2.data_type
})
null_change = c2.null_percentage - c1.null_percentage
if abs(null_change) > 10:
comparison['null_rate_changes'].append({
'column': name,
'change': null_change
})
return comparison
Quick Start
import pandas as pd
# Load construction data
df = pd.read_excel("project_costs.xlsx")
# Profile the data
profiler = ConstructionDataProfiler()
profile = profiler.profile_dataframe(df, "Project Costs 2025")
# Generate report
report = profiler.generate_profile_report(profile)
print(report)
# Export to JSON
profile_dict = profiler.profile_to_dict(profile)
with open("profile.json", "w") as f:
json.dump(profile_dict, f, indent=2)
# Compare with previous profile
old_profile = profiler.profile_dataframe(old_df, "Project Costs 2024")
comparison = profiler.compare_profiles(old_profile, profile)
print(f"Quality changed by: {comparison['quality_change']}")
Common Use Cases
- Pre-ETL Analysis: Profile source data before building pipelines
- Quality Monitoring: Track data quality over time
- Schema Validation: Detect unexpected changes in data structure
- Anomaly Detection: Find outliers and data quality issues
Dependencies
pip install pandas numpy
Resources
- Data Profiling Best Practices: DAMA DMBOK
- Construction Data Standards: CSI MasterFormat, UniFormat