data-profiler

What this skill does

# 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

```python
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_ty