Clash Resolution Analyzer

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
from enum import Enum
from datetime import datetime
from collections import defaultdict

class ClashPriority(Enum):
    CRITICAL = 1  # Must resolve before construction
    HIGH = 2      # Resolve in next coordination cycle
    MEDIUM = 3    # Resolve before trade starts
    LOW = 4       # Minor, can resolve in field

class ClashStatus(Enum):
    NEW = "new"
    ASSIGNED = "assigned"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    APPROVED = "approved"
    VOID = "void"  # Not a real clash

class ResolutionType(Enum):
    ROUTE_AROUND = "Route around obstruction"
    RAISE_LOWER = "Raise or lower element"
    RESIZE = "Resize element"
    RELOCATE = "Relocate element"
    STRUCTURAL_MOD = "Structural modification required"
    DESIGN_CHANGE = "Design change required"
    NO_CLASH = "Not a real clash (tolerance)"
    SEQUENCE = "Resolve by construction sequence"

@dataclass
class ClashElement:
    id: str
    name: str
    category: str
    discipline: str
    level: str
    system: str

@dataclass
class Clash:
    id: str
    name: str
    element1: ClashElement
    element2: ClashElement
    location: Tuple[float, float, float]
    distance: float  # Negative = hard clash, positive = clearance violation
    clash_type: str  # hard, clearance, duplicate
    priority: ClashPriority = ClashPriority.MEDIUM
    status: ClashStatus = ClashStatus.NEW
    assigned_to: str = ""
    resolution_type: Optional[ResolutionType] = None
    resolution_notes: str = ""
    created_date: datetime = field(default_factory=datetime.now)
    resolved_date: Optional[datetime] = None

@dataclass
class ClashPattern:
    pattern_type: str
    disciplines: Tuple[str, str]
    systems: Tuple[str, str]
    clash_count: int
    example_clashes: List[str]
    suggested_resolution: str
    root_cause: str

@dataclass
class ClashReport:
    report_name: str
    total_clashes: int
    new_clashes: int
    resolved_clashes: int
    clashes_by_priority: Dict[str, int]
    clashes_by_discipline: Dict[str, int]
    clashes_by_status: Dict[str, int]
    patterns: List[ClashPattern]
    resolution_rate: float

