""" VALVE 분류 시스템 주조 밸브 + 단조 밸브 구분 포함 """ import re from typing import Dict, List, Optional from .material_classifier import classify_material, get_manufacturing_method_from_material # ========== 밸브 타입별 분류 ========== VALVE_TYPES = { "GATE_VALVE": { "dat_file_patterns": ["GATE_", "GV_"], "description_keywords": ["GATE VALVE", "GATE", "게이트"], "characteristics": "완전 개폐용, 직선 유로", "typical_connections": ["FLANGED", "SOCKET_WELD", "BUTT_WELD"], "pressure_range": "150LB ~ 2500LB", "special_features": ["OS&Y", "RS", "NRS"] }, "BALL_VALVE": { "dat_file_patterns": ["BALL_", "BV_"], "description_keywords": ["BALL VALVE", "BALL", "볼밸브"], "characteristics": "빠른 개폐, 낮은 압력손실", "typical_connections": ["FLANGED", "THREADED", "SOCKET_WELD"], "pressure_range": "150LB ~ 6000LB", "special_features": ["FULL_PORT", "REDUCED_PORT", "3_WAY", "4_WAY"] }, "GLOBE_VALVE": { "dat_file_patterns": ["GLOBE_", "GLV_"], "description_keywords": ["GLOBE VALVE", "GLOBE", "글로브"], "characteristics": "유량 조절용, 정밀 제어", "typical_connections": ["FLANGED", "SOCKET_WELD", "BUTT_WELD"], "pressure_range": "150LB ~ 2500LB", "special_features": ["ANGLE_TYPE", "Y_TYPE"] }, "CHECK_VALVE": { "dat_file_patterns": ["CHK_", "CHECK_", "CV_"], "description_keywords": ["CHECK VALVE", "CHECK", "체크", "역지"], "characteristics": "역류 방지용", "typical_connections": ["FLANGED", "SOCKET_WELD", "WAFER"], "pressure_range": "150LB ~ 2500LB", "special_features": ["SWING_TYPE", "LIFT_TYPE", "DUAL_PLATE", "PISTON_TYPE"] }, "BUTTERFLY_VALVE": { "dat_file_patterns": ["BUTTERFLY_", "BFV_"], "description_keywords": ["BUTTERFLY VALVE", "BUTTERFLY", "버터플라이"], "characteristics": "대구경용, 경량", "typical_connections": ["WAFER", "LUG", "FLANGED"], "pressure_range": "150LB ~ 600LB", "special_features": ["GEAR_OPERATED", "LEVER_OPERATED"] }, "NEEDLE_VALVE": { "dat_file_patterns": ["NEEDLE_", "NV_"], "description_keywords": ["NEEDLE VALVE", "NEEDLE", "니들"], "characteristics": "정밀 유량 조절용", "typical_connections": ["THREADED", "SOCKET_WELD"], "pressure_range": "800LB ~ 6000LB", "special_features": ["FINE_ADJUSTMENT"], "typically_forged": True }, "RELIEF_VALVE": { "dat_file_patterns": ["RELIEF_", "RV_", "PSV_"], "description_keywords": ["RELIEF VALVE", "PRESSURE RELIEF", "SAFETY VALVE", "PSV", "압력 안전", "릴리프"], "characteristics": "안전 압력 방출용", "typical_connections": ["FLANGED", "THREADED"], "pressure_range": "150LB ~ 2500LB", "special_features": ["SET_PRESSURE", "PILOT_OPERATED"] }, "SOLENOID_VALVE": { "dat_file_patterns": ["SOLENOID_", "SOL_"], "description_keywords": ["SOLENOID VALVE", "SOLENOID", "솔레노이드"], "characteristics": "전기 제어용", "typical_connections": ["THREADED"], "pressure_range": "150LB ~ 600LB", "special_features": ["2_WAY", "3_WAY", "NC", "NO"] }, "PLUG_VALVE": { "dat_file_patterns": ["PLUG_", "PV_"], "description_keywords": ["PLUG VALVE", "PLUG", "플러그"], "characteristics": "다방향 제어용", "typical_connections": ["FLANGED", "THREADED"], "pressure_range": "150LB ~ 600LB", "special_features": ["LUBRICATED", "NON_LUBRICATED"] } } # ========== 연결 방식별 분류 ========== VALVE_CONNECTIONS = { "FLANGED": { "codes": ["FL", "FLANGED", "플랜지"], "dat_patterns": ["_FL_"], "size_range": "1\" ~ 48\"", "pressure_range": "150LB ~ 2500LB", "manufacturing": "CAST", "confidence": 0.95 }, "THREADED": { "codes": ["THD", "THRD", "NPT", "THREADED", "나사"], "dat_patterns": ["_THD_", "_TR_"], "size_range": "1/4\" ~ 4\"", "pressure_range": "150LB ~ 6000LB", "manufacturing": "FORGED_OR_CAST", "confidence": 0.95 }, "SOCKET_WELD": { "codes": ["SW", "SOCKET WELD", "소켓웰드"], "dat_patterns": ["_SW_"], "size_range": "1/4\" ~ 4\"", "pressure_range": "800LB ~ 9000LB", "manufacturing": "FORGED", "confidence": 0.95 }, "BUTT_WELD": { "codes": ["BW", "BUTT WELD", "맞대기용접"], "dat_patterns": ["_BW_"], "size_range": "1/2\" ~ 48\"", "pressure_range": "150LB ~ 2500LB", "manufacturing": "CAST_OR_FORGED", "confidence": 0.95 }, "WAFER": { "codes": ["WAFER", "WAFER TYPE", "웨이퍼"], "dat_patterns": ["_WAF_"], "size_range": "2\" ~ 48\"", "pressure_range": "150LB ~ 600LB", "manufacturing": "CAST", "confidence": 0.9 }, "LUG": { "codes": ["LUG", "LUG TYPE", "러그"], "dat_patterns": ["_LUG_"], "size_range": "2\" ~ 48\"", "pressure_range": "150LB ~ 600LB", "manufacturing": "CAST", "confidence": 0.9 } } # ========== 작동 방식별 분류 ========== VALVE_ACTUATION = { "MANUAL": { "keywords": ["MANUAL", "HAND WHEEL", "LEVER", "수동"], "characteristics": "수동 조작", "applications": "일반 개폐용" }, "GEAR_OPERATED": { "keywords": ["GEAR OPERATED", "GEAR", "기어"], "characteristics": "기어 구동", "applications": "대구경 수동 조작" }, "PNEUMATIC": { "keywords": ["PNEUMATIC", "AIR OPERATED", "공압"], "characteristics": "공압 구동", "applications": "자동 제어" }, "ELECTRIC": { "keywords": ["ELECTRIC", "MOTOR OPERATED", "전동"], "characteristics": "전동 구동", "applications": "원격 제어" }, "SOLENOID": { "keywords": ["SOLENOID", "솔레노이드"], "characteristics": "전자 밸브", "applications": "빠른 전기 제어" } } # ========== 압력 등급별 분류 ========== VALVE_PRESSURE_RATINGS = { "patterns": [ r"(\d+)LB", r"CLASS\s*(\d+)", r"CL\s*(\d+)", r"(\d+)#", r"(\d+)\s*WOG" # Water Oil Gas ], "standard_ratings": { "150LB": {"max_pressure": "285 PSI", "typical_manufacturing": "CAST"}, "300LB": {"max_pressure": "740 PSI", "typical_manufacturing": "CAST"}, "600LB": {"max_pressure": "1480 PSI", "typical_manufacturing": "CAST_OR_FORGED"}, "800LB": {"max_pressure": "2000 PSI", "typical_manufacturing": "FORGED"}, "900LB": {"max_pressure": "2220 PSI", "typical_manufacturing": "CAST_OR_FORGED"}, "1500LB": {"max_pressure": "3705 PSI", "typical_manufacturing": "FORGED"}, "2500LB": {"max_pressure": "6170 PSI", "typical_manufacturing": "FORGED"}, "3000LB": {"max_pressure": "7400 PSI", "typical_manufacturing": "FORGED"}, "6000LB": {"max_pressure": "14800 PSI", "typical_manufacturing": "FORGED"}, "9000LB": {"max_pressure": "22200 PSI", "typical_manufacturing": "FORGED"} } } def classify_valve(dat_file: str, description: str, main_nom: str, length: float = None) -> Dict: """ 완전한 VALVE 분류 Args: dat_file: DAT_FILE 필드 description: DESCRIPTION 필드 main_nom: MAIN_NOM 필드 (사이즈) Returns: 완전한 밸브 분류 결과 """ desc_upper = description.upper() dat_upper = dat_file.upper() # 1. 밸브 키워드 확인 (재질만 있어도 통합 분류기가 이미 밸브로 분류했으므로 진행) valve_keywords = ['VALVE', 'GATE', 'BALL', 'GLOBE', 'CHECK', 'BUTTERFLY', 'NEEDLE', 'RELIEF', 'SOLENOID', '밸브', '게이트', '볼', '글로브', '체크', '버터플라이', '니들', '릴리프', '솔레노이드'] has_valve_keyword = any(keyword in desc_upper or keyword in dat_upper for keyword in valve_keywords) # 밸브 재질 확인 (A216, A217, A351, A352) valve_materials = ['A216', 'A217', 'A351', 'A352'] has_valve_material = any(material in desc_upper for material in valve_materials) # 밸브 키워드도 없고 밸브 재질도 없으면 UNKNOWN if not has_valve_keyword and not has_valve_material: return { "category": "UNKNOWN", "overall_confidence": 0.0, "reason": "밸브 키워드 및 재질 없음" } # 2. 재질 분류 (공통 모듈 사용) material_result = classify_material(description) # 2. 밸브 타입 분류 valve_type_result = classify_valve_type(dat_file, description) # 3. 연결 방식 분류 connection_result = classify_valve_connection(dat_file, description) # 4. 압력 등급 분류 pressure_result = classify_valve_pressure_rating(dat_file, description) # 5. 작동 방식 분류 actuation_result = classify_valve_actuation(description) # 6. 제작 방법 결정 (주조 vs 단조) manufacturing_result = determine_valve_manufacturing( material_result, valve_type_result, connection_result, pressure_result, main_nom ) # 7. 특수 기능 추출 special_features = extract_valve_special_features(description, valve_type_result) # 8. 최종 결과 조합 return { "category": "VALVE", # 재질 정보 (공통 모듈) "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) }, # 밸브 분류 정보 "valve_type": { "type": valve_type_result.get('type', 'UNKNOWN'), "characteristics": valve_type_result.get('characteristics', ''), "confidence": valve_type_result.get('confidence', 0.0), "evidence": valve_type_result.get('evidence', []) }, "connection_method": { "method": connection_result.get('method', 'UNKNOWN'), "confidence": connection_result.get('confidence', 0.0), "size_range": connection_result.get('size_range', ''), "pressure_range": connection_result.get('pressure_range', '') }, "pressure_rating": { "rating": pressure_result.get('rating', 'UNKNOWN'), "confidence": pressure_result.get('confidence', 0.0), "max_pressure": pressure_result.get('max_pressure', ''), "typical_manufacturing": pressure_result.get('typical_manufacturing', '') }, "actuation": { "method": actuation_result.get('method', 'MANUAL'), "characteristics": actuation_result.get('characteristics', ''), "confidence": actuation_result.get('confidence', 0.6) }, "manufacturing": { "method": manufacturing_result.get('method', 'UNKNOWN'), "confidence": manufacturing_result.get('confidence', 0.0), "evidence": manufacturing_result.get('evidence', []), "characteristics": manufacturing_result.get('characteristics', '') }, "special_features": special_features, "size_info": { "valve_size": main_nom, "size_description": main_nom }, # 전체 신뢰도 "overall_confidence": calculate_valve_confidence({ "material": material_result.get('confidence', 0), "valve_type": valve_type_result.get('confidence', 0), "connection": connection_result.get('confidence', 0), "pressure": pressure_result.get('confidence', 0) }) } def classify_valve_type(dat_file: str, description: str) -> Dict: """밸브 타입 분류""" dat_upper = dat_file.upper() desc_upper = description.upper() # 1. DAT_FILE 패턴으로 1차 분류 (가장 신뢰도 높음) for valve_type, type_data in VALVE_TYPES.items(): for pattern in type_data["dat_file_patterns"]: if pattern in dat_upper: return { "type": valve_type, "characteristics": type_data["characteristics"], "confidence": 0.95, "evidence": [f"DAT_FILE_PATTERN: {pattern}"], "typical_connections": type_data["typical_connections"], "special_features": type_data.get("special_features", []) } # 2. DESCRIPTION 키워드로 2차 분류 for valve_type, type_data in VALVE_TYPES.items(): for keyword in type_data["description_keywords"]: if keyword in desc_upper: return { "type": valve_type, "characteristics": type_data["characteristics"], "confidence": 0.85, "evidence": [f"DESCRIPTION_KEYWORD: {keyword}"], "typical_connections": type_data["typical_connections"], "special_features": type_data.get("special_features", []) } # 3. 분류 실패 return { "type": "UNKNOWN", "characteristics": "", "confidence": 0.0, "evidence": ["NO_VALVE_TYPE_IDENTIFIED"], "typical_connections": [], "special_features": [] } def classify_valve_connection(dat_file: str, description: str) -> Dict: """밸브 연결 방식 분류""" dat_upper = dat_file.upper() desc_upper = description.upper() combined_text = f"{dat_upper} {desc_upper}" # 1. DAT_FILE 패턴 우선 확인 for connection_type, conn_data in VALVE_CONNECTIONS.items(): for pattern in conn_data["dat_patterns"]: if pattern in dat_upper: return { "method": connection_type, "confidence": 0.95, "matched_pattern": pattern, "source": "DAT_FILE_PATTERN", "size_range": conn_data["size_range"], "pressure_range": conn_data["pressure_range"], "typical_manufacturing": conn_data["manufacturing"] } # 2. 키워드 확인 for connection_type, conn_data in VALVE_CONNECTIONS.items(): for code in conn_data["codes"]: if code in combined_text: return { "method": connection_type, "confidence": conn_data["confidence"], "matched_code": code, "source": "KEYWORD_MATCH", "size_range": conn_data["size_range"], "pressure_range": conn_data["pressure_range"], "typical_manufacturing": conn_data["manufacturing"] } return { "method": "UNKNOWN", "confidence": 0.0, "matched_code": "", "source": "NO_CONNECTION_METHOD_FOUND" } def classify_valve_pressure_rating(dat_file: str, description: str) -> Dict: """밸브 압력 등급 분류""" combined_text = f"{dat_file} {description}".upper() # 패턴 매칭으로 압력 등급 추출 for pattern in VALVE_PRESSURE_RATINGS["patterns"]: match = re.search(pattern, combined_text) if match: rating_num = match.group(1) # WOG 처리 (Water Oil Gas) if "WOG" in pattern: rating = f"{rating_num}WOG" # WOG를 LB로 변환 (대략적) if int(rating_num) <= 600: equivalent_lb = "150LB" elif int(rating_num) <= 1000: equivalent_lb = "300LB" else: equivalent_lb = "600LB" return { "rating": f"{rating} ({equivalent_lb} 상당)", "confidence": 0.8, "matched_pattern": pattern, "max_pressure": f"{rating_num} PSI", "typical_manufacturing": "CAST_OR_FORGED" } else: rating = f"{rating_num}LB" # 표준 등급 정보 확인 rating_info = VALVE_PRESSURE_RATINGS["standard_ratings"].get(rating, {}) if rating_info: confidence = 0.95 else: confidence = 0.8 rating_info = { "max_pressure": "확인 필요", "typical_manufacturing": "UNKNOWN" } return { "rating": rating, "confidence": confidence, "matched_pattern": pattern, "matched_value": rating_num, "max_pressure": rating_info.get("max_pressure", ""), "typical_manufacturing": rating_info.get("typical_manufacturing", "") } return { "rating": "UNKNOWN", "confidence": 0.0, "matched_pattern": "", "max_pressure": "", "typical_manufacturing": "" } def classify_valve_actuation(description: str) -> Dict: """밸브 작동 방식 분류""" desc_upper = description.upper() # 키워드 기반 작동 방식 분류 for actuation_type, act_data in VALVE_ACTUATION.items(): for keyword in act_data["keywords"]: if keyword in desc_upper: return { "method": actuation_type, "characteristics": act_data["characteristics"], "confidence": 0.9, "matched_keyword": keyword, "applications": act_data["applications"] } # 기본값: MANUAL return { "method": "MANUAL", "characteristics": "수동 조작 (기본값)", "confidence": 0.6, "matched_keyword": "DEFAULT", "applications": "일반 수동 조작" } def determine_valve_manufacturing(material_result: Dict, valve_type_result: Dict, connection_result: Dict, pressure_result: Dict, main_nom: str) -> Dict: """밸브 제작 방법 결정 (주조 vs 단조)""" evidence = [] # 1. 재질 기반 제작방법 (가장 확실) material_manufacturing = get_manufacturing_method_from_material(material_result) if material_manufacturing in ["FORGED", "CAST"]: evidence.append(f"MATERIAL_STANDARD: {material_result.get('standard')}") characteristics = { "FORGED": "단조품 - 고강도, 소구경, 고압용", "CAST": "주조품 - 대구경, 복잡형상, 중저압용" }.get(material_manufacturing, "") return { "method": material_manufacturing, "confidence": 0.9, "evidence": evidence, "characteristics": characteristics } # 2. 단조 밸브 조건 확인 forged_indicators = 0 # 연결방식이 소켓웰드 connection_method = connection_result.get('method', '') if connection_method == "SOCKET_WELD": forged_indicators += 2 evidence.append(f"SOCKET_WELD_CONNECTION") # 고압 등급 pressure_rating = pressure_result.get('rating', '') high_pressure = ["800LB", "1500LB", "2500LB", "3000LB", "6000LB", "9000LB"] if any(pressure in pressure_rating for pressure in high_pressure): forged_indicators += 2 evidence.append(f"HIGH_PRESSURE: {pressure_rating}") # 소구경 try: size_num = float(re.findall(r'(\d+(?:\.\d+)?)', main_nom)[0]) if size_num <= 4.0: forged_indicators += 1 evidence.append(f"SMALL_SIZE: {main_nom}") except: pass # 니들 밸브는 일반적으로 단조 valve_type = valve_type_result.get('type', '') if valve_type == "NEEDLE_VALVE": forged_indicators += 2 evidence.append("NEEDLE_VALVE_TYPICALLY_FORGED") # 단조 결정 if forged_indicators >= 3: return { "method": "FORGED", "confidence": 0.85, "evidence": evidence, "characteristics": "단조품 - 고압, 소구경용" } # 3. 압력등급별 일반적 제작방법 pressure_manufacturing = pressure_result.get('typical_manufacturing', '') if pressure_manufacturing: if pressure_manufacturing == "FORGED": evidence.append(f"PRESSURE_BASED: {pressure_rating}") return { "method": "FORGED", "confidence": 0.75, "evidence": evidence, "characteristics": "고압용 단조품" } elif pressure_manufacturing == "CAST": evidence.append(f"PRESSURE_BASED: {pressure_rating}") return { "method": "CAST", "confidence": 0.75, "evidence": evidence, "characteristics": "저중압용 주조품" } # 4. 연결방식별 일반적 제작방법 connection_manufacturing = connection_result.get('typical_manufacturing', '') if connection_manufacturing: evidence.append(f"CONNECTION_BASED: {connection_method}") if connection_manufacturing == "FORGED": return { "method": "FORGED", "confidence": 0.7, "evidence": evidence, "characteristics": "소구경 단조품" } elif connection_manufacturing == "CAST": return { "method": "CAST", "confidence": 0.7, "evidence": evidence, "characteristics": "대구경 주조품" } # 5. 기본 추정 (사이즈 기반) try: size_num = float(re.findall(r'(\d+(?:\.\d+)?)', main_nom)[0]) if size_num <= 2.0: return { "method": "FORGED", "confidence": 0.6, "evidence": ["SIZE_BASED_SMALL"], "characteristics": "소구경 - 일반적으로 단조품" } else: return { "method": "CAST", "confidence": 0.6, "evidence": ["SIZE_BASED_LARGE"], "characteristics": "대구경 - 일반적으로 주조품" } except: pass return { "method": "UNKNOWN", "confidence": 0.0, "evidence": ["INSUFFICIENT_MANUFACTURING_INFO"], "characteristics": "" } def extract_valve_special_features(description: str, valve_type_result: Dict) -> List[str]: """밸브 특수 기능 추출""" desc_upper = description.upper() features = [] # 밸브 타입별 특수 기능 valve_special_features = valve_type_result.get('special_features', []) for feature in valve_special_features: # 기능별 키워드 매핑 feature_keywords = { "OS&Y": ["OS&Y", "OUTSIDE SCREW"], "FULL_PORT": ["FULL PORT", "FULL BORE"], "REDUCED_PORT": ["REDUCED PORT", "REDUCED BORE"], "3_WAY": ["3 WAY", "3-WAY", "THREE WAY"], "SWING_TYPE": ["SWING", "SWING TYPE"], "LIFT_TYPE": ["LIFT", "LIFT TYPE"], "GEAR_OPERATED": ["GEAR OPERATED", "GEAR"], "SET_PRESSURE": ["SET PRESSURE", "SET @"] } keywords = feature_keywords.get(feature, [feature]) for keyword in keywords: if keyword in desc_upper: features.append(feature) break # 일반적인 특수 기능들 general_features = { "FIRE_SAFE": ["FIRE SAFE", "FIRE-SAFE"], "ANTI_STATIC": ["ANTI STATIC", "ANTI-STATIC"], "BLOW_OUT_PROOF": ["BLOW OUT PROOF", "BOP"], "EXTENDED_STEM": ["EXTENDED STEM", "EXT STEM"], "CRYOGENIC": ["CRYOGENIC", "CRYO"], "HIGH_TEMPERATURE": ["HIGH TEMP", "HT"] } for feature, keywords in general_features.items(): for keyword in keywords: if keyword in desc_upper: features.append(feature) break return list(set(features)) # 중복 제거 def calculate_valve_confidence(confidence_scores: Dict) -> float: """밸브 분류 전체 신뢰도 계산""" scores = [score for score in confidence_scores.values() if score > 0] if not scores: return 0.0 # 가중 평균 weights = { "material": 0.2, "valve_type": 0.4, "connection": 0.25, "pressure": 0.15 } weighted_sum = sum( confidence_scores.get(key, 0) * weight for key, weight in weights.items() ) return round(weighted_sum, 2) # ========== 특수 기능들 ========== def is_forged_valve(valve_result: Dict) -> bool: """단조 밸브 여부 판단""" return valve_result.get("manufacturing", {}).get("method") == "FORGED" def is_high_pressure_valve(valve_result: Dict) -> bool: """고압 밸브 여부 판단""" pressure_rating = valve_result.get("pressure_rating", {}).get("rating", "") high_pressure_ratings = ["800LB", "1500LB", "2500LB", "3000LB", "6000LB", "9000LB"] return any(pressure in pressure_rating for pressure in high_pressure_ratings) def get_valve_purchase_info(valve_result: Dict) -> Dict: """밸브 구매 정보 생성""" valve_type = valve_result["valve_type"]["type"] connection = valve_result["connection_method"]["method"] pressure = valve_result["pressure_rating"]["rating"] manufacturing = valve_result["manufacturing"]["method"] actuation = valve_result["actuation"]["method"] # 공급업체 타입 결정 if manufacturing == "FORGED": supplier_type = "단조 밸브 전문업체" elif valve_type == "BUTTERFLY_VALVE": supplier_type = "버터플라이 밸브 전문업체" elif actuation in ["PNEUMATIC", "ELECTRIC"]: supplier_type = "자동 밸브 전문업체" else: supplier_type = "일반 밸브 업체" # 납기 추정 if manufacturing == "FORGED" and is_high_pressure_valve(valve_result): lead_time = "8-12주 (단조 고압용)" elif actuation in ["PNEUMATIC", "ELECTRIC"]: lead_time = "6-10주 (자동 밸브)" elif manufacturing == "FORGED": lead_time = "6-8주 (단조품)" else: lead_time = "4-8주 (일반품)" return { "supplier_type": supplier_type, "lead_time_estimate": lead_time, "purchase_category": f"{valve_type} {connection} {pressure}", "manufacturing_note": valve_result["manufacturing"]["characteristics"], "actuation_note": valve_result["actuation"]["characteristics"], "special_requirements": valve_result["special_features"] }