Input Validation for Construction Data

Overview

Validate incoming construction data before processing to catch errors early. Domain-specific validation rules for estimates, schedules, BIM exports, and field data.

Validation Framework

Core Validator Class

from dataclasses import dataclass, field
from typing import List, Dict, Any, Callable, Optional
from enum import Enum
import re
from datetime import datetime

class ValidationSeverity(Enum):
    ERROR = "error"      # Must fix, blocks processing
    WARNING = "warning"  # Should review, allows processing
    INFO = "info"        # FYI, no action needed

@dataclass
class ValidationIssue:
    field: str
    message: str
    severity: ValidationSeverity
    value: Any = None
    suggestion: str = None

@dataclass
class ValidationResult:
    is_valid: bool
    issues: List[ValidationIssue] = field(default_factory=list)

    def add_error(self, field: str, message: str, value: Any = None, suggestion: str = None):
        self.issues.append(ValidationIssue(field, message, ValidationSeverity.ERROR, value, suggestion))
        self.is_valid = False

    def add_warning(self, field: str, message: str, value: Any = None, suggestion: str = None):
        self.issues.append(ValidationIssue(field, message, ValidationSeverity.WARNING, value, suggestion))

    def add_info(self, field: str, message: str, value: Any = None):
        self.issues.append(ValidationIssue(field, message, ValidationSeverity.INFO, value))

    @property
    def errors(self) -> List[ValidationIssue]:
        return [i for i in self.issues if i.severity == ValidationSeverity.ERROR]

    @property
    def warnings(self) -> List[ValidationIssue]:
        return [i for i in self.issues if i.severity == ValidationSeverity.WARNING]

    def to_report(self) -> str:
        lines = ["VALIDATION REPORT", "=" * 50]
        lines.append(f"Status: {'PASSED' if self.is_valid else 'FAILED'}")
        lines.append(f"Errors: {len(self.errors)}, Warnings: {len(self.warnings)}")
        lines.append("")

        for issue in self.issues:
            icon = "❌" if issue.severity == ValidationSeverity.ERROR else "⚠️" if issue.severity == ValidationSeverity.WARNING else "ℹ️"
            lines.append(f"{icon} [{issue.field}] {issue.message}")
            if issue.suggestion:
                lines.append(f"   Suggestion: {issue.suggestion}")

        return "\n".join(lines)

Cost Estimate Validation

class CostEstimateValidator:
    """Validate cost estimate inputs."""

    # Typical cost ranges per CSI division ($/SF)
    TYPICAL_RANGES = {
        '03': (15, 45),    # Concrete
        '04': (8, 25),     # Masonry
        '05': (12, 35),    # Metals
        '06': (5, 20),     # Wood/Plastics
        '07': (8, 30),     # Thermal/Moisture
        '08': (15, 50),    # Openings
        '09': (10, 40),    # Finishes
        '22': (8, 25),     # Plumbing
        '23': (12, 40),    # HVAC
        '26': (10, 35),    # Electrical
    }

    def validate(self, estimate_data: Dict[str, Any]) -> ValidationResult:
        result = ValidationResult(is_valid=True)

        # Required fields
        self._validate_required_fields(estimate_data, result)

        # Line item validation
        if 'line_items' in estimate_data:
            self._validate_line_items(estimate_data['line_items'], result)

        # Total validation
        self._validate_totals(estimate_data, result)

        # Cost range validation
        if 'gross_area' in estimate_data:
            self._validate_cost_ranges(estimate_data, result)

        return result

    def _validate_required_fields(self, data: dict, result: ValidationResult):
        required = ['project_name', 'estimate_date', 'line_items', 'total']
        for field in required:
            if field not in data or data[field] is None:
                result.add_error(field, f"Required field '{field}' is missing")

    def _validate_line_items(self, items: list, result: ValidationResult):
        for i, item in enumerate(items):
            # Check for negative values
            if item.get('quantity', 0) < 0:
                result.add_error(f"line_items[{i}].quantity", "Quantity cannot be negative", item.get('quantity'))

            if item.get('unit_cost', 0) < 0:
                result.add_error(f"line_items[{i}].unit_cost", "Unit cost cannot be negative", item.get('unit_cost'))

            # Check for missing descriptions
            if not item.get('description'):
                result.add_warning(f"line_items[{i}].description", "Line item missing description")

            # Check for valid CSI code
            if item.get('csi_code'):
                if not re.match(r'^\d{2}\s?\d{2}\s?\d{2}$', item['csi_code']):
                    result.add_warning(f"line_items[{i}].csi_code", f"Invalid CSI code format: {item['csi_code']}", suggestion="Use format: XX XX XX")

            # Check for zero amounts
            amount = item.get('quantity', 0) * item.get('unit_cost', 0)
            if amount == 0:
                result.add_warning(f"line_items[{i}]", "Line item has zero amount")

    def _validate_totals(self, data: dict, result: ValidationResult):
        if 'line_items' not in data or 'total' not in data:
            return

        calculated = sum(
            item.get('quantity', 0) * item.get('unit_cost', 0)
            for item in data['line_items']
        )

        declared = data['total']
        variance = abs(calculated - declared)

        if variance > 0.01:
            result.add_error("total", f"Total mismatch: calculated {calculated:.2f}, declared {declared:.2f}", variance)

    def _validate_cost_ranges(self, data: dict, result: ValidationResult):
        gross_area = data['gross_area']

        for item in data.get('line_items', []):
            csi_div = item.get('csi_code', '')[:2]
            if csi_div in self.TYPICAL_RANGES:
                amount = item.get('quantity', 0) * item.get('unit_cost', 0)
                cost_per_sf = amount / gross_area if gross_area > 0 else 0

                low, high = self.TYPICAL_RANGES[csi_div]
                if cost_per_sf < low * 0.5 or cost_per_sf > high * 2:
                    result.add_warning(
                        f"line_items[{item.get('description', 'Unknown')}]",
                        f"Cost ${cost_per_sf:.2f}/SF outside typical range ${low}-${high}/SF for Division {csi_div}",
                        cost_per_sf,
                        "Review unit costs and quantities"
                    )

Schedule Validation

class ScheduleValidator:
    """Validate schedule/planning inputs."""

    def validate(self, schedule_data: Dict[str, Any]) -> ValidationResult:
        result = ValidationResult(is_valid=True)

        # Required fields
        self._validate_required_fields(schedule_data, result)

        # Task validation
        if 'tasks' in schedule_data:
            self._validate_tasks(schedule_data['tasks'], result)
            self._validate_dependencies(schedule_data['tasks'], result)
            self._validate_resources(schedule_data['tasks'], result)

        return result

    def _validate_required_fields(self, data: dict, result: ValidationResult):
        required = ['project_name', 'start_date', 'tasks']
        for field in required:
            if field not in data:
                result.add_error(field, f"Required field '{field}' is missing")

    def _validate_tasks(self, tasks: list, result: ValidationResult):
        task_ids = set()

        for i, task in enumerate(tasks):
            # Check for duplicate IDs
            task_id = task.get('id')
            if task_id in task_ids:
                result.add_error(f"tasks[{i}].id", f"Duplicate task ID: {task_id}")
            task_ids.add(task_id)

            # Check dates
            start = task.get('start_date')
            end = task.get('end_date')

            if start and end:
                try: