name: "batch-cad-converter" description: "Batch convert multiple CAD/BIM files (Revit, IFC, DWG, DGN) with progress tracking, error handling, and consolidated reporting." homepage: "https://datadrivenconstruction.io" metadata: {"openclaw":{"emoji":"🔄","os":["win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"],"anyBins":["RvtExporter","DwgExporter","DgnExporter","RVT2IFCconverter"]}}}

Batch CAD/BIM Converter

Business Case

Problem Statement

Large projects and archives contain hundreds or thousands of CAD/BIM files:

• Manual conversion is tedious and error-prone
• Different formats require different converters
• Progress tracking is needed for long operations
• Error handling is critical for large batches

Solution

Unified batch converter handling all supported formats with progress tracking, error recovery, and consolidated reporting.

Business Value

• Multi-format - Revit, IFC, DWG, DGN in one workflow
• Error recovery - Continue on failures
• Progress tracking - Monitor large batches
• Reporting - Consolidated conversion results

Python Implementation

import subprocess
from pathlib import Path
from typing import List, Optional, Dict, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime
import time
import json
from enum import Enum
from concurrent.futures import ThreadPoolExecutor, as_completed


class CADFormat(Enum):
    """Supported CAD/BIM formats."""
    REVIT = (".rvt", ".rfa")
    IFC = (".ifc",)
    DWG = (".dwg",)
    DGN = (".dgn",)


class ConversionStatus(Enum):
    """Status of conversion operation."""
    PENDING = "pending"
    CONVERTING = "converting"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class ConversionResult:
    """Result of single file conversion."""
    input_file: str
    output_file: Optional[str]
    format: str
    status: ConversionStatus
    start_time: datetime
    end_time: Optional[datetime]
    duration_seconds: float
    error_message: Optional[str] = None
    file_size_kb: float = 0


@dataclass
class BatchResult:
    """Result of batch conversion."""
    total_files: int
    successful: int
    failed: int
    skipped: int
    total_duration: float
    results: List[ConversionResult]
    start_time: datetime
    end_time: datetime


