import subprocess
import pandas as pd
from pathlib import Path
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class DGNElementType(Enum):
"""DGN element types."""
CELL_HEADER = 2
LINE = 3
LINE_STRING = 4
SHAPE = 6
TEXT_NODE = 7
CURVE = 11
COMPLEX_CHAIN = 12
COMPLEX_SHAPE = 14
ELLIPSE = 15
ARC = 16
TEXT = 17
SURFACE = 18
SOLID = 19
BSPLINE_CURVE = 21
POINT_STRING = 22
DIMENSION = 33
SHARED_CELL = 35
@dataclass
class DGNElement:
"""Represents a DGN element."""
element_id: int
element_type: int
type_name: str
level: int
color: int
weight: int
style: int
range_low_x: Optional[float] = None
range_low_y: Optional[float] = None
range_low_z: Optional[float] = None
range_high_x: Optional[float] = None
range_high_y: Optional[float] = None
range_high_z: Optional[float] = None
cell_name: Optional[str] = None
text_content: Optional[str] = None
@dataclass
class DGNLevel:
"""Represents a DGN level."""
number: int
name: str
is_displayed: bool
is_frozen: bool
element_count: int
class DGNExporter:
"""DGN to Excel converter using DDC DgnExporter CLI."""
def __init__(self, exporter_path: str = "DgnExporter.exe"):
self.exporter = Path(exporter_path)
if not self.exporter.exists():
raise FileNotFoundError(f"DgnExporter not found: {exporter_path}")
def convert(self, dgn_file: str) -> Path:
"""Convert DGN file to Excel."""
dgn_path = Path(dgn_file)
if not dgn_path.exists():
raise FileNotFoundError(f"DGN file not found: {dgn_file}")
cmd = [str(self.exporter), str(dgn_path)]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Export failed: {result.stderr}")
return dgn_path.with_suffix('.xlsx')
def batch_convert(self, folder: str,
include_subfolders: bool = True) -> List[Dict[str, Any]]:
"""Convert all DGN files in folder."""
folder_path = Path(folder)
pattern = "**/*.dgn" if include_subfolders else "*.dgn"
results = []
for dgn_file in folder_path.glob(pattern):
try:
output = self.convert(str(dgn_file))
results.append({
'input': str(dgn_file),
'output': str(output),
'status': 'success'
})
print(f"✓ Converted: {dgn_file.name}")
except Exception as e:
results.append({
'input': str(dgn_file),
'output': None,
'status': 'failed',
'error': str(e)
})
print(f"✗ Failed: {dgn_file.name} - {e}")
return results
def read_elements(self, xlsx_file: str) -> pd.DataFrame:
"""Read converted Excel as DataFrame."""
return pd.read_excel(xlsx_file, sheet_name="Elements")
def get_levels(self, xlsx_file: str) -> pd.DataFrame:
"""Get level summary."""
df = self.read_elements(xlsx_file)
if 'Level' not in df.columns:
raise ValueError("Level column not found")
summary = df.groupby('Level').agg({
'ElementId': 'count'
}).reset_index()
summary.columns = ['Level', 'Element_Count']
return summary.sort_values('Level')
def get_element_types(self, xlsx_file: str) -> pd.DataFrame:
"""Get element type statistics."""
df = self.read_elements(xlsx_file)
type_col = 'ElementType' if 'ElementType' in df.columns else 'Type'
if type_col not in df.columns:
return pd.DataFrame()
summary = df.groupby(type_col).agg({
'ElementId': 'count'
}).reset_index()
summary.columns = ['Element_Type', 'Count']
return summary.sort_values('Count', ascending=False)
def get_cells(self, xlsx_file: str) -> pd.DataFrame:
"""Get cell references (similar to blocks in DWG)."""
df = self.read_elements(xlsx_file)
cells = df[df['ElementType'].isin([2, 35])]
if cells.empty or 'CellName' not in cells.columns:
return pd.DataFrame(columns=['Cell_Name', 'Count'])
summary = cells.groupby('CellName').agg({
'ElementId': 'count'
}).reset_index()
summary.columns = ['Cell_Name', 'Count']
return summary.sort_values('Count', ascending=False)
def get_text_content(self, xlsx_file: str) -> pd.DataFrame:
"""Extract all text from DGN."""
df = self.read_elements(xlsx_file)
text_types = [7, 17]
texts = df[df['ElementType'].isin(text_types)]
if 'TextContent' in texts.columns:
return texts[['ElementId', 'Level', 'TextContent']].copy()
return texts[['ElementId', 'Level']].copy()
def get_statistics(self, xlsx_file: str) -> Dict[str, Any]:
"""Get comprehensive DGN statistics."""
df = self.read_elements(xlsx_file)
stats = {
'total_elements': len(df),
'levels_used': df['Level'].nunique() if 'Level' in df.columns else 0,
'element_types': df['ElementType'].nunique() if 'ElementType' in df.columns else 0
}
for coord in ['X', 'Y', 'Z']:
low_col = f'RangeLow{coord}'
high_col = f'RangeHigh{coord}'
if low_col in df.columns and high_col in df.columns:
stats[f'min_{coord.lower()}'] = df[low_col].min()
stats[f'max_{coord.lower()}'] = df[high_col].max()
return stats
class DGNAnalyzer:
"""Advanced DGN analysis for infrastructure projects."""
def __init__(self, exporter: DGNExporter):
self.exporter = exporter
def analyze_infrastructure(self, dgn_file: str) -> Dict[str, Any]:
"""Analyze DGN for infrastructure elements."""
xlsx = self.exporter.convert(dgn_file)
df = self.exporter.read_elements(str(xlsx))
analysis = {
'file': dgn_file,
'statistics': self.exporter.get_statistics(str(xlsx)),
'levels': self.exporter.get_levels(str(xlsx)).to_dict('records'),
'element_types': self.exporter.get_element_types(str(xlsx)).to_dict('records'),
'cells': self.exporter.get_cells(str(xlsx)).to_dict('records')
}
if 'ElementType' in df.columns:
lines = df[df['ElementType'].isin([3, 4, 6, 14])].shape[0]
analysis['linear_elements'] = lines
complex_elements = df[df['ElementType'].isin([12, 14, 18, 19])].shape[0]
analysis['complex_elements'] = complex_elements
annotations = df[df['ElementType'].isin([7, 17, 33])].shape[0]
analysis['annotations'] = annotations
return analysis
def compare_revisions(self, dgn1: str, dgn2: str) -> Dict[str, Any]:
"""Compare two DGN revisions."""
xlsx1 = self.exporter.convert(dgn1)
xlsx2 = self.exporter.convert(dgn2)
df1 = self.exporter.read_elements(str(xlsx1))
df2 = self.exporter.read_elements(str(xlsx2))
levels1 = set(df1['Level'].unique()) if 'Level' in df1.columns else set()
levels2 = set(df2['Level'].unique()) if 'Level' in df2.columns else set()
return {
'revision1': dgn1,
'revision2': dgn2,
'element_count_diff': len(df2) - len(df1),
'levels_added': list(levels2 - levels1),
'levels_removed': list(levels1 - levels2),
'common_levels': len(levels1 & levels2)
}
def extract_coordinates(self, xlsx_file: str) -> pd.DataFrame:
"""Extract element coordinates for GIS integration."""
df = self.exporter.read_elements(xlsx_file)
coord_cols = ['ElementId', 'Level', 'ElementType']
for col in ['RangeLowX', 'RangeLowY', 'RangeLowZ',
'RangeHighX', 'RangeHighY', 'RangeHighZ',
'CenterX', 'CenterY', 'CenterZ']:
if col in df.columns:
coord_cols.append(col)
return df[coord_cols].copy()
class DGNLevelManager:
"""Manage DGN level structures."""
def __init__(self, exporter: DGNExporter):
self.exporter = exporter
def get_level_map(self, xlsx_file: str) -> Dict[int, str]:
"""Create level number to name mapping."""
df = self.exporter.read_elements(xlsx_file)
if 'Level' not in df.columns:
return {}
level_map = {}
for level in df['Level'].unique():
level_map[int(level)] = f"Level_{level}"
return level_map
def filter_by_levels(self, xlsx_file: str,
levels: List[int]) -> pd.DataFrame:
"""Filter elements by level numbers."""
df = self.exporter.read_elements(xlsx_file)
return df[df['Level'].isin(levels)]
def get_level_usage_report(self, xlsx_file: str) -> pd.DataFrame:
"""Generate level usage report."""
df = self.exporter.read_elements(xlsx_file)
if 'Level' not in df.columns or 'ElementType' not in df.columns:
return pd.DataFrame()
report = pd.crosstab(df['Level'], df['ElementType'], margins=True)
return report
def convert_dgn_to_excel(dgn_file: str,
exporter_path: str = "DgnExporter.exe") -> str:
"""Quick conversion of DGN to Excel."""
exporter = DGNExporter(exporter_path)
output = exporter.convert(dgn_file)
return str(output)
def analyze_dgn(dgn_file: str,
exporter_path: str = "DgnExporter.exe") -> Dict[str, Any]:
"""Analyze DGN file and return summary."""
exporter = DGNExporter(exporter_path)
analyzer = DGNAnalyzer(exporter)
return analyzer.analyze_infrastructure(dgn_file)
