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.