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.