--- 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

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

--- 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

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

--- 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

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

--- 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

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