input-validation

name	input-validation
description	Validate construction data inputs before processing: cost estimates, schedules, BIM data, field reports. Catch errors early with domain-specific rules.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"✔️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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
        ._validate_required_fields(estimate_data, result)

        
           estimate_data:
            ._validate_line_items(estimate_data[], result)

        
        ._validate_totals(estimate_data, result)

        
           estimate_data:
            ._validate_cost_ranges(estimate_data, result)

         result

     ():
        required = [, , , ]
         field  required:
             field   data  data[field]  :
                result.add_error(field, )

     ():
         i, item  (items):
            
             item.get(, ) < :
                result.add_error(, , item.get())

             item.get(, ) < :
                result.add_error(, , item.get())

            
              item.get():
                result.add_warning(, )

            
             item.get():
                  re.(, item[]):
                    result.add_warning(, , suggestion=)

            
            amount = item.get(, ) * item.get(, )
             amount == :
                result.add_warning(, )

     ():
            data     data:
            

        calculated = (
            item.get(, ) * item.get(, )
             item  data[]
        )

        declared = data[]
        variance = (calculated - declared)

         variance > :
            result.add_error(, , variance)

     ():
        gross_area = data[]

         item  data.get(, []):
            csi_div = item.get(, )[:]
             csi_div  .TYPICAL_RANGES:
                amount = item.get(, ) * item.get(, )
                cost_per_sf = amount / gross_area  gross_area >   

                low, high = .TYPICAL_RANGES[csi_div]
                 cost_per_sf < low *   cost_per_sf > high * :
                    result.add_warning(
                        ,
                        ,
                        cost_per_sf,
                        
                    )

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):
            
            task_id = task.get()
             task_id  task_ids:
                result.add_error(, )
            task_ids.add(task_id)

            
            start = task.get()
            end = task.get()

             start  end:
                :
                    start_dt = datetime.fromisoformat(start)  (start, )  start
                    end_dt = datetime.fromisoformat(end)  (end, )  end

                     end_dt < start_dt:
                        result.add_error(, , )

                    
                    duration = (end_dt - start_dt).days
                     duration > :
                        result.add_warning(, )
                     duration ==   task.get() != :
                        result.add_warning(, )

                 ValueError  e:
                    result.add_error(, )

            
              task.get()   (start  end):
                result.add_error(, )

     ():
        task_ids = {t.get()  t  tasks}
        task_dict = {t.get(): t  t  tasks}

         task  tasks:
            predecessors = task.get(, [])
             pred_id  predecessors:
                
                 pred_id   task_ids:
                    result.add_error(, )
                    

                
                pred = task_dict.get(pred_id)
                 pred  pred.get()  task.get():
                    pred_end = datetime.fromisoformat(pred[])  (pred[], )  pred[]
                    task_start = datetime.fromisoformat(task[])  (task[], )  task[]

                     task_start < pred_end:
                        result.add_error(
                            ,
                            ,
                            
                        )

     ():
        
        resource_usage = {}

         task  tasks:
            resources = task.get(, [])
            start = task.get()
            end = task.get()

              (resources  start  end):
                

            
             resource  resources:
                res_id = resource.get()  resource.get()
                 res_id   resource_usage:
                    resource_usage[res_id] = []
                resource_usage[res_id].append({
                    : task.get(),
                    : start,
                    : end,
                    : resource.get(, )
                })

        
         res_id, assignments  resource_usage.items():
             (assignments) > :
                
                total_allocation = (a[]  a  assignments)
                 total_allocation > :
                    result.add_warning(
                        ,
                        ,
                        suggestion=
                    )

BIM Data Validation

