\r

Data Profiler for Construction\r

\r

Overview\r

\r 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.\r \r

Business Case\r

\r Before using any construction data, you need to understand:\r

• What data types are present\r
• Distribution of values\r
• Missing data patterns\r
• Anomalies and outliers\r
• Referential integrity issues\r \r This skill profiles data to answer these questions and provides actionable insights.\r \r

Technical Implementation\r

\r

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