Files
fletimageanalysis/dxf_processor_fixed.py
2025-07-16 17:33:20 +09:00

638 lines
26 KiB
Python

# -*- coding: utf-8 -*-
"""
수정된 DXF 파일 처리 모듈 - 속성 추출 문제 해결
ezdxf 라이브러리를 사용하여 DXF 파일에서 모든 속성을 정확히 추출
"""
import os
import json
import logging
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass, asdict, field
try:
import ezdxf
from ezdxf.document import Drawing
from ezdxf.entities import Insert, Attrib, AttDef, Text, MText
from ezdxf.layouts import BlockLayout, Modelspace
from ezdxf import bbox, disassemble
EZDXF_AVAILABLE = True
except ImportError:
EZDXF_AVAILABLE = False
logging.warning("ezdxf 라이브러리가 설치되지 않았습니다. DXF 기능이 비활성화됩니다.")
from config import Config
@dataclass
class BoundingBox:
"""바운딩 박스 정보를 담는 데이터 클래스"""
min_x: float
min_y: float
max_x: float
max_y: float
@property
def width(self) -> float:
return self.max_x - self.min_x
@property
def height(self) -> float:
return self.max_y - self.min_y
@property
def center(self) -> Tuple[float, float]:
return ((self.min_x + self.max_x) / 2, (self.min_y + self.max_y) / 2)
def merge(self, other: 'BoundingBox') -> 'BoundingBox':
"""다른 바운딩 박스와 병합하여 가장 큰 외곽 박스 반환"""
return BoundingBox(
min_x=min(self.min_x, other.min_x),
min_y=min(self.min_y, other.min_y),
max_x=max(self.max_x, other.max_x),
max_y=max(self.max_y, other.max_y)
)
@dataclass
class AttributeInfo:
"""속성 정보를 담는 데이터 클래스"""
tag: str
text: str
position: Tuple[float, float, float]
height: float
rotation: float
layer: str
prompt: Optional[str] = None
style: Optional[str] = None
invisible: bool = False
const: bool = False
bounding_box: Optional[BoundingBox] = None
entity_handle: Optional[str] = None
# 좌표 정보
insert_x: float = 0.0
insert_y: float = 0.0
insert_z: float = 0.0
@dataclass
class BlockInfo:
"""블록 정보를 담는 데이터 클래스"""
name: str
position: Tuple[float, float, float]
scale: Tuple[float, float, float]
rotation: float
layer: str
attributes: List[AttributeInfo]
bounding_box: Optional[BoundingBox] = None
@dataclass
class TitleBlockInfo:
"""도곽 정보를 담는 데이터 클래스"""
drawing_name: Optional[str] = None
drawing_number: Optional[str] = None
construction_field: Optional[str] = None
construction_stage: Optional[str] = None
scale: Optional[str] = None
project_name: Optional[str] = None
designer: Optional[str] = None
date: Optional[str] = None
revision: Optional[str] = None
location: Optional[str] = None
bounding_box: Optional[BoundingBox] = None
block_name: Optional[str] = None
# 모든 attributes 정보 저장
all_attributes: List[AttributeInfo] = field(default_factory=list)
attributes_count: int = 0
def __post_init__(self):
"""초기화 후 처리"""
self.attributes_count = len(self.all_attributes)
class FixedDXFProcessor:
"""수정된 DXF 파일 처리기 - 속성 추출 문제 해결"""
def __init__(self):
self.logger = logging.getLogger(self.__class__.__name__)
# 도곽 식별을 위한 키워드 (한국어/영어)
self.title_block_keywords = {
'도면명': ['도면명', '도면', 'drawing', 'title', 'name', 'dwg'],
'도면번호': ['도면번호', '번호', 'number', 'no', 'dwg_no'],
'건설분야': ['건설분야', '분야', 'field', 'construction', 'civil'],
'건설단계': ['건설단계', '단계', 'stage', 'phase', 'step'],
'축척': ['축척', 'scale', 'ratio'],
'설계자': ['설계자', '설계', 'designer', 'design', 'engineer'],
'프로젝트': ['프로젝트', '사업', 'project', 'work'],
'날짜': ['날짜', '일자', 'date', 'time'],
'리비전': ['리비전', '개정', 'revision', 'rev'],
'위치': ['위치', '장소', 'location', 'place', 'site']
}
if not EZDXF_AVAILABLE:
self.logger.warning("ezdxf 라이브러리가 설치되지 않았습니다.")
def validate_dxf_file(self, file_path: str) -> bool:
"""DXF 파일 유효성 검사"""
if not EZDXF_AVAILABLE:
self.logger.error("ezdxf 라이브러리가 설치되지 않음")
return False
if not os.path.exists(file_path):
self.logger.error(f"DXF 파일이 존재하지 않음: {file_path}")
return False
if not file_path.lower().endswith('.dxf'):
self.logger.error(f"DXF 파일이 아님: {file_path}")
return False
try:
# 파일 열기 시도
doc = ezdxf.readfile(file_path)
self.logger.info(f"DXF 파일 유효성 검사 통과: {file_path}")
return True
except Exception as e:
self.logger.error(f"DXF 파일 유효성 검사 실패: {e}")
return False
def load_dxf_document(self, file_path: str) -> Optional[Drawing]:
"""DXF 문서 로드"""
try:
doc = ezdxf.readfile(file_path)
self.logger.info(f"DXF 문서 로드 성공: {file_path}")
return doc
except Exception as e:
self.logger.error(f"DXF 문서 로드 실패: {e}")
return None
def extract_all_insert_attributes(self, doc: Drawing) -> List[BlockInfo]:
"""모든 INSERT 엔티티에서 속성 추출 - 수정된 로직"""
block_refs = []
try:
# 모델스페이스에서 INSERT 엔티티 검색
msp = doc.modelspace()
inserts = msp.query('INSERT')
self.logger.info(f"모델스페이스에서 {len(inserts)}개의 INSERT 엔티티 발견")
for insert in inserts:
block_info = self._process_insert_entity(insert, doc)
if block_info:
block_refs.append(block_info)
# 페이퍼스페이스에서도 검색
for layout in doc.layouts:
if layout.is_any_paperspace:
inserts = layout.query('INSERT')
self.logger.info(f"페이퍼스페이스 '{layout.name}'에서 {len(inserts)}개의 INSERT 엔티티 발견")
for insert in inserts:
block_info = self._process_insert_entity(insert, doc)
if block_info:
block_refs.append(block_info)
self.logger.info(f"{len(block_refs)}개의 블록 참조 처리 완료")
return block_refs
except Exception as e:
self.logger.error(f"INSERT 속성 추출 중 오류: {e}")
return []
def _process_insert_entity(self, insert: Insert, doc: Drawing) -> Optional[BlockInfo]:
"""개별 INSERT 엔티티 처리 - 향상된 속성 추출"""
try:
attributes = []
# 방법 1: INSERT에 연결된 ATTRIB 엔티티들 추출
self.logger.debug(f"INSERT '{insert.dxf.name}'의 연결된 속성 추출 중...")
# insert.attribs는 리스트를 반환
insert_attribs = insert.attribs
self.logger.debug(f"INSERT.attribs: {len(insert_attribs)}개 발견")
for attrib in insert_attribs:
attr_info = self._extract_attrib_info(attrib)
if attr_info:
attributes.append(attr_info)
self.logger.debug(f"ATTRIB 추출: tag='{attr_info.tag}', text='{attr_info.text}'")
# 방법 2: 블록 정의에서 ATTDEF 정보 추출 (search_const=True와 유사한 효과)
try:
block_layout = doc.blocks.get(insert.dxf.name)
if block_layout:
# 블록 정의에서 ATTDEF 엔티티 검색
attdefs = block_layout.query('ATTDEF')
self.logger.debug(f"블록 정의에서 {len(attdefs)}개의 ATTDEF 발견")
for attdef in attdefs:
# ATTDEF에서 기본값이나 상수 값 추출
if hasattr(attdef.dxf, 'text') and attdef.dxf.text.strip():
attr_info = self._extract_attdef_info(attdef, insert)
if attr_info:
# 중복 체크 (같은 tag가 이미 있으면 건너뛰기)
existing_tags = [attr.tag for attr in attributes]
if attr_info.tag not in existing_tags:
attributes.append(attr_info)
self.logger.debug(f"ATTDEF 추출: tag='{attr_info.tag}', text='{attr_info.text}'")
# 방법 3: 블록 내부의 TEXT/MTEXT 엔티티도 추출
text_entities = block_layout.query('TEXT MTEXT')
self.logger.debug(f"블록 정의에서 {len(text_entities)}개의 TEXT/MTEXT 발견")
for text_entity in text_entities:
attr_info = self._extract_text_as_attribute(text_entity, insert)
if attr_info:
attributes.append(attr_info)
self.logger.debug(f"TEXT 추출: text='{attr_info.text}'")
except Exception as e:
self.logger.warning(f"블록 정의 처리 중 오류: {e}")
# 방법 4: get_attrib_text 메서드를 사용한 확인
try:
# 일반적인 속성 태그들로 시도
common_tags = ['TITLE', 'NAME', 'NUMBER', 'DATE', 'SCALE', 'DESIGNER',
'PROJECT', 'DRAWING', 'DWG_NO', 'REV', 'REVISION']
for tag in common_tags:
try:
# search_const=True로 ATTDEF도 검색
text_value = insert.get_attrib_text(tag, search_const=True)
if text_value and text_value.strip():
# 이미 있는 태그인지 확인
existing_tags = [attr.tag for attr in attributes]
if tag not in existing_tags:
attr_info = AttributeInfo(
tag=tag,
text=text_value.strip(),
position=insert.dxf.insert,
height=12.0, # 기본값
rotation=insert.dxf.rotation,
layer=insert.dxf.layer,
insert_x=insert.dxf.insert[0],
insert_y=insert.dxf.insert[1],
insert_z=insert.dxf.insert[2] if len(insert.dxf.insert) > 2 else 0.0
)
attributes.append(attr_info)
self.logger.debug(f"get_attrib_text 추출: tag='{tag}', text='{text_value.strip()}'")
except:
continue
except Exception as e:
self.logger.warning(f"get_attrib_text 처리 중 오류: {e}")
# BlockInfo 생성
if attributes or True: # 속성이 없어도 블록 정보는 수집
block_info = BlockInfo(
name=insert.dxf.name,
position=insert.dxf.insert,
scale=(insert.dxf.xscale, insert.dxf.yscale, insert.dxf.zscale),
rotation=insert.dxf.rotation,
layer=insert.dxf.layer,
attributes=attributes
)
self.logger.info(f"INSERT '{insert.dxf.name}' 처리 완료: {len(attributes)}개 속성")
return block_info
return None
except Exception as e:
self.logger.error(f"INSERT 엔티티 처리 중 오류: {e}")
return None
def _extract_attrib_info(self, attrib: Attrib) -> Optional[AttributeInfo]:
"""ATTRIB 엔티티에서 속성 정보 추출"""
try:
# 텍스트가 비어있으면 건너뛰기
text_value = attrib.dxf.text.strip()
if not text_value:
return None
attr_info = AttributeInfo(
tag=attrib.dxf.tag,
text=text_value,
position=attrib.dxf.insert,
height=getattr(attrib.dxf, 'height', 12.0),
rotation=getattr(attrib.dxf, 'rotation', 0.0),
layer=getattr(attrib.dxf, 'layer', '0'),
style=getattr(attrib.dxf, 'style', None),
invisible=getattr(attrib, 'is_invisible', False),
const=getattr(attrib, 'is_const', False),
entity_handle=attrib.dxf.handle,
insert_x=attrib.dxf.insert[0],
insert_y=attrib.dxf.insert[1],
insert_z=attrib.dxf.insert[2] if len(attrib.dxf.insert) > 2 else 0.0
)
return attr_info
except Exception as e:
self.logger.warning(f"ATTRIB 정보 추출 중 오류: {e}")
return None
def _extract_attdef_info(self, attdef: AttDef, insert: Insert) -> Optional[AttributeInfo]:
"""ATTDEF 엔티티에서 속성 정보 추출"""
try:
# 텍스트가 비어있으면 건너뛰기
text_value = attdef.dxf.text.strip()
if not text_value:
return None
# INSERT의 위치를 기준으로 실제 위치 계산
actual_position = (
insert.dxf.insert[0] + attdef.dxf.insert[0] * insert.dxf.xscale,
insert.dxf.insert[1] + attdef.dxf.insert[1] * insert.dxf.yscale,
insert.dxf.insert[2] + (attdef.dxf.insert[2] if len(attdef.dxf.insert) > 2 else 0.0)
)
attr_info = AttributeInfo(
tag=attdef.dxf.tag,
text=text_value,
position=actual_position,
height=getattr(attdef.dxf, 'height', 12.0),
rotation=getattr(attdef.dxf, 'rotation', 0.0) + insert.dxf.rotation,
layer=getattr(attdef.dxf, 'layer', insert.dxf.layer),
prompt=getattr(attdef.dxf, 'prompt', None),
style=getattr(attdef.dxf, 'style', None),
invisible=getattr(attdef, 'is_invisible', False),
const=getattr(attdef, 'is_const', False),
entity_handle=attdef.dxf.handle,
insert_x=actual_position[0],
insert_y=actual_position[1],
insert_z=actual_position[2]
)
return attr_info
except Exception as e:
self.logger.warning(f"ATTDEF 정보 추출 중 오류: {e}")
return None
def _extract_text_as_attribute(self, text_entity, insert: Insert) -> Optional[AttributeInfo]:
"""TEXT/MTEXT 엔티티를 속성으로 추출"""
try:
# 텍스트 내용 추출
if hasattr(text_entity, 'text'):
text_value = text_entity.text.strip()
elif hasattr(text_entity.dxf, 'text'):
text_value = text_entity.dxf.text.strip()
else:
return None
if not text_value:
return None
# INSERT의 위치를 기준으로 실제 위치 계산
text_pos = text_entity.dxf.insert
actual_position = (
insert.dxf.insert[0] + text_pos[0] * insert.dxf.xscale,
insert.dxf.insert[1] + text_pos[1] * insert.dxf.yscale,
insert.dxf.insert[2] + (text_pos[2] if len(text_pos) > 2 else 0.0)
)
# 태그는 텍스트 내용의 첫 단어나 전체 내용으로 설정
tag = text_value.split()[0] if ' ' in text_value else text_value[:20]
attr_info = AttributeInfo(
tag=f"TEXT_{tag}",
text=text_value,
position=actual_position,
height=getattr(text_entity.dxf, 'height', 12.0),
rotation=getattr(text_entity.dxf, 'rotation', 0.0) + insert.dxf.rotation,
layer=getattr(text_entity.dxf, 'layer', insert.dxf.layer),
style=getattr(text_entity.dxf, 'style', None),
entity_handle=text_entity.dxf.handle,
insert_x=actual_position[0],
insert_y=actual_position[1],
insert_z=actual_position[2]
)
return attr_info
except Exception as e:
self.logger.warning(f"TEXT 엔티티 추출 중 오류: {e}")
return None
def identify_title_block(self, block_refs: List[BlockInfo]) -> Optional[TitleBlockInfo]:
"""도곽 블록 식별 및 정보 추출"""
if not block_refs:
return None
title_block_candidates = []
# 각 블록 참조에 대해 도곽 가능성 점수 계산
for block_ref in block_refs:
score = self._calculate_title_block_score(block_ref)
if score > 0:
title_block_candidates.append((block_ref, score))
if not title_block_candidates:
self.logger.warning("도곽 블록을 찾을 수 없습니다.")
return None
# 가장 높은 점수의 블록을 도곽으로 선택
title_block_candidates.sort(key=lambda x: x[1], reverse=True)
best_candidate = title_block_candidates[0][0]
self.logger.info(f"도곽 블록 식별: '{best_candidate.name}' (점수: {title_block_candidates[0][1]})")
# TitleBlockInfo 생성
title_block = TitleBlockInfo(
block_name=best_candidate.name,
all_attributes=best_candidate.attributes
)
# 속성들을 분류하여 해당 필드에 할당
for attr in best_candidate.attributes:
self._assign_attribute_to_field(title_block, attr)
title_block.attributes_count = len(best_candidate.attributes)
return title_block
def _calculate_title_block_score(self, block_ref: BlockInfo) -> int:
"""블록이 도곽일 가능성 점수 계산"""
score = 0
# 블록 이름에 도곽 관련 키워드가 있는지 확인
name_lower = block_ref.name.lower()
title_keywords = ['title', 'titleblock', 'title_block', '도곽', '타이틀', 'border', 'frame']
for keyword in title_keywords:
if keyword in name_lower:
score += 10
break
# 속성 개수 (도곽은 보통 많은 속성을 가짐)
if len(block_ref.attributes) >= 5:
score += 5
elif len(block_ref.attributes) >= 3:
score += 3
elif len(block_ref.attributes) >= 1:
score += 1
# 도곽 관련 속성이 있는지 확인
for attr in block_ref.attributes:
for field_name, keywords in self.title_block_keywords.items():
if self._contains_any_keyword(attr.tag.lower(), keywords) or \
self._contains_any_keyword(attr.text.lower(), keywords):
score += 2
return score
def _contains_any_keyword(self, text: str, keywords: List[str]) -> bool:
"""텍스트에 키워드 중 하나라도 포함되어 있는지 확인"""
text_lower = text.lower()
return any(keyword.lower() in text_lower for keyword in keywords)
def _assign_attribute_to_field(self, title_block: TitleBlockInfo, attr: AttributeInfo):
"""속성을 도곽 정보의 해당 필드에 할당"""
attr_text = attr.text.strip()
if not attr_text:
return
# 태그나 텍스트에서 키워드 매칭
attr_tag_lower = attr.tag.lower()
attr_text_lower = attr_text.lower()
# 각 필드별로 키워드 매칭
for field_name, keywords in self.title_block_keywords.items():
if self._contains_any_keyword(attr_tag_lower, keywords) or \
self._contains_any_keyword(attr_text_lower, keywords):
if field_name == '도면명' and not title_block.drawing_name:
title_block.drawing_name = attr_text
elif field_name == '도면번호' and not title_block.drawing_number:
title_block.drawing_number = attr_text
elif field_name == '건설분야' and not title_block.construction_field:
title_block.construction_field = attr_text
elif field_name == '건설단계' and not title_block.construction_stage:
title_block.construction_stage = attr_text
elif field_name == '축척' and not title_block.scale:
title_block.scale = attr_text
elif field_name == '설계자' and not title_block.designer:
title_block.designer = attr_text
elif field_name == '프로젝트' and not title_block.project_name:
title_block.project_name = attr_text
elif field_name == '날짜' and not title_block.date:
title_block.date = attr_text
elif field_name == '리비전' and not title_block.revision:
title_block.revision = attr_text
elif field_name == '위치' and not title_block.location:
title_block.location = attr_text
break
def process_dxf_file_comprehensive(self, file_path: str) -> Dict[str, Any]:
"""DXF 파일 종합적인 처리 - 수정된 버전"""
result = {
'success': False,
'error': None,
'file_path': file_path,
'title_block': None,
'block_references': [],
'summary': {}
}
try:
# 파일 유효성 검사
if not self.validate_dxf_file(file_path):
result['error'] = "유효하지 않은 DXF 파일입니다."
return result
# DXF 문서 로드
doc = self.load_dxf_document(file_path)
if not doc:
result['error'] = "DXF 문서를 로드할 수 없습니다."
return result
# 모든 INSERT 엔티티에서 속성 추출
self.logger.info("INSERT 엔티티 속성 추출 중...")
block_references = self.extract_all_insert_attributes(doc)
# 도곽 정보 추출
self.logger.info("도곽 정보 추출 중...")
title_block = self.identify_title_block(block_references)
# 요약 정보 생성
total_attributes = sum(len(br.attributes) for br in block_references)
non_empty_attributes = sum(len([attr for attr in br.attributes if attr.text.strip()])
for br in block_references)
summary = {
'total_blocks': len(block_references),
'title_block_found': title_block is not None,
'title_block_name': title_block.block_name if title_block else None,
'attributes_count': title_block.attributes_count if title_block else 0,
'total_attributes': total_attributes,
'non_empty_attributes': non_empty_attributes
}
# 결과 설정
result['title_block'] = asdict(title_block) if title_block else None
result['block_references'] = [asdict(br) for br in block_references]
result['summary'] = summary
result['success'] = True
self.logger.info(f"DXF 파일 처리 완료: {file_path}")
self.logger.info(f"처리 요약: 블록 {len(block_references)}개, 속성 {total_attributes}개 (비어있지 않은 속성: {non_empty_attributes}개)")
except Exception as e:
self.logger.error(f"DXF 파일 처리 중 오류: {e}")
result['error'] = str(e)
return result
# 기존 클래스명과의 호환성을 위한 별칭
DXFProcessor = FixedDXFProcessor
def main():
"""테스트용 메인 함수"""
logging.basicConfig(level=logging.DEBUG)
if not EZDXF_AVAILABLE:
print("ezdxf 라이브러리가 설치되지 않았습니다.")
return
processor = FixedDXFProcessor()
# 업로드 폴더에서 DXF 파일 찾기
upload_dir = "uploads"
if os.path.exists(upload_dir):
for file in os.listdir(upload_dir):
if file.lower().endswith('.dxf'):
test_file = os.path.join(upload_dir, file)
print(f"\n테스트 파일: {test_file}")
result = processor.process_dxf_file_comprehensive(test_file)
if result['success']:
print("✅ DXF 파일 처리 성공!")
summary = result['summary']
print(f" 블록 수: {summary['total_blocks']}")
print(f" 도곽 발견: {summary['title_block_found']}")
print(f" 도곽 속성 수: {summary['attributes_count']}")
print(f" 전체 속성 수: {summary['total_attributes']}")
print(f" 비어있지 않은 속성 수: {summary['non_empty_attributes']}")
if summary['title_block_name']:
print(f" 도곽 블록명: {summary['title_block_name']}")
else:
print(f"❌ 처리 실패: {result['error']}")
break
else:
print("uploads 폴더를 찾을 수 없습니다.")
if __name__ == "__main__":
main()