Files
ExperionCrawler/mcp-server/pipeline/legend_parser.py
2026-05-08 17:22:10 +09:00

248 lines
7.8 KiB
Python

#!/usr/bin/env python3
"""
P&ID 도면의 레전드 페이지를 파싱하여 계측기 식별 패턴을 추출합니다.
레전드 페이지는 일반적으로 DXF 파일의 첫 번째 도면에 위치하며,
계측기 유형별 패턴(예: FICQ = Flow Indicating Control with Differential)을 정의합니다.
이 모듈은:
1. 레전드 페이지에서 계측기 식별 테이블 추출
2. 계측기 유형별 패턴 정의
3. 실제 도면에서 태그 필터링
"""
import re
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class InstrumentPattern:
"""계측기 패턴 정의"""
first_letter: str # 첫 글자 (기능)
first_meaning: str # 첫 글자 의미
succeeding_letters: Dict[str, str] # 후속 글자: {글자: 의미}
def matches(self, tag: str) -> bool:
"""태그가 이 패턴과 일치하는지 확인"""
if not tag:
return False
# 첫 글자 확인
if tag[0].upper() != self.first_letter.upper():
return False
return True
class LegendParser:
"""
P&ID 도면의 레전드 페이지를 파싱하여 계측기 식별 패턴을 추출합니다.
"""
# 계측기 식별 테이블 (ISA-5.1 표준 기반)
INSTRUMENT_IDENTIFICATION = {
# 첫 글자 (기능)
'A': 'ANALYSIS, AUTO',
'B': 'BURNER COMBUSTION',
'C': 'CONDUCTIVITY',
'D': 'DENSITY',
'E': 'VOLTAGE(EMF), HEAT ENERGY',
'F': 'FLOW RATE',
'G': 'GAUGE, GAS',
'H': 'HAND, HAND-CONTROLLED SHUT-OFF VALVE',
'I': 'CURRENT(ELECTRICAL)',
'J': 'POWER(MW,MVAR)',
'K': 'TIME',
'L': 'LEVEL',
'M': 'MAN, MOTOR',
'N': 'NUMBER OF OBJECTS',
'P': 'PRESSURE',
'Q': 'WEIGHT',
'R': 'RADIATION, RADIOACTIVITY',
'S': 'SPEED, FREQUENCY',
'T': 'TEMPERATURE',
'U': 'MULTIVARIABLE',
'V': 'VALVE, DAMPER',
'W': 'ACTUATOR',
'X': 'UNSPECIFIED',
'Y': 'RELAY, EVENT',
'Z': 'POSITION, LINEAR',
# 후속 글자 (수식어)
'AA': 'ALARM, AUDIBLE',
'AB': 'ABSORBENCY',
'AC': 'ACCELERATION',
'AD': 'ADHESION',
'AE': 'AREA',
'AF': 'AVG FLOW RATE',
'AG': 'GRAVIMETRIC',
'AH': 'HUMIDITY',
'AI': 'IONIZATION',
'AJ': 'JOULE',
'AK': 'KINETIC',
'AL': 'LEVEL',
'AM': 'MAGNETIC',
'AN': 'NORMAL',
'AO': 'OPTICAL',
'AP': 'PHASE',
'AQ': 'QUANTITY',
'AR': 'RATIO',
'AS': 'SPEED',
'AT': 'TEMPERATURE',
'AU': 'UNSPECIFIED',
'AV': 'VOLTAGE',
'AW': 'WAVELENGTH',
'AX': 'X-RAY',
'AY': 'YIELD',
'AZ': 'Z-POTENTIAL',
}
# 계측기 유형별 그룹
INSTRUMENT_GROUPS = {
'Sensor': ['FT', 'FIT', 'LT', 'PT', 'TE', 'PG', 'LG', 'TG', 'FI', 'PI', 'TI', 'LI'],
'Valve': ['FCV', 'TCV', 'LCV', 'PCV', 'XV', 'FV', 'LV', 'PV', 'TV'],
'Controller': ['FIC', 'TIC', 'PIC', 'LIC', 'FICQ', 'TICA', 'PICA', 'LICA'],
'Indicator': ['FI', 'PI', 'TI', 'LI', 'FIT', 'LIT', 'PIT', 'TIT'],
'Transmitter': ['FT', 'LT', 'PT', 'TT', 'FIT', 'LIT', 'PIT', 'TIT'],
'Gauge': ['PG', 'TG', 'LG'],
'Safety': ['PSV', 'SRV', 'SDV'],
'Pump': ['P-', 'P-1', 'P-2'],
'Compressor': ['C-', 'K-'],
'Heat Exchanger': ['E-'],
'Tank': ['T-', 'D-'],
'Column': ['C-'],
'Filter': ['F-'],
'Separator': ['SP-'],
}
# 배관 번호 패턴
PIPELINE_PATTERN = re.compile(r'[A-Z]{1,6}-\d{3,6}(-[A-Z0-9]+)*')
# 계측기 태그 패턴
INSTRUMENT_TAG_PATTERN = re.compile(r'[A-Z]{1,6}-\d{3,6}(-[A-Z0-9]+)*')
def __init__(self):
self.patterns: List[InstrumentPattern] = []
self._build_patterns()
def _build_patterns(self):
"""계측기 식별 패턴을 빌드합니다."""
for letter, meaning in self.INSTRUMENT_IDENTIFICATION.items():
if len(letter) == 1: # 첫 글자만
self.patterns.append(InstrumentPattern(
first_letter=letter,
first_meaning=meaning,
succeeding_letters={}
))
def extract_instrument_tags(self, text: str) -> List[str]:
"""
텍스트에서 계측기 태그를 추출합니다.
Args:
text: DXF에서 추출한 텍스트
Returns:
계측기 태그 목록
"""
tags = []
for match in self.INSTRUMENT_TAG_PATTERN.finditer(text):
tag = match.group(0)
# 계측기 태그인지 확인
if self._is_instrument_tag(tag):
tags.append(tag)
return tags
def _is_instrument_tag(self, tag: str) -> bool:
"""태그가 계측기 태그인지 확인합니다."""
if not tag or len(tag) < 3:
return False
# 첫 글자가 계측기 식별 테이블에 있는지 확인
first_letter = tag[0].upper()
if first_letter not in self.INSTRUMENT_IDENTIFICATION:
return False
# 숫자 부분이 있는지 확인
if not re.search(r'\d', tag):
return False
return True
def group_by_type(self, tags: List[str]) -> Dict[str, List[str]]:
"""
태그를 유형별로 그룹핑합니다.
Args:
tags: 계측기 태그 목록
Returns:
유형별 태그 그룹 {유형: [태그1, 태그2, ...]}
"""
groups = {group: [] for group in self.INSTRUMENT_GROUPS}
for tag in tags:
for group, patterns in self.INSTRUMENT_GROUPS.items():
for pattern in patterns:
if tag.startswith(pattern):
groups[group].append(tag)
break
# 빈 그룹 제거
return {k: v for k, v in groups.items() if v}
def extract_area_from_tag(self, tag: str) -> Optional[str]:
"""
태그에서 AREA 번호를 추출합니다.
예: FICQ-6113 → "6" (6호 플랜트)
FICQ-10113 → "10" (10호 플랜트)
Args:
tag: 계측기 태그
Returns:
AREA 번호 또는 None
"""
match = re.search(r'-([1-9]\d*)\d+', tag)
if match:
return match.group(1)
return None
def parse_legend_page(self, dxf_filepath: str) -> Dict:
"""
레전드 페이지를 파싱하여 계측기 식별 패턴을 추출합니다.
Args:
dxf_filepath: DXF 파일 경로
Returns:
레전드 페이지 정보
"""
import ezdxf
from ezdxf.tools.text import plain_mtext
doc = ezdxf.readfile(dxf_filepath)
msp = doc.modelspace()
# 레전드 페이지 영역 (X: -176 ~ 2000)
legend_texts = []
for entity in msp:
if entity.dxftype() in ('TEXT', 'MTEXT'):
x = entity.dxf.insert.x
if -176 <= x <= 2000:
if entity.dxftype() == 'TEXT':
text = entity.dxf.text
else:
text = plain_mtext(entity.dxf.text) if hasattr(entity.dxf, 'text') else entity.text
if text.strip():
legend_texts.append((x, entity.dxf.insert.y, text))
return {
'legend_texts': legend_texts,
'instrument_identification': self.INSTRUMENT_IDENTIFICATION,
'instrument_groups': self.INSTRUMENT_GROUPS,
}
# 단일 인스턴스 (싱글톤)
legend_parser = LegendParser()