class BatchCADConverter:
    """Batch convert multiple CAD/BIM files."""

    # Default converter paths
    DEFAULT_CONVERTERS = {
        'revit': 'RvtExporter.exe',
        'ifc': 'IfcExporter.exe',
        'dwg': 'DwgExporter.exe',
        'dgn': 'DgnExporter.exe'
    }

    def __init__(self, converter_dir: str = ".",
                 converters: Dict[str, str] = None):
        self.converter_dir = Path(converter_dir)
        self.converters = converters or self.DEFAULT_CONVERTERS
        self.results: List[ConversionResult] = []
        self.progress_callback: Optional[Callable] = None

    def set_progress_callback(self, callback: Callable[[int, int, str], None]):
        """Set callback for progress updates."""
        self.progress_callback = callback

    def _get_format(self, file_path: Path) -> Optional[str]:
        """Detect CAD format from extension."""
        ext = file_path.suffix.lower()

        for format_name, extensions in [
            ('revit', ('.rvt', '.rfa')),
            ('ifc', ('.ifc',)),
            ('dwg', ('.dwg',)),
            ('dgn', ('.dgn',))
        ]:
            if ext in extensions:
                return format_name

        return None

    def _get_converter(self, format_name: str) -> Optional[Path]:
        """Get converter path for format."""
        if format_name not in self.converters:
            return None

        converter = self.converter_dir / self.converters[format_name]
        if converter.exists():
            return converter

        # Try in system PATH
        return Path(self.converters[format_name])

    def convert_file(self, input_file: str,
                     output_dir: Optional[str] = None,
                     options: List[str] = None) -> ConversionResult:
        """Convert single file."""

        input_path = Path(input_file)
        start_time = datetime.now()

        # Detect format
        format_name = self._get_format(input_path)
        if not format_name:
            return ConversionResult(
                input_file=input_file,
                output_file=None,
                format='unknown',
                status=ConversionStatus.SKIPPED,
                start_time=start_time,
                end_time=datetime.now(),
                duration_seconds=0,
                error_message="Unsupported format"
            )

        # Get converter
        converter = self._get_converter(format_name)
        if not converter:
            return ConversionResult(
                input_file=input_file,
                output_file=None,
                format=format_name,
                status=ConversionStatus.FAILED,
                start_time=start_time,
                end_time=datetime.now(),
                duration_seconds=0,
                error_message=f"Converter not found for {format_name}"
            )

        # Build command
        cmd = [str(converter), str(input_path)]
        if options:
            cmd.extend(options)

        # Execute
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600)
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()

            # Determine output file
            output_file = input_path.with_suffix('.xlsx')
            if output_dir:
                output_file = Path(output_dir) / output_file.name

            if result.returncode == 0 and output_file.exists():
                return ConversionResult(
                    input_file=input_file,
                    output_file=str(output_file),
                    format=format_name,
                    status=ConversionStatus.SUCCESS,
                    start_time=start_time,
                    end_time=end_time,
                    duration_seconds=duration,
                    file_size_kb=output_file.stat().st_size / 1024
                )
            else:
                return ConversionResult(
                    input_file=input_file,
                    output_file=None,
                    format=format_name,
                    status=ConversionStatus.FAILED,
                    start_time=start_time,
                    end_time=end_time,
                    duration_seconds=duration,
                    error_message=result.stderr or "Conversion failed"
                )

        except subprocess.TimeoutExpired:
            return ConversionResult(
                input_file=input_file,
                output_file=None,
                format=format_name,
                status=ConversionStatus.FAILED,
                start_time=start_time,
                end_time=datetime.now(),
                duration_seconds=3600,
                error_message="Timeout exceeded (1 hour)"
            )

        except Exception as e:
            return ConversionResult(
                input_file=input_file,
                output_file=None,
                format=format_name,
                status=ConversionStatus.FAILED,
                start_time=start_time,
                end_time=datetime.now(),
                duration_seconds=0,
                error_message=str(e)
            )

    def batch_convert(self, input_folder: str,
                      output_folder: Optional[str] = None,
                      include_subfolders: bool = True,
                      formats: List[str] = None,
                      options: Dict[str, List[str]] = None,
                      parallel: bool = False,
                      max_workers: int = 4) -> BatchResult:
        """Convert all files in folder."""

        start_time = datetime.now()
        input_path = Path(input_folder)

        # Find all supported files
        files = []
        pattern = "**/*" if include_subfolders else "*"

        for ext in ['.rvt', '.rfa', '.ifc', '.dwg', '.dgn']:
            files.extend(input_path.glob(f"{pattern}{ext}"))

        # Filter by format if specified
        if formats:
            files = [f for f in files if self._get_format(f) in formats]

        total_files = len(files)
        self.results = []

        # Create output directory
        if output_folder:
            Path(output_folder).mkdir(parents=True, exist_ok=True)

        # Process files
        if parallel and total_files > 1:
            self._convert_parallel(files, output_folder, options, max_workers)
        else:
            self._convert_sequential(files, output_folder, options)

        end_time = datetime.now()

        # Calculate statistics
        successful = sum(1 for r in self.results if r.status == ConversionStatus.SUCCESS)
        failed = sum(1 for r in self.results if r.status == ConversionStatus.FAILED)
        skipped = sum(1 for r in self.results if r.status == ConversionStatus.SKIPPED)

        return BatchResult(
            total_files=total_files,
            successful=successful,
            failed=failed,
            skipped=skipped,
            total_duration=(end_time - start_time).total_seconds(),
            results=self.results,
            start_time=start_time,
            end_time=end_time
        )

    def _convert_sequential(self, files: List[Path],
                            output_folder: Optional[str],
                            options: Dict[str, List[str]]):
        """Convert files sequentially."""

        total = len(files)
        for i, file_path in enumerate(files, 1):
            if self.progress_callback:
                self.progress_callback(i, total, str(file_path))

            format_name = self._get_format(file_path)
            format_options = options.get(format_name, []) if options else []

            result = self.convert_file(str(file_path), output_folder, format_options)
            self.results.append(result)

            status_symbol = "✓" if result.status == ConversionStatus.SUCCESS else "✗"
            print(f"[{i}/{total}] {status_symbol} {file_path.name}")

    def _convert_parallel(self, files: List[Path],
                          output_folder: Optional[str],
                          options: Dict[str, List[str]],
                          max_workers: int):
        """Convert files in parallel."""

        total = len(files)

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {}

            for file_path in files:
                format_name = self._get_format(file_path)
                format_options = options.get(format_name, []) if options else []
                future = executor.submit(self.convert_file, str(file_path), output_folder, format_options)
                futures[future] = file_path

            completed = 0
            for future in as_completed(futures):
                completed += 1
                result = future.result()
                self.results.append(result)

                if self.progress_callback:
                    self.progress_callback(completed, total, str(futures[future]))

    def generate_report(self, batch_result: BatchResult,
                        output_path: str = None) -> str:
        """Generate conversion report."""

        report = {
            'summary': {
                'total_files': batch_result.total_files,
                'successful': batch_result.successful,
                'failed': batch_result.failed,
                'skipped': batch_result.skipped,
                'success_rate': round(batch_result.successful / batch_result.total_files * 100, 1) if batch_result.total_files > 0 else 0,
                'total_duration_seconds': round(batch_result.total_duration, 2),
                'start_time': batch_result.start_time.isoforma