class BIMDataValidator:
    """Validate BIM export data (IFC, COBie, etc.)."""

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

        # Check element data
        if 'elements' in bim_data:
            self._validate_elements(bim_data['elements'], result)

        # Check property sets
        if 'property_sets' in bim_data:
            self._validate_properties(bim_data['property_sets'], result)

        # Check spatial structure
        if 'spatial_structure' in bim_data:
            self._validate_spatial(bim_data['spatial_structure'], result)

        return result

    def _validate_elements(self, elements: list, result: ValidationResult):
        guids = set()

        for i, elem in enumerate(elements):
            # Check for unique GUIDs
            guid = elem.get('guid')
            if guid in guids:
                result.add_error(f"elements[{i}].guid", f"Duplicate GUID: {guid}")
            guids.add(guid)

            
              elem.get():
                result.add_warning(, )

              elem.get():
                result.add_warning(, )

            
              elem.get()   elem.get():
                result.add_warning(, )

            
             qty_name  [, , ]:
                 qty_name  elem  elem[qty_name] < :
                    result.add_error(, , elem[qty_name])

     ():
         pset  property_sets:
            pset_name = pset.get(, )

            
              pset.get():
                result.add_warning(, )

            
             prop  pset.get(, []):
                 prop.get()  :
                    result.add_info(, )

     ():
        
          spatial.get():
            result.add_warning(, )
          spatial.get():
            result.add_warning(, )
          spatial.get()  (spatial.get(, [])) == :
            result.add_warning(, )

Field Data Validation

class FieldDataValidator:
    """Validate field/site data inputs."""

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

        # Daily report validation
        if field_data.get('type') == 'daily_report':
            self._validate_daily_report(field_data, result)

        # Inspection data
        if field_data.get('type') == 'inspection':
            self._validate_inspection(field_data, result)

        # Progress data
        if field_data.get('type') == 'progress':
            self._validate_progress(field_data, result)

        return result

    def _validate_daily_report(self, data: dict, result: ValidationResult):
        required = ['date', 'weather', 'workforce']
        for field in required:
            if field not in data:
                result.add_error(field, f"Daily report missing '{field}'")

        # Validate workforce
        if 'workforce' in data:
            total = sum(w.get(, )  w  data[])
             total == :
                result.add_warning(, )
             total > :
                result.add_warning(, )

        
           data:
            :
                report_date = datetime.fromisoformat(data[])  (data[], )  data[]
                 report_date > datetime.now():
                    result.add_error(, )
             ValueError:
                result.add_error(, )

     ():
        required = [, , , ]
         field  required:
             field   data:
                result.add_error(field, )

        
        valid_results = [, , , ]
         data.get()  data[].lower()   valid_results:
            result.add_warning(, )

     ():
        
           data:
            pct = data[]
             pct <   pct > :
                result.add_error(, , suggestion=)

        
           data    data:
             data[] < data[]:
                result.add_warning(, ,
                                  )

Usage Examples

# Validate a cost estimate
estimate = {
    'project_name': 'Office Building',
    'estimate_date': '2026-01-15',
    'gross_area': 50000,
    'line_items': [
        {'description': 'Concrete', 'csi_code': '03 30 00', 'quantity': 5000, 'unit_cost': 150},
        {'description': 'Steel', 'csi_code': '05 12 00', 'quantity': 200, 'unit_cost': 2500},
    ],
    'total': 1250000
}

validator = CostEstimateValidator()
result = validator.validate(estimate)
print(result.to_report())

# Validate before processing
if result.is_valid:
    process_estimate(estimate)
else:
    print("Fix errors before processing")
    for error in result.errors:
        print(f"  - {error.field}: {error.message}")

Integration with DDC Pipeline

# Validate all inputs before pipeline execution
def validate_pipeline_inputs(inputs: dict) -> bool:
    validators = {
        'estimate': CostEstimateValidator(),
        'schedule': ScheduleValidator(),
        'bim_data': BIMDataValidator(),
        'field_data': FieldDataValidator()
    }

    all_valid = True
    for input_type, data in inputs.items():
        if input_type in validators:
            result = validators[input_type].validate(data)
            if not result.is_valid:
                print(f"\n{input_type.upper()} VALIDATION FAILED:")
                print(result.to_report())
                all_valid = False

    return all_valid

Resources

• Data Quality Best Practices: Validate early, validate often
• Construction Data Standards: CSI, IFC, COBie specifications
• Error Handling: Always provide actionable suggestions