class ClashResolutionAnalyzer:
    """Analyze and manage BIM clash detection results."""

    # Discipline priority for resolution responsibility
    DISCIPLINE_PRIORITY = {
        'Structural': 1,
        'Architectural': 2,
        'Mechanical': 3,
        'Plumbing': 4,
        'Electrical': 5,
        'Fire Protection': 6,
    }

    # Common resolution strategies by clash type
    RESOLUTION_STRATEGIES = {
        ('Mechanical', 'Structural'): {
            'strategy': ResolutionType.ROUTE_AROUND,
            'responsible': 'Mechanical',
            'notes': 'MEP typically routes around structure'
        },
        ('Plumbing', 'Structural'): {
            'strategy': ResolutionType.ROUTE_AROUND,
            'responsible': 'Plumbing',
            'notes': 'Coordinate sleeves/penetrations with SE'
        },
        ('Electrical', 'Mechanical'): {
            'strategy': ResolutionType.RAISE_LOWER,
            'responsible': 'Electrical',
            'notes': 'Conduit typically more flexible than ductwork'
        },
        ('Mechanical', 'Mechanical'): {
            'strategy': ResolutionType.RESIZE,
            'responsible': 'Mechanical',
            'notes': 'Review duct sizing and routing options'
        },
        ('Fire Protection', 'Mechanical'): {
            'strategy': ResolutionType.ROUTE_AROUND,
            'responsible': 'Fire Protection',
            'notes': 'Sprinkler typically routes around major duct'
        },
    }

    def __init__(self):
        self.clashes: Dict[str, Clash] = {}
        self.patterns: List[ClashPattern] = []
        self.history: List[Dict] = []

    def import_clashes(self, clash_data: List[Dict]) -> int:
        """Import clashes from Navisworks or other clash detection software."""
        count = 0

        for data in clash_data:
            clash = Clash(
                id=data.get('id', f'CLH-{count}'),
                name=data.get('name', ''),
                element1=ClashElement(
                    id=data.get('element1_id', ''),
                    name=data.get('element1_name', ''),
                    category=data.get('element1_category', ''),
                    discipline=data.get('element1_discipline', ''),
                    level=data.get('element1_level', ''),
                    system=data.get('element1_system', '')
                ),
                element2=ClashElement(
                    id=data.get('element2_id', ''),
                    name=data.get('element2_name', ''),
                    category=data.get('element2_category', ''),
                    discipline=data.get('element2_discipline', ''),
                    level=data.get('element2_level', ''),
                    system=data.get('element2_system', '')
                ),
                location=(
                    data.get('x', 0),
                    data.get('y', 0),
                    data.get('z', 0)
                ),
                distance=data.get('distance', 0),
                clash_type=data.get('clash_type', 'hard')
            )

            # Auto-prioritize
            clash.priority = self._auto_prioritize(clash)

            # Auto-assign
            clash.assigned_to = self._auto_assign(clash)

            self.clashes[clash.id] = clash
            count += 1

        return count

    def _auto_prioritize(self, clash: Clash) -> ClashPriority:
        """Automatically prioritize clash based on characteristics."""
        # Hard clashes with structure are critical
        if clash.element1.discipline == 'Structural' or clash.element2.discipline == 'Structural':
            if clash.clash_type == 'hard':
                return ClashPriority.CRITICAL

        # Large penetration clashes
        if abs(clash.distance) > 0.1:  # More than 100mm overlap
            return ClashPriority.HIGH

        # MEP-MEP clashes
        mep_disciplines = ['Mechanical', 'Electrical', 'Plumbing', 'Fire Protection']
        if clash.element1.discipline in mep_disciplines and clash.element2.discipline in mep_disciplines:
            return ClashPriority.MEDIUM

        # Clearance violations
        if clash.clash_type == 'clearance':
            return ClashPriority.LOW

        return ClashPriority.MEDIUM

    def _auto_assign(self, clash: Clash) -> str:
        """Automatically assign responsibility based on discipline priority."""
        d1 = clash.element1.discipline
        d2 = clash.element2.discipline

        # Check for known resolution strategy
        key = (d1, d2) if (d1, d2) in self.RESOLUTION_STRATEGIES else (d2, d1)
        if key in self.RESOLUTION_STRATEGIES:
            return self.RESOLUTION_STRATEGIES[key]['responsible']

        # Default to lower priority discipline (typically more flexible)
        p1 = self.DISCIPLINE_PRIORITY.get(d1, 10)
        p2 = self.DISCIPLINE_PRIORITY.get(d2, 10)

        return d2 if p2 > p1 else d1

    def analyze_patterns(self) -> List[ClashPattern]:
        """Identify patterns in clashes."""
        patterns = []

        # Group by discipline pair
        discipline_pairs = defaultdict(list)
        for clash in self.clashes.values():
            pair = tuple(sorted([clash.element1.discipline, clash.element2.discipline]))
            discipline_pairs[pair].append(clash)

        for (d1, d2), clashes in discipline_pairs.items():
            if len(clashes) >= 3:  # Pattern threshold
                # Further group by system
                system_pairs = defaultdict(list)
                for clash in clashes:
                    sys_pair = tuple(sorted([clash.element1.system, clash.element2.system]))
                    system_pairs[sys_pair].append(clash)

                for (s1, s2), sys_clashes in system_pairs.items():
                    if len(sys_clashes) >= 2:
                        # Get resolution strategy
                        key = (d1, d2) if (d1, d2) in self.RESOLUTION_STRATEGIES else (d2, d1)
                        strategy = self.RESOLUTION_STRATEGIES.get(key, {})

                        patterns.append(ClashPattern(
                            pattern_type=f"{d1} vs {d2}",
                            disciplines=(d1, d2),
                            systems=(s1, s2),
                            clash_count=len(sys_clashes),
                            example_clashes=[c.id for c in sys_clashes[:3]],
                            suggested_resolution=strategy.get('strategy', ResolutionType.ROUTE_AROUND).value,
                            root_cause=f"Coordination needed between {s1} and {s2} systems"
                        ))

        self.patterns = sorted(patterns, key=lambda p: -p.clash_count)
        return self.patterns

    def suggest_resolution(self, clash_id: str) -> Dict:
        """Suggest resolution for a specific clash."""
        if clash_id not in self.clashes:
            return {'error': 'Clash not found'}

        clash = self.clashes[clash_id]
        d1, d2 = clash.element1.discipline, clash.element2.discipline

        # Get strategy
        key = (d1, d2) if (d1, d2) in self.RESOLUTION_STRATEGIES else (d2, d1)
        strategy = self.RESOLUTION_STRATEGIES.get(key, {})

        suggestion = {
            'clash_id': clash_id,
            'resolution_type': strategy.get('strategy', ResolutionType.ROUTE_AROUND),
            'responsible_discipline': strategy.get('responsible', self._auto_assign(clash)),
            'notes': strategy.get('notes', 'Review and coordinate'),
            'similar_clashes': [],
        }

        # Find similar clashes
        for pattern in self.patterns:
            if d1 in pattern.disciplines and d2 in pattern.disciplines:
                suggestion['similar_clashes'] = pattern.example_clashes
                suggestion['pattern_root_cause'] = pattern.root_cause
                break

        return suggestion

    def update_clash_status(self, clash_id: str, status: ClashStatus,
                            resolution_type: ResolutionType = None,
                            notes: str = "") -> bool:
        """Update clash status."""
        if clash_id not in self.clashes:
            return False

        clash = self.clashes[clash_id]
        old_status = clash.status

        clash.status = status
        if resolution_type:
            clash.resolution_type = resolution_type
        if notes:
            clash.resolution_notes = notes
        if status in [ClashStatus.RESOLVED, ClashStatus.APPROVED]:
            clash.resolved_date = datetime.now()

        # Track history
        self.history.append({
            'clash_id': clash_id,
            'timestamp': datetime.now(),
            'old_status': old_status.value,
            'new_status': status.value,
            'notes': notes
        })

        return True

    def get_clashes_by_discipline(self, discipline: str) -> List[Clash]:
        """Get all clashes assigned to a discipline."""
        return [c for c in self.clashes.values() if c.assigned_to == discipline]

    def get_clashes_by_level(self, level: str) -> List[Clash]:
        """Get all clashes on a specific level."""
        return [c for c in self.clashes.values()
                if c.element1.level == level or c.element2.level == level]

    def generate_coordination_matrix(self) -> Dict[str, Dict[str, int]]:
        """Generate matrix showing clashes between disciplines."""
        matrix = defaultdict(lambda: defaultdict(int))

        for clash in self.clashes.values():
            d1 = clash.element1.discipline
            d2 = clash.element2.discipline
            matrix[d1][d2] += 1
            if d1 != d2:
                matrix[d2][d1] += 1

        return dict(matrix)

    def generate_report(self) -> ConsistencyReport:
        """Generate comprehensive clash analysis report."""
        clashes_by_priority = defaultdict(int)
        clashes_by_discipline = defaultdict(int)
        clashes_by_status = defaultdict(int)

        for clash in self.clashes.values():
            clashes_by_priority[clash.priority.name] += 1
            clashes_by_discipline[clash.assigned_to] += 1
            clashes_by_status[clash.status.value] += 1

        resolved = clashes_by_status.get('resolved', 0) + clashes_by_status.get('approved', 0)
        resolution_rate = resolved / len(self.clashes) * 100 if self.clashes else 0

        return ClashReport(
            report_name=f"Clash Report {datetime.now().strftime('%Y-%m-%d')}",
            total_clashes=len(self.clashes),
            new_clashes=clashes_by_status.get('new', 0),
            resolved_clashes=resolved,
            clashes_by_priority=dict(clashes_by_priority),
            clashes_by_discipline=dict(clashes_by_discipline),
            clashes_by_status=dict(clashes_by_status),
            patterns=self.patterns,
            resolution_rate=resolution_rate
        )

    def generate_report_markdown(self) -> str:
        """Generate markdown report."""
        report = self.generate_report()

        lines = ["# Clash Resolution Report", ""]
        lines.append(f"**Date:** {datetime.now().strftime('%Y-%m-%d')}")
        lines.append(f"**Total Clashes:** {report.total_clashes}")
        lines.append(f"**Resolution Rate:** {report.resolution_rate:.1f}%")
        lines.append("")

        # By status
        lines.append("## Status Summary")
        for status, count in report.clashes_by_status.items():
            lines.append(f"- {status}: {count}")
        lines.append("")

        # By priority
        lines.append("## Priority Breakdown")
        for priority, count in sorted(report.clashes_by_priority.items()):
            lines.append(f"- {priority}: {count}")
        lines.append("")

        # By discipline
        lines.append("## By Responsible Discipline")
        for disc, count in sorted(report.clashes_by_discipline.items(), key=lambda x: -x[1]):
            lines.append(f"- {disc}: {count}")
        lines.append("")

        # Patterns
        if report.patterns:
            lines.append("## Clash Patterns Identified")
            for pattern in report.patterns[:5]:
                lines.append(f"\n### {pattern.pattern_type}")
                lines.append(f"- **Count:** {pattern.clash_count} clashes")
                lines.append(f"- **Systems:** {pattern.systems[0]} vs {pattern.systems[1]}")
                lines.append(f"- **Root Cause:** {pattern.root_cause}")
                lines.append(f"- **Suggested Resolution:** {pattern.suggested_resolution}")

        # Critical clashes
        critical = [c for c in self.clashes.values() if c.priority == ClashPriority.CRITICAL and c.status == ClashStatus.NEW]
        if critical:
            lines.append("\n## Critical Unresolved Clashes")
            for clash in critical[:10]:
                lines.append(f"- **{clash.id}**: {clash.element1.name} vs {clash.element2.name}")
                lines.append(f"  - Location: Level {clash.element1.level}")
                lines.append(f"  - Assigned: {clash.assigned_to}")

        return "\n".join(lines)

