""" FLANGE 분류 시스템 일반 플랜지 + SPECIAL 플랜지 분류 """ import re from typing import Dict, List, Optional from .material_classifier import classify_material, get_manufacturing_method_from_material # ========== SPECIAL FLANGE 타입 ========== SPECIAL_FLANGE_TYPES = { "ORIFICE": { "dat_file_patterns": ["FLG_ORI_", "ORI_", "ORIFICE_"], "description_keywords": ["ORIFICE", "오리피스", "유량측정", "구멍"], "characteristics": "유량 측정용 구멍", "special_features": ["BORE_SIZE", "FLOW_MEASUREMENT"] }, "SPECTACLE_BLIND": { "dat_file_patterns": ["FLG_SPB_", "SPB_", "SPEC_"], "description_keywords": ["SPECTACLE BLIND", "SPECTACLE", "안경형", "SPB"], "characteristics": "안경형 차단판 (운전/정지 전환)", "special_features": ["SWITCHING", "ISOLATION"] }, "PADDLE_BLIND": { "dat_file_patterns": ["FLG_PAD_", "PAD_", "PADDLE_"], "description_keywords": ["PADDLE BLIND", "PADDLE", "패들형"], "characteristics": "패들형 차단판", "special_features": ["PERMANENT_ISOLATION"] }, "SPACER": { "dat_file_patterns": ["FLG_SPC_", "SPC_", "SPACER_"], "description_keywords": ["SPACER", "스페이서", "거리조정"], "characteristics": "거리 조정용", "special_features": ["SPACING", "THICKNESS"] }, "REDUCING": { "dat_file_patterns": ["FLG_RED_", "RED_FLG"], "description_keywords": ["REDUCING", "축소", "RED"], "characteristics": "사이즈 축소용", "special_features": ["SIZE_CHANGE", "REDUCING"], "requires_two_sizes": True }, "EXPANDER": { "dat_file_patterns": ["FLG_EXP_", "EXP_"], "description_keywords": ["EXPANDER", "EXPANDING", "확대"], "characteristics": "사이즈 확대용", "special_features": ["SIZE_CHANGE", "EXPANDING"], "requires_two_sizes": True }, "SWIVEL": { "dat_file_patterns": ["FLG_SWV_", "SWV_"], "description_keywords": ["SWIVEL", "회전", "ROTATING"], "characteristics": "회전/각도 조정용", "special_features": ["ROTATION", "ANGLE_ADJUSTMENT"] }, "INSULATION_SET": { "dat_file_patterns": ["FLG_INS_", "INS_SET"], "description_keywords": ["INSULATION", "절연", "ISOLATING", "INS SET"], "characteristics": "절연 플랜지 세트", "special_features": ["ELECTRICAL_ISOLATION"] }, "DRIP_RING": { "dat_file_patterns": ["FLG_DRP_", "DRP_"], "description_keywords": ["DRIP RING", "드립링", "DRIP"], "characteristics": "드립 링", "special_features": ["DRIP_PREVENTION"] }, "NOZZLE": { "dat_file_patterns": ["FLG_NOZ_", "NOZ_"], "description_keywords": ["NOZZLE", "노즐", "OUTLET"], "characteristics": "노즐 플랜지 (특수 형상)", "special_features": ["SPECIAL_SHAPE", "OUTLET"] } } # ========== 일반 FLANGE 타입 ========== STANDARD_FLANGE_TYPES = { "WELD_NECK": { "dat_file_patterns": ["FLG_WN_", "WN_", "WELD_NECK"], "description_keywords": ["WELD NECK", "WN", "웰드넥"], "characteristics": "목 부분 용접형", "size_range": "1/2\" ~ 48\"", "pressure_range": "150LB ~ 2500LB" }, "SLIP_ON": { "dat_file_patterns": ["FLG_SO_", "SO_", "SLIP_ON"], "description_keywords": ["SLIP ON", "SO", "슬립온"], "characteristics": "끼워서 용접형", "size_range": "1/2\" ~ 24\"", "pressure_range": "150LB ~ 600LB" }, "SOCKET_WELD": { "dat_file_patterns": ["FLG_SW_", "SW_"], "description_keywords": ["SOCKET WELD", "SW", "소켓웰드"], "characteristics": "소켓 용접형", "size_range": "1/8\" ~ 4\"", "pressure_range": "150LB ~ 9000LB" }, "THREADED": { "dat_file_patterns": ["FLG_THD_", "THD_", "FLG_TR_"], "description_keywords": ["THREADED", "THD", "나사", "NPT"], "characteristics": "나사 연결형", "size_range": "1/8\" ~ 4\"", "pressure_range": "150LB ~ 6000LB" }, "BLIND": { "dat_file_patterns": ["FLG_BL_", "BL_", "BLIND_"], "description_keywords": ["BLIND", "BL", "막음", "차단"], "characteristics": "막음용", "size_range": "1/2\" ~ 48\"", "pressure_range": "150LB ~ 2500LB" }, "LAP_JOINT": { "dat_file_patterns": ["FLG_LJ_", "LJ_", "LAP_"], "description_keywords": ["LAP JOINT", "LJ", "랩조인트"], "characteristics": "스터브엔드 조합형", "size_range": "1/2\" ~ 24\"", "pressure_range": "150LB ~ 600LB" } } # ========== 면 가공별 분류 ========== FACE_FINISHES = { "RAISED_FACE": { "codes": ["RF", "RAISED FACE", "볼록면"], "characteristics": "볼록한 면 (가장 일반적)", "pressure_range": "150LB ~ 600LB", "confidence": 0.95 }, "FLAT_FACE": { "codes": ["FF", "FLAT FACE", "평면"], "characteristics": "평평한 면", "pressure_range": "150LB ~ 300LB", "confidence": 0.95 }, "RING_TYPE_JOINT": { "codes": ["RTJ", "RING TYPE JOINT", "링타입"], "characteristics": "홈이 파진 고압용", "pressure_range": "600LB ~ 2500LB", "confidence": 0.95 } } # ========== 압력 등급별 분류 ========== FLANGE_PRESSURE_RATINGS = { "patterns": [ r"(\d+)LB", r"CLASS\s*(\d+)", r"CL\s*(\d+)", r"(\d+)#", r"(\d+)\s*LB" ], "standard_ratings": { "150LB": {"max_pressure": "285 PSI", "common_use": "저압 일반용", "typical_face": "RF"}, "300LB": {"max_pressure": "740 PSI", "common_use": "중압용", "typical_face": "RF"}, "600LB": {"max_pressure": "1480 PSI", "common_use": "고압용", "typical_face": "RTJ"}, "900LB": {"max_pressure": "2220 PSI", "common_use": "고압용", "typical_face": "RTJ"}, "1500LB": {"max_pressure": "3705 PSI", "common_use": "고압용", "typical_face": "RTJ"}, "2500LB": {"max_pressure": "6170 PSI", "common_use": "초고압용", "typical_face": "RTJ"}, "3000LB": {"max_pressure": "7400 PSI", "common_use": "소구경 고압용", "typical_face": "RTJ"}, "6000LB": {"max_pressure": "14800 PSI", "common_use": "소구경 초고압용", "typical_face": "RTJ"}, "9000LB": {"max_pressure": "22200 PSI", "common_use": "소구경 극고압용", "typical_face": "RTJ"} } } def classify_flange(dat_file: str, description: str, main_nom: str, red_nom: str = None, length: float = None) -> Dict: """ 완전한 FLANGE 분류 Args: dat_file: DAT_FILE 필드 description: DESCRIPTION 필드 main_nom: MAIN_NOM 필드 (주 사이즈) red_nom: RED_NOM 필드 (축소 사이즈, REDUCING 플랜지용) Returns: 완전한 플랜지 분류 결과 """ desc_upper = description.upper() dat_upper = dat_file.upper() # 1. 명칭 우선 확인 (플랜지 키워드가 있으면 플랜지) flange_keywords = ['FLG', 'FLANGE', '플랜지', 'ORIFICE', 'SPECTACLE', 'PADDLE', 'SPACER'] is_flange = any(keyword in desc_upper or keyword in dat_upper for keyword in flange_keywords) if not is_flange: return { "category": "UNKNOWN", "overall_confidence": 0.0, "reason": "플랜지 키워드 없음" } # 2. 재질 분류 (공통 모듈 사용) material_result = classify_material(description) # 2. SPECIAL vs STANDARD 분류 flange_category_result = classify_flange_category(dat_file, description) # 3. 플랜지 타입 분류 flange_type_result = classify_flange_type( dat_file, description, main_nom, red_nom, flange_category_result ) # 4. 면 가공 분류 face_finish_result = classify_face_finish(dat_file, description) # 5. 압력 등급 분류 pressure_result = classify_flange_pressure_rating(dat_file, description) # 6. 제작 방법 추정 manufacturing_result = determine_flange_manufacturing( material_result, flange_type_result, pressure_result, main_nom ) # 7. 최종 결과 조합 return { "category": "FLANGE", # 재질 정보 (공통 모듈) "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) }, # 플랜지 분류 정보 "flange_category": { "category": flange_category_result.get('category', 'UNKNOWN'), "is_special": flange_category_result.get('is_special', False), "confidence": flange_category_result.get('confidence', 0.0) }, "flange_type": { "type": flange_type_result.get('type', 'UNKNOWN'), "characteristics": flange_type_result.get('characteristics', ''), "confidence": flange_type_result.get('confidence', 0.0), "evidence": flange_type_result.get('evidence', []), "special_features": flange_type_result.get('special_features', []) }, "face_finish": { "finish": face_finish_result.get('finish', 'UNKNOWN'), "characteristics": face_finish_result.get('characteristics', ''), "confidence": face_finish_result.get('confidence', 0.0) }, "pressure_rating": { "rating": pressure_result.get('rating', 'UNKNOWN'), "confidence": pressure_result.get('confidence', 0.0), "max_pressure": pressure_result.get('max_pressure', ''), "common_use": pressure_result.get('common_use', ''), "typical_face": pressure_result.get('typical_face', '') }, "manufacturing": { "method": manufacturing_result.get('method', 'UNKNOWN'), "confidence": manufacturing_result.get('confidence', 0.0), "evidence": manufacturing_result.get('evidence', []), "characteristics": manufacturing_result.get('characteristics', '') }, "size_info": { "main_size": main_nom, "reduced_size": red_nom, "size_description": format_flange_size(main_nom, red_nom), "requires_two_sizes": flange_type_result.get('requires_two_sizes', False) }, # 전체 신뢰도 "overall_confidence": calculate_flange_confidence({ "material": material_result.get('confidence', 0), "flange_type": flange_type_result.get('confidence', 0), "face_finish": face_finish_result.get('confidence', 0), "pressure": pressure_result.get('confidence', 0) }) } def classify_flange_category(dat_file: str, description: str) -> Dict: """SPECIAL vs STANDARD 플랜지 분류""" dat_upper = dat_file.upper() desc_upper = description.upper() combined_text = f"{dat_upper} {desc_upper}" # SPECIAL 플랜지 확인 (우선) for special_type, type_data in SPECIAL_FLANGE_TYPES.items(): # DAT_FILE 패턴 확인 for pattern in type_data["dat_file_patterns"]: if pattern in dat_upper: return { "category": "SPECIAL", "special_type": special_type, "is_special": True, "confidence": 0.95, "evidence": [f"SPECIAL_DAT_PATTERN: {pattern}"] } # DESCRIPTION 키워드 확인 for keyword in type_data["description_keywords"]: if keyword in combined_text: return { "category": "SPECIAL", "special_type": special_type, "is_special": True, "confidence": 0.9, "evidence": [f"SPECIAL_KEYWORD: {keyword}"] } # STANDARD 플랜지로 분류 return { "category": "STANDARD", "is_special": False, "confidence": 0.8, "evidence": ["NO_SPECIAL_INDICATORS"] } def classify_flange_type(dat_file: str, description: str, main_nom: str, red_nom: str, category_result: Dict) -> Dict: """플랜지 타입 분류 (SPECIAL 또는 STANDARD)""" dat_upper = dat_file.upper() desc_upper = description.upper() if category_result.get('is_special'): # SPECIAL 플랜지 타입 확인 special_type = category_result.get('special_type') if special_type and special_type in SPECIAL_FLANGE_TYPES: type_data = SPECIAL_FLANGE_TYPES[special_type] return { "type": special_type, "characteristics": type_data["characteristics"], "confidence": 0.95, "evidence": [f"SPECIAL_TYPE: {special_type}"], "special_features": type_data["special_features"], "requires_two_sizes": type_data.get("requires_two_sizes", False) } else: # STANDARD 플랜지 타입 확인 for flange_type, type_data in STANDARD_FLANGE_TYPES.items(): # DAT_FILE 패턴 확인 for pattern in type_data["dat_file_patterns"]: if pattern in dat_upper: return { "type": flange_type, "characteristics": type_data["characteristics"], "confidence": 0.95, "evidence": [f"STANDARD_DAT_PATTERN: {pattern}"], "special_features": [], "requires_two_sizes": False } # DESCRIPTION 키워드 확인 for keyword in type_data["description_keywords"]: if keyword in desc_upper: return { "type": flange_type, "characteristics": type_data["characteristics"], "confidence": 0.85, "evidence": [f"STANDARD_KEYWORD: {keyword}"], "special_features": [], "requires_two_sizes": False } # 분류 실패 return { "type": "UNKNOWN", "characteristics": "", "confidence": 0.0, "evidence": ["NO_FLANGE_TYPE_IDENTIFIED"], "special_features": [], "requires_two_sizes": False } def classify_face_finish(dat_file: str, description: str) -> Dict: """플랜지 면 가공 분류""" combined_text = f"{dat_file} {description}".upper() for finish_type, finish_data in FACE_FINISHES.items(): for code in finish_data["codes"]: if code in combined_text: return { "finish": finish_type, "confidence": finish_data["confidence"], "matched_code": code, "characteristics": finish_data["characteristics"], "pressure_range": finish_data["pressure_range"] } # 기본값: RF (가장 일반적) return { "finish": "RAISED_FACE", "confidence": 0.6, "matched_code": "DEFAULT", "characteristics": "기본값 (가장 일반적)", "pressure_range": "150LB ~ 600LB" } def classify_flange_pressure_rating(dat_file: str, description: str) -> Dict: """플랜지 압력 등급 분류""" combined_text = f"{dat_file} {description}".upper() # 패턴 매칭으로 압력 등급 추출 for pattern in FLANGE_PRESSURE_RATINGS["patterns"]: match = re.search(pattern, combined_text) if match: rating_num = match.group(1) rating = f"{rating_num}LB" # 표준 등급 정보 확인 rating_info = FLANGE_PRESSURE_RATINGS["standard_ratings"].get(rating, {}) if rating_info: confidence = 0.95 else: confidence = 0.8 rating_info = {"max_pressure": "확인 필요", "common_use": "비표준 등급", "typical_face": "RF"} return { "rating": rating, "confidence": confidence, "matched_pattern": pattern, "matched_value": rating_num, "max_pressure": rating_info.get("max_pressure", ""), "common_use": rating_info.get("common_use", ""), "typical_face": rating_info.get("typical_face", "RF") } return { "rating": "UNKNOWN", "confidence": 0.0, "matched_pattern": "", "max_pressure": "", "common_use": "", "typical_face": "" } def determine_flange_manufacturing(material_result: Dict, flange_type_result: Dict, pressure_result: Dict, main_nom: str) -> Dict: """플랜지 제작 방법 결정""" 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. 압력등급 + 사이즈 조합으로 추정 pressure_rating = pressure_result.get('rating', '') # 고압 = 단조 high_pressure = ["600LB", "900LB", "1500LB", "2500LB", "3000LB", "6000LB", "9000LB"] if any(pressure in pressure_rating for pressure in high_pressure): evidence.append(f"HIGH_PRESSURE: {pressure_rating}") return { "method": "FORGED", "confidence": 0.85, "evidence": evidence, "characteristics": "고압용 단조품" } # 저압 + 대구경 = 주조 가능 low_pressure = ["150LB", "300LB"] if any(pressure in pressure_rating for pressure in low_pressure): try: size_num = float(re.findall(r'(\d+(?:\.\d+)?)', main_nom)[0]) if size_num >= 12.0: # 12인치 이상 evidence.append(f"LARGE_SIZE_LOW_PRESSURE: {main_nom}, {pressure_rating}") return { "method": "CAST", "confidence": 0.8, "evidence": evidence, "characteristics": "대구경 저압용 주조품" } except: pass # 3. 기본 추정 return { "method": "FORGED", "confidence": 0.7, "evidence": ["DEFAULT_FORGED"], "characteristics": "일반적으로 단조품" } def format_flange_size(main_nom: str, red_nom: str = None) -> str: """플랜지 사이즈 표기 포맷팅""" main_nom_str = str(main_nom) if main_nom is not None else "" red_nom_str = str(red_nom) if red_nom is not None else "" if red_nom_str.strip() and red_nom_str != main_nom_str: return f"{main_nom_str} x {red_nom_str}" else: return main_nom_str def calculate_flange_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.25, "flange_type": 0.4, "face_finish": 0.2, "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_high_pressure_flange(pressure_rating: str) -> bool: """고압 플랜지 여부 판단""" high_pressure_ratings = ["600LB", "900LB", "1500LB", "2500LB", "3000LB", "6000LB", "9000LB"] return pressure_rating in high_pressure_ratings def is_special_flange(flange_result: Dict) -> bool: """특수 플랜지 여부 판단""" return flange_result.get("flange_category", {}).get("is_special", False) def get_flange_purchase_info(flange_result: Dict) -> Dict: """플랜지 구매 정보 생성""" flange_type = flange_result["flange_type"]["type"] pressure = flange_result["pressure_rating"]["rating"] manufacturing = flange_result["manufacturing"]["method"] is_special = flange_result["flange_category"]["is_special"] # 공급업체 타입 결정 if is_special: supplier_type = "특수 플랜지 전문업체" elif manufacturing == "FORGED": supplier_type = "단조 플랜지 업체" elif manufacturing == "CAST": supplier_type = "주조 플랜지 업체" else: supplier_type = "일반 플랜지 업체" # 납기 추정 if is_special: lead_time = "8-12주 (특수품)" elif is_high_pressure_flange(pressure): lead_time = "6-10주 (고압용)" elif manufacturing == "FORGED": lead_time = "4-8주 (단조품)" else: lead_time = "2-6주 (일반품)" return { "supplier_type": supplier_type, "lead_time_estimate": lead_time, "purchase_category": f"{flange_type} {pressure}", "manufacturing_note": flange_result["manufacturing"]["characteristics"], "special_requirements": flange_result["flange_type"]["special_features"] }