Files
TK-BOM-Project/backend/app/services/instrument_classifier.py
Hyungi Ahn 3dd301cb57 볼트 분류 개선 및 업로드 성능 최적화
- 볼트 길이 추출 로직 개선: '70.0000 LG' 형태 인식 추가
- 재질 중복 표시 수정: 'ASTM A193 ASTM A193 B7' → 'B7'
- A193/A194 등급 추출 로직 개선: 'GR B7/2H' 형태 지원
- bolt_details 테이블에 pressure_rating 컬럼 추가
- 볼트 분류기 오분류 방지: 플랜지/피팅이 볼트로 분류되지 않도록 수정
- 업로드 성능 개선: 키워드 기반 빠른 분류기 선택 로직 추가
- 분류 키워드 대폭 확장: 피팅/파이프/플랜지 키워드 추가
2025-07-18 12:48:24 +09:00

234 lines
8.0 KiB
Python

"""
INSTRUMENT 분류 시스템 (간단 버전)
완제품 구매용 기본 분류만
"""
import re
from typing import Dict, List, Optional
from .material_classifier import classify_material
# ========== 기본 계기 타입 ==========
INSTRUMENT_TYPES = {
"PRESSURE_GAUGE": {
"dat_file_patterns": ["PG_", "PRESS_G"],
"description_keywords": ["PRESSURE GAUGE", "압력계", "PG"],
"characteristics": "압력 측정용 게이지"
},
"TEMPERATURE_GAUGE": {
"dat_file_patterns": ["TG_", "TEMP_G"],
"description_keywords": ["TEMPERATURE GAUGE", "온도계", "TG", "THERMOMETER"],
"characteristics": "온도 측정용 게이지"
},
"FLOW_METER": {
"dat_file_patterns": ["FM_", "FLOW_"],
"description_keywords": ["FLOW METER", "유량계", "FM"],
"characteristics": "유량 측정용"
},
"LEVEL_GAUGE": {
"dat_file_patterns": ["LG_", "LEVEL_"],
"description_keywords": ["LEVEL GAUGE", "액위계", "LG", "SIGHT GLASS"],
"characteristics": "액위 측정용"
},
"TRANSMITTER": {
"dat_file_patterns": ["PT_", "TT_", "FT_", "LT_"],
"description_keywords": ["TRANSMITTER", "트랜스미터", "4-20MA"],
"characteristics": "신호 전송용"
},
"INDICATOR": {
"dat_file_patterns": ["PI_", "TI_", "FI_", "LI_"],
"description_keywords": ["INDICATOR", "지시계", "DISPLAY"],
"characteristics": "표시용"
},
"SPECIAL_INSTRUMENT": {
"dat_file_patterns": ["INST_", "SPEC_"],
"description_keywords": ["THERMOWELL", "ORIFICE PLATE", "MANOMETER", "ROTAMETER"],
"characteristics": "특수 계기류"
}
}
def classify_instrument(dat_file: str, description: str, main_nom: str, length: Optional[float] = None) -> Dict:
"""
간단한 INSTRUMENT 분류
Args:
dat_file: DAT_FILE 필드
description: DESCRIPTION 필드
main_nom: MAIN_NOM 필드 (연결 사이즈)
Returns:
간단한 계기 분류 결과
"""
# 1. 먼저 계기인지 확인 (계기 키워드가 있어야 함)
desc_upper = description.upper()
dat_upper = dat_file.upper()
combined_text = f"{dat_upper} {desc_upper}"
# 계기 관련 키워드 확인
instrument_keywords = [
"GAUGE", "METER", "TRANSMITTER", "INDICATOR", "SENSOR",
"THERMOMETER", "MANOMETER", "ROTAMETER", "THERMOWELL",
"ORIFICE PLATE", "DISPLAY", "4-20MA", "4-20 MA",
"압력계", "온도계", "유량계", "액위계", "게이지", "계기",
"트랜스미터", "지시계", "센서"
]
# 계기가 아닌 것들의 키워드
non_instrument_keywords = [
"BOLT", "SCREW", "STUD", "NUT", "WASHER", "볼트", "나사", "너트", "와셔",
"PIPE", "TUBE", "파이프", "배관",
"ELBOW", "TEE", "REDUCER", "CAP", "엘보", "", "리듀서",
"VALVE", "GATE", "BALL", "GLOBE", "CHECK", "밸브",
"FLANGE", "FLG", "플랜지",
"GASKET", "GASK", "가스켓"
]
# 계기가 아닌 키워드가 있으면 거부
if any(keyword in combined_text for keyword in non_instrument_keywords):
return {
"category": "UNKNOWN",
"overall_confidence": 0.1,
"reason": "NON_INSTRUMENT_KEYWORDS_DETECTED"
}
# 계기 키워드가 없으면 거부
has_instrument_keyword = any(keyword in combined_text for keyword in instrument_keywords)
if not has_instrument_keyword:
return {
"category": "UNKNOWN",
"overall_confidence": 0.1,
"reason": "NO_INSTRUMENT_KEYWORDS_FOUND"
}
# 2. 재질 분류 (공통 모듈)
material_result = classify_material(description)
# 3. 계기 타입 분류
instrument_type_result = classify_instrument_type(dat_file, description)
# 4. 측정 범위 추출 (있다면)
measurement_range = extract_measurement_range(description)
# 5. 전체 신뢰도 계산
base_confidence = 0.8 if has_instrument_keyword else 0.1
instrument_confidence = instrument_type_result.get('confidence', 0.0)
overall_confidence = (base_confidence + instrument_confidence) / 2
# 6. 최종 결과
return {
"category": "INSTRUMENT",
# 재질 정보
"material": {
"standard": material_result.get('standard', 'UNKNOWN'),
"grade": material_result.get('grade', 'UNKNOWN'),
"material_type": material_result.get('material_type', 'UNKNOWN'),
"confidence": material_result.get('confidence', 0.0)
},
# 계기 정보
"instrument_type": {
"type": instrument_type_result.get('type', 'UNKNOWN'),
"characteristics": instrument_type_result.get('characteristics', ''),
"confidence": instrument_type_result.get('confidence', 0.0)
},
"measurement_info": {
"range": measurement_range.get('range', ''),
"unit": measurement_range.get('unit', ''),
"signal_type": measurement_range.get('signal_type', '')
},
"size_info": {
"connection_size": main_nom,
"size_description": main_nom
},
"purchase_info": {
"category": "완제품 구매",
"supplier_type": "계기 전문업체",
"lead_time": "2-4주",
"note": "사양서 확인 후 주문"
},
# 전체 신뢰도
"overall_confidence": overall_confidence
}
def classify_instrument_type(dat_file: str, description: str) -> Dict:
"""계기 타입 분류"""
dat_upper = dat_file.upper()
desc_upper = description.upper()
# DAT_FILE 패턴 확인
for inst_type, type_data in INSTRUMENT_TYPES.items():
for pattern in type_data["dat_file_patterns"]:
if pattern in dat_upper:
return {
"type": inst_type,
"characteristics": type_data["characteristics"],
"confidence": 0.9,
"evidence": [f"DAT_PATTERN: {pattern}"]
}
# DESCRIPTION 키워드 확인
for inst_type, type_data in INSTRUMENT_TYPES.items():
for keyword in type_data["description_keywords"]:
if keyword in desc_upper:
return {
"type": inst_type,
"characteristics": type_data["characteristics"],
"confidence": 0.8,
"evidence": [f"KEYWORD: {keyword}"]
}
return {
"type": "UNKNOWN",
"characteristics": "분류되지 않은 계기",
"confidence": 0.0,
"evidence": ["NO_INSTRUMENT_TYPE_FOUND"]
}
def extract_measurement_range(description: str) -> Dict:
"""측정 범위 추출 (간단히)"""
desc_upper = description.upper()
# 압력 범위
pressure_match = re.search(r'(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)\s*(PSI|BAR|KPA)', desc_upper)
if pressure_match:
return {
"range": f"{pressure_match.group(1)}-{pressure_match.group(2)}",
"unit": pressure_match.group(3),
"signal_type": "PRESSURE"
}
# 온도 범위
temp_match = re.search(r'(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)\s*(°?C|°?F)', desc_upper)
if temp_match:
return {
"range": f"{temp_match.group(1)}-{temp_match.group(2)}",
"unit": temp_match.group(3),
"signal_type": "TEMPERATURE"
}
# 신호 타입
if "4-20MA" in desc_upper or "4-20 MA" in desc_upper:
return {
"range": "4-20mA",
"unit": "mA",
"signal_type": "ANALOG"
}
return {
"range": "",
"unit": "",
"signal_type": ""
}
def calculate_simple_confidence(scores: List[float]) -> float:
"""간단한 신뢰도 계산"""
valid_scores = [s for s in scores if s > 0]
return round(sum(valid_scores) / len(valid_scores), 2) if valid_scores else 0.0