# Initialize analyzer
analyzer = ClashResolutionAnalyzer()

# Import clashes (from Navisworks export)
clash_data = [
    {
        'id': 'CLH-001',
        'name': 'Duct vs Beam',
        'element1_discipline': 'Mechanical',
        'element1_system': 'Supply Air',
        'element1_level': 'Level 2',
        'element2_discipline': 'Structural',
        'element2_system': 'Steel Frame',
        'element2_level': 'Level 2',
        'distance': -0.15,
        'clash_type': 'hard'
    }
]

count = analyzer.import_clashes(clash_data)
print(f"Imported {count} clashes")

# Analyze patterns
patterns = analyzer.analyze_patterns()
for pattern in patterns:
    print(f"Pattern: {pattern.pattern_type} - {pattern.clash_count} clashes")

# Get resolution suggestion
suggestion = analyzer.suggest_resolution('CLH-001')
print(f"Suggested resolution: {suggestion['resolution_type'].value}")
print(f"Responsible: {suggestion['responsible_discipline']}")

# Update status
analyzer.update_clash_status(
    'CLH-001',
    ClashStatus.RESOLVED,
    ResolutionType.ROUTE_AROUND,
    'Duct rerouted below beam'
)

# Generate report
print(analyzer.generate_report_markdown())

pip install (no external dependencies)