HC900-Crawler/mcp-server/pipeline/legend_parser.py

#!/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()