batch-cad-converter

CAD/BIM バッチ変換ツール

ビジネスケース

問題

大規模なプロジェクトやアーカイブには、数百から数千の CAD/BIM ファイルが含まれています。

• 手動での変換は手間がかかり、エラーが発生しやすいです
• 形式ごとに異なるコンバーターが必要です
• 時間のかかる処理では、進捗状況の追跡が必要です
• 大量のファイルを一括処理する場合、エラー処理が重要です

ソリューション

進捗状況の追跡、エラーからの復旧、結果の一元的なレポート作成に対応した、すべての対応形式を処理する統合バッチコンバーターです。

ビジネス価値

• 複数形式に対応 - Revit、IFC、DWG、DGN を 1 つのワークフローで処理
• エラーからの復旧 - 失敗しても処理を続行
• 進捗状況の追跡 - 大規模なバッチ処理を監視
• レポート作成 - 変換結果を一元的に集約

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.isoformat(),
                'end_time': batch_result.end_time.isoformat()
            },
            'results': [
                {
                    'input': r.input_file,
                    'output': r.output_file,
                    'format': r.format,
                    'status': r.status.value,
                    'duration': round(r.duration_seconds, 2),
                    'error': r.error_message
                }
                for r in batch_result.results
            ]
        }

        report_json = json.dumps(report, indent=2)

        if output_path:
            with open(output_path, 'w') as f:
                f.write(report_json)

        return report_json


# Progress callback example
def print_progress(current: int, total: int, file_name: str):
    """Print progress to console."""
    percent = current / total * 100
    print(f"Progress: {current}/{total} ({percent:.1f}%) - {file_name}")

クイックスタート

# Initialize batch converter
converter = BatchCADConverter(converter_dir="C:/DDC/")

# Set progress callback
converter.set_progress_callback(print_progress)

# Convert all files
result = converter.batch_convert(
    input_folder="C:/Projects",
    output_folder="C:/Converted",
    include_subfolders=True
)

print(f"Success: {result.successful}/{result.total_files}")

よくあるユースケース

1. 特定の形式のみを変換

result = converter.batch_convert(
    input_folder="C:/Archive",
    formats=['revit', 'ifc'],  # Only Revit and IFC
    parallel=True,
    max_workers=4
)

2. 形式固有のオプションを指定

options = {
    'revit': ['complete', 'bbox', 'rooms'],
    'ifc': ['bbox'],
    'dwg': []
}
result = converter.batch_convert(
    input_folder="C:/Projects",
    options=options
)

3. レポートを生成

result = converter.batch_convert("C:/Projects")
report = converter.generate_report(result, "conversion_report.json")

リソース

• GitHub： cad2data Pipeline