diff --git a/backend/app/api/spools.py b/backend/app/api/spools.py new file mode 100644 index 0000000..a3e9314 --- /dev/null +++ b/backend/app/api/spools.py @@ -0,0 +1,231 @@ +""" +스풀 관리 API 엔드포인트 +""" + +from fastapi import APIRouter, Depends, HTTPException, Form +from sqlalchemy.orm import Session +from sqlalchemy import text +from typing import List, Optional +from datetime import datetime + +from ..database import get_db +from ..services.spool_manager import SpoolManager, classify_pipe_with_spool + +router = APIRouter() + +@router.get("/") +async def get_spools_info(): + return { + "message": "스풀 관리 API", + "features": [ + "스풀 식별자 생성", + "에리어/스풀 넘버 관리", + "스풀별 파이프 그룹핑", + "스풀 유효성 검증" + ] + } + +@router.post("/validate-identifier") +async def validate_spool_identifier( + spool_identifier: str = Form(...), + db: Session = Depends(get_db) +): + """스풀 식별자 유효성 검증""" + + spool_manager = SpoolManager() + validation_result = spool_manager.validate_spool_identifier(spool_identifier) + + return { + "spool_identifier": spool_identifier, + "validation": validation_result, + "timestamp": datetime.now().isoformat() + } + +@router.post("/generate-identifier") +async def generate_spool_identifier( + dwg_name: str = Form(...), + area_number: str = Form(...), + spool_number: str = Form(...), + db: Session = Depends(get_db) +): + """새로운 스풀 식별자 생성""" + + try: + spool_manager = SpoolManager() + spool_identifier = spool_manager.generate_spool_identifier( + dwg_name, area_number, spool_number + ) + + # 중복 확인 + duplicate_check = db.execute( + text("SELECT id FROM spools WHERE spool_identifier = :spool_id"), + {"spool_id": spool_identifier} + ).fetchone() + + return { + "success": True, + "spool_identifier": spool_identifier, + "is_duplicate": bool(duplicate_check), + "components": { + "dwg_name": dwg_name, + "area_number": spool_manager.format_area_number(area_number), + "spool_number": spool_manager.format_spool_number(spool_number) + } + } + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + +@router.get("/next-spool-number") +async def get_next_spool_number( + dwg_name: str, + area_number: str, + db: Session = Depends(get_db) +): + """다음 사용 가능한 스풀 넘버 추천""" + + try: + # 기존 스풀들 조회 + existing_spools_query = text(""" + SELECT spool_identifier + FROM spools + WHERE dwg_name = :dwg_name + """) + + result = db.execute(existing_spools_query, {"dwg_name": dwg_name}) + existing_spools = [row[0] for row in result.fetchall()] + + spool_manager = SpoolManager() + next_spool = spool_manager.get_next_spool_number( + dwg_name, area_number, existing_spools + ) + + return { + "dwg_name": dwg_name, + "area_number": spool_manager.format_area_number(area_number), + "next_spool_number": next_spool, + "existing_spools_count": len(existing_spools) + } + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + +@router.post("/materials/{material_id}/assign-spool") +async def assign_spool_to_material( + material_id: int, + area_number: str = Form(...), + spool_number: str = Form(...), + db: Session = Depends(get_db) +): + """자재에 스풀 정보 할당""" + + try: + # 자재 정보 조회 + material_query = text(""" + SELECT m.id, m.file_id, f.project_id, f.dwg_name + FROM materials m + JOIN files f ON m.file_id = f.id + WHERE m.id = :material_id + """) + + material = db.execute(material_query, {"material_id": material_id}).fetchone() + if not material: + raise HTTPException(status_code=404, detail="자재를 찾을 수 없습니다") + + # 스풀 식별자 생성 + spool_manager = SpoolManager() + area_formatted = spool_manager.format_area_number(area_number) + spool_formatted = spool_manager.format_spool_number(spool_number) + spool_identifier = spool_manager.generate_spool_identifier( + material.dwg_name, area_formatted, spool_formatted + ) + + # 자재 업데이트 + update_query = text(""" + UPDATE materials + SET area_number = :area_number, + spool_number = :spool_number, + spool_identifier = :spool_identifier, + spool_input_required = FALSE, + spool_validated = TRUE, + updated_at = NOW() + WHERE id = :material_id + """) + + db.execute(update_query, { + "material_id": material_id, + "area_number": area_formatted, + "spool_number": spool_formatted, + "spool_identifier": spool_identifier + }) + + # 스풀 테이블에 등록 (없다면) + spool_insert_query = text(""" + INSERT INTO spools (project_id, dwg_name, area_number, spool_number, spool_identifier, created_at) + VALUES (:project_id, :dwg_name, :area_number, :spool_number, :spool_identifier, NOW()) + ON CONFLICT (project_id, dwg_name, area_number, spool_number) DO NOTHING + """) + + db.execute(spool_insert_query, { + "project_id": material.project_id, + "dwg_name": material.dwg_name, + "area_number": area_formatted, + "spool_number": spool_formatted, + "spool_identifier": spool_identifier + }) + + db.commit() + + return { + "success": True, + "material_id": material_id, + "spool_identifier": spool_identifier, + "message": "스풀 정보가 할당되었습니다" + } + + except ValueError as e: + db.rollback() + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + db.rollback() + raise HTTPException(status_code=500, detail=f"스풀 할당 실패: {str(e)}") + +@router.get("/project/{project_id}/spools") +async def get_project_spools( + project_id: int, + db: Session = Depends(get_db) +): + """프로젝트의 모든 스풀 조회""" + + query = text(""" + SELECT s.*, + COUNT(m.id) as material_count, + SUM(m.quantity) as total_quantity + FROM spools s + LEFT JOIN materials m ON s.spool_identifier = m.spool_identifier + WHERE s.project_id = :project_id + GROUP BY s.id + ORDER BY s.dwg_name, s.area_number, s.spool_number + """) + + result = db.execute(query, {"project_id": project_id}) + spools = result.fetchall() + + return { + "project_id": project_id, + "spools_count": len(spools), + "spools": [ + { + "id": spool.id, + "spool_identifier": spool.spool_identifier, + "dwg_name": spool.dwg_name, + "area_number": spool.area_number, + "spool_number": spool.spool_number, + "material_count": spool.material_count or 0, + "total_quantity": float(spool.total_quantity or 0), + "status": spool.status, + "created_at": spool.created_at + } + for spool in spools + ] + } diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000..57d4295 --- /dev/null +++ b/backend/app/services/__init__.py @@ -0,0 +1,13 @@ +""" +서비스 모듈 +""" + +from .material_classifier import classify_material, get_manufacturing_method_from_material +from .materials_schema import MATERIAL_STANDARDS, SPECIAL_MATERIALS + +__all__ = [ + 'classify_material', + 'get_manufacturing_method_from_material', + 'MATERIAL_STANDARDS', + 'SPECIAL_MATERIALS' +] diff --git a/backend/app/services/fitting_classifier.py b/backend/app/services/fitting_classifier.py new file mode 100644 index 0000000..addb852 --- /dev/null +++ b/backend/app/services/fitting_classifier.py @@ -0,0 +1,588 @@ +""" +FITTING 분류 시스템 V2 +재질 분류 + 피팅 특화 분류 + 스풀 시스템 통합 +""" + +import re +from typing import Dict, List, Optional +from .material_classifier import classify_material, get_manufacturing_method_from_material + +# ========== FITTING 타입별 분류 (실제 BOM 기반) ========== +FITTING_TYPES = { + "ELBOW": { + "dat_file_patterns": ["90L_", "45L_", "ELL_", "ELBOW_"], + "description_keywords": ["ELBOW", "ELL", "엘보"], + "subtypes": { + "90DEG": ["90", "90°", "90DEG", "90도"], + "45DEG": ["45", "45°", "45DEG", "45도"], + "LONG_RADIUS": ["LR", "LONG RADIUS", "장반경"], + "SHORT_RADIUS": ["SR", "SHORT RADIUS", "단반경"] + }, + "default_subtype": "90DEG", + "common_connections": ["BUTT_WELD", "SOCKET_WELD"], + "size_range": "1/2\" ~ 48\"" + }, + + "TEE": { + "dat_file_patterns": ["TEE_", "T_"], + "description_keywords": ["TEE", "티"], + "subtypes": { + "EQUAL": ["EQUAL TEE", "등경티", "EQUAL"], + "REDUCING": ["REDUCING TEE", "RED TEE", "축소티", "REDUCING", "RD"] + }, + "size_analysis": True, # RED_NOM으로 REDUCING 여부 판단 + "common_connections": ["BUTT_WELD", "SOCKET_WELD"], + "size_range": "1/2\" ~ 48\"" + }, + + "REDUCER": { + "dat_file_patterns": ["CNC_", "ECC_", "RED_", "REDUCER_"], + "description_keywords": ["REDUCER", "RED", "리듀서"], + "subtypes": { + "CONCENTRIC": ["CONCENTRIC", "CNC", "동심", "CON"], + "ECCENTRIC": ["ECCENTRIC", "ECC", "편심"] + }, + "requires_two_sizes": True, + "common_connections": ["BUTT_WELD"], + "size_range": "1/2\" ~ 48\"" + }, + + "CAP": { + "dat_file_patterns": ["CAP_"], + "description_keywords": ["CAP", "캡", "막음"], + "subtypes": { + "BUTT_WELD": ["BW", "BUTT WELD"], + "SOCKET_WELD": ["SW", "SOCKET WELD"], + "THREADED": ["THD", "THREADED", "나사", "NPT"] + }, + "common_connections": ["BUTT_WELD", "SOCKET_WELD", "THREADED"], + "size_range": "1/4\" ~ 24\"" + }, + + "NIPPLE": { + "dat_file_patterns": ["NIP_", "NIPPLE_"], + "description_keywords": ["NIPPLE", "니플"], + "subtypes": { + "THREADED": ["THREADED", "THD", "NPT", "나사"], + "SOCKET_WELD": ["SOCKET WELD", "SW", "소켓웰드"], + "CLOSE": ["CLOSE NIPPLE", "CLOSE"], + "SHORT": ["SHORT NIPPLE", "SHORT"], + "LONG": ["LONG NIPPLE", "LONG"] + }, + "common_connections": ["THREADED", "SOCKET_WELD"], + "size_range": "1/8\" ~ 4\"" + }, + + "SWAGE": { + "dat_file_patterns": ["SWG_"], + "description_keywords": ["SWAGE", "스웨지"], + "subtypes": { + "CONCENTRIC": ["CONCENTRIC", "CN", "CON", "동심"], + "ECCENTRIC": ["ECCENTRIC", "EC", "ECC", "편심"] + }, + "requires_two_sizes": True, + "common_connections": ["BUTT_WELD", "SOCKET_WELD"], + "size_range": "1/2\" ~ 12\"" + }, + + "OLET": { + "dat_file_patterns": ["SOL_", "WOL_", "TOL_", "OLET_"], + "description_keywords": ["OLET", "올렛", "O-LET"], + "subtypes": { + "SOCKOLET": ["SOCK-O-LET", "SOCKOLET", "SOL"], + "WELDOLET": ["WELD-O-LET", "WELDOLET", "WOL"], + "THREADOLET": ["THREAD-O-LET", "THREADOLET", "TOL"], + "ELBOLET": ["ELB-O-LET", "ELBOLET", "EOL"] + }, + "requires_two_sizes": True, # 주배관 x 분기관 + "common_connections": ["SOCKET_WELD", "THREADED", "BUTT_WELD"], + "size_range": "1/8\" ~ 4\"" + }, + + "COUPLING": { + "dat_file_patterns": ["CPL_", "COUPLING_"], + "description_keywords": ["COUPLING", "커플링"], + "subtypes": { + "FULL": ["FULL COUPLING", "FULL"], + "HALF": ["HALF COUPLING", "HALF"], + "REDUCING": ["REDUCING COUPLING", "RED"] + }, + "common_connections": ["SOCKET_WELD", "THREADED"], + "size_range": "1/8\" ~ 4\"" + } +} + +# ========== 연결 방식별 분류 ========== +CONNECTION_METHODS = { + "BUTT_WELD": { + "codes": ["BW", "BUTT WELD", "맞대기용접", "BUTT-WELD"], + "dat_patterns": ["_BW"], + "size_range": "1/2\" ~ 48\"", + "pressure_range": "150LB ~ 2500LB", + "typical_manufacturing": "WELDED_FABRICATED", + "confidence": 0.95 + }, + "SOCKET_WELD": { + "codes": ["SW", "SOCKET WELD", "소켓웰드", "SOCKET-WELD"], + "dat_patterns": ["_SW_"], + "size_range": "1/8\" ~ 4\"", + "pressure_range": "150LB ~ 9000LB", + "typical_manufacturing": "FORGED", + "confidence": 0.95 + }, + "THREADED": { + "codes": ["THD", "THRD", "NPT", "THREADED", "나사", "TR"], + "dat_patterns": ["_TR", "_THD"], + "size_range": "1/8\" ~ 4\"", + "pressure_range": "150LB ~ 6000LB", + "typical_manufacturing": "FORGED", + "confidence": 0.95 + }, + "FLANGED": { + "codes": ["FL", "FLG", "FLANGED", "플랜지"], + "dat_patterns": ["_FL_"], + "size_range": "1/2\" ~ 48\"", + "pressure_range": "150LB ~ 2500LB", + "typical_manufacturing": "FORGED_OR_CAST", + "confidence": 0.9 + } +} + +# ========== 압력 등급별 분류 ========== +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": "저압 일반용"}, + "300LB": {"max_pressure": "740 PSI", "common_use": "중압용"}, + "600LB": {"max_pressure": "1480 PSI", "common_use": "고압용"}, + "900LB": {"max_pressure": "2220 PSI", "common_use": "고압용"}, + "1500LB": {"max_pressure": "3705 PSI", "common_use": "고압용"}, + "2500LB": {"max_pressure": "6170 PSI", "common_use": "초고압용"}, + "3000LB": {"max_pressure": "7400 PSI", "common_use": "소구경 고압용"}, + "6000LB": {"max_pressure": "14800 PSI", "common_use": "소구경 초고압용"}, + "9000LB": {"max_pressure": "22200 PSI", "common_use": "소구경 극고압용"} + } +} + +def classify_fitting(dat_file: str, description: str, main_nom: str, + red_nom: str = None) -> Dict: + """ + 완전한 FITTING 분류 + + Args: + dat_file: DAT_FILE 필드 + description: DESCRIPTION 필드 + main_nom: MAIN_NOM 필드 (주 사이즈) + red_nom: RED_NOM 필드 (축소 사이즈, 선택사항) + + Returns: + 완전한 피팅 분류 결과 + """ + + # 1. 재질 분류 (공통 모듈 사용) + material_result = classify_material(description) + + # 2. 피팅 타입 분류 + fitting_type_result = classify_fitting_type(dat_file, description, main_nom, red_nom) + + # 3. 연결 방식 분류 + connection_result = classify_connection_method(dat_file, description) + + # 4. 압력 등급 분류 + pressure_result = classify_pressure_rating(dat_file, description) + + # 5. 제작 방법 추정 + manufacturing_result = determine_fitting_manufacturing( + material_result, connection_result, pressure_result, main_nom + ) + + # 6. 최종 결과 조합 + return { + "category": "FITTING", + + # 재질 정보 (공통 모듈) + "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) + }, + + # 피팅 특화 정보 + "fitting_type": { + "type": fitting_type_result.get('type', 'UNKNOWN'), + "subtype": fitting_type_result.get('subtype', 'UNKNOWN'), + "confidence": fitting_type_result.get('confidence', 0.0), + "evidence": fitting_type_result.get('evidence', []) + }, + + "connection_method": { + "method": connection_result.get('method', 'UNKNOWN'), + "confidence": connection_result.get('confidence', 0.0), + "matched_code": connection_result.get('matched_code', ''), + "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', ''), + "common_use": pressure_result.get('common_use', '') + }, + + "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_fitting_size(main_nom, red_nom), + "requires_two_sizes": fitting_type_result.get('requires_two_sizes', False) + }, + + # 전체 신뢰도 + "overall_confidence": calculate_fitting_confidence({ + "material": material_result.get('confidence', 0), + "fitting_type": fitting_type_result.get('confidence', 0), + "connection": connection_result.get('confidence', 0), + "pressure": pressure_result.get('confidence', 0) + }) + } + +def classify_fitting_type(dat_file: str, description: str, + main_nom: str, red_nom: str = None) -> Dict: + """피팅 타입 분류""" + + dat_upper = dat_file.upper() + desc_upper = description.upper() + + # 1. DAT_FILE 패턴으로 1차 분류 (가장 신뢰도 높음) + for fitting_type, type_data in FITTING_TYPES.items(): + for pattern in type_data["dat_file_patterns"]: + if pattern in dat_upper: + subtype_result = classify_fitting_subtype( + fitting_type, desc_upper, main_nom, red_nom, type_data + ) + + return { + "type": fitting_type, + "subtype": subtype_result["subtype"], + "confidence": 0.95, + "evidence": [f"DAT_FILE_PATTERN: {pattern}"], + "subtype_confidence": subtype_result["confidence"], + "requires_two_sizes": type_data.get("requires_two_sizes", False) + } + + # 2. DESCRIPTION 키워드로 2차 분류 + for fitting_type, type_data in FITTING_TYPES.items(): + for keyword in type_data["description_keywords"]: + if keyword in desc_upper: + subtype_result = classify_fitting_subtype( + fitting_type, desc_upper, main_nom, red_nom, type_data + ) + + return { + "type": fitting_type, + "subtype": subtype_result["subtype"], + "confidence": 0.85, + "evidence": [f"DESCRIPTION_KEYWORD: {keyword}"], + "subtype_confidence": subtype_result["confidence"], + "requires_two_sizes": type_data.get("requires_two_sizes", False) + } + + # 3. 분류 실패 + return { + "type": "UNKNOWN", + "subtype": "UNKNOWN", + "confidence": 0.0, + "evidence": ["NO_FITTING_TYPE_IDENTIFIED"], + "requires_two_sizes": False + } + +def classify_fitting_subtype(fitting_type: str, description: str, + main_nom: str, red_nom: str, type_data: Dict) -> Dict: + """피팅 서브타입 분류""" + + subtypes = type_data.get("subtypes", {}) + + # 1. 키워드 기반 서브타입 분류 (우선) + for subtype, keywords in subtypes.items(): + for keyword in keywords: + if keyword in description: + return { + "subtype": subtype, + "confidence": 0.9, + "evidence": [f"SUBTYPE_KEYWORD: {keyword}"] + } + + # 2. 사이즈 분석이 필요한 경우 (TEE, REDUCER 등) + if type_data.get("size_analysis"): + if red_nom and red_nom.strip() and red_nom != main_nom: + return { + "subtype": "REDUCING", + "confidence": 0.85, + "evidence": [f"SIZE_ANALYSIS_REDUCING: {main_nom} x {red_nom}"] + } + else: + return { + "subtype": "EQUAL", + "confidence": 0.8, + "evidence": [f"SIZE_ANALYSIS_EQUAL: {main_nom}"] + } + + # 3. 두 사이즈가 필요한 경우 확인 + if type_data.get("requires_two_sizes"): + if red_nom and red_nom.strip(): + confidence = 0.8 + evidence = [f"TWO_SIZES_PROVIDED: {main_nom} x {red_nom}"] + else: + confidence = 0.6 + evidence = [f"TWO_SIZES_EXPECTED_BUT_MISSING"] + else: + confidence = 0.7 + evidence = ["SINGLE_SIZE_FITTING"] + + # 4. 기본값 + default_subtype = type_data.get("default_subtype", "GENERAL") + return { + "subtype": default_subtype, + "confidence": confidence, + "evidence": evidence + } + +def classify_connection_method(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 method, method_data in CONNECTION_METHODS.items(): + for pattern in method_data["dat_patterns"]: + if pattern in dat_upper: + return { + "method": method, + "confidence": 0.95, + "matched_code": pattern, + "source": "DAT_FILE_PATTERN", + "size_range": method_data["size_range"], + "pressure_range": method_data["pressure_range"], + "typical_manufacturing": method_data["typical_manufacturing"] + } + + # 2. 키워드 확인 + for method, method_data in CONNECTION_METHODS.items(): + for code in method_data["codes"]: + if code in combined_text: + return { + "method": method, + "confidence": method_data["confidence"], + "matched_code": code, + "source": "KEYWORD_MATCH", + "size_range": method_data["size_range"], + "pressure_range": method_data["pressure_range"], + "typical_manufacturing": method_data["typical_manufacturing"] + } + + return { + "method": "UNKNOWN", + "confidence": 0.0, + "matched_code": "", + "source": "NO_CONNECTION_METHOD_FOUND" + } + +def classify_pressure_rating(dat_file: str, description: str) -> Dict: + """압력 등급 분류""" + + combined_text = f"{dat_file} {description}".upper() + + # 패턴 매칭으로 압력 등급 추출 + for pattern in PRESSURE_RATINGS["patterns"]: + match = re.search(pattern, combined_text) + if match: + rating_num = match.group(1) + rating = f"{rating_num}LB" + + # 표준 등급 정보 확인 + rating_info = PRESSURE_RATINGS["standard_ratings"].get(rating, {}) + + if rating_info: + confidence = 0.95 + else: + confidence = 0.8 + rating_info = {"max_pressure": "확인 필요", "common_use": "비표준 등급"} + + 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", "") + } + + return { + "rating": "UNKNOWN", + "confidence": 0.0, + "matched_pattern": "", + "max_pressure": "", + "common_use": "" + } + +def determine_fitting_manufacturing(material_result: Dict, connection_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. 연결방식 + 압력등급 조합 추정 + connection_method = connection_result.get("method", "") + pressure_rating = pressure_result.get("rating", "") + + # 고압 + 소켓웰드/나사 = 단조 + high_pressure = ["3000LB", "6000LB", "9000LB"] + forged_connections = ["SOCKET_WELD", "THREADED"] + + if (any(pressure in pressure_rating for pressure in high_pressure) and + connection_method in forged_connections): + evidence.append(f"HIGH_PRESSURE: {pressure_rating}") + evidence.append(f"FORGED_CONNECTION: {connection_method}") + return { + "method": "FORGED", + "confidence": 0.85, + "evidence": evidence, + "characteristics": "고압용 단조품" + } + + # 3. 연결방식별 일반적 제작방법 + connection_manufacturing = connection_result.get("typical_manufacturing", "") + if connection_manufacturing: + evidence.append(f"CONNECTION_TYPICAL: {connection_method}") + + characteristics_map = { + "FORGED": "단조품, 고강도", + "WELDED_FABRICATED": "용접제작품, 대구경", + "FORGED_OR_CAST": "단조 또는 주조" + } + + return { + "method": connection_manufacturing, + "confidence": 0.7, + "evidence": evidence, + "characteristics": characteristics_map.get(connection_manufacturing, "") + } + + # 4. 기본 추정 + return { + "method": "UNKNOWN", + "confidence": 0.0, + "evidence": ["INSUFFICIENT_MANUFACTURING_INFO"], + "characteristics": "" + } + +def format_fitting_size(main_nom: str, red_nom: str = None) -> str: + """피팅 사이즈 표기 포맷팅""" + + if red_nom and red_nom.strip() and red_nom != main_nom: + return f"{main_nom} x {red_nom}" + else: + return main_nom + +def calculate_fitting_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, + "fitting_type": 0.4, + "connection": 0.25, + "pressure": 0.1 + } + + weighted_sum = sum( + confidence_scores.get(key, 0) * weight + for key, weight in weights.items() + ) + + return round(weighted_sum, 2) + +# ========== 특수 분류 함수들 ========== + +def is_high_pressure_fitting(pressure_rating: str) -> bool: + """고압 피팅 여부 판단""" + high_pressure_ratings = ["3000LB", "6000LB", "9000LB"] + return pressure_rating in high_pressure_ratings + +def is_small_bore_fitting(main_nom: str) -> bool: + """소구경 피팅 여부 판단""" + try: + # 간단한 사이즈 파싱 (인치 기준) + size_num = float(re.findall(r'(\d+(?:\.\d+)?)', main_nom)[0]) + return size_num <= 2.0 + except: + return False + +def get_fitting_purchase_info(fitting_result: Dict) -> Dict: + """피팅 구매 정보 생성""" + + fitting_type = fitting_result["fitting_type"]["type"] + connection = fitting_result["connection_method"]["method"] + pressure = fitting_result["pressure_rating"]["rating"] + manufacturing = fitting_result["manufacturing"]["method"] + + # 공급업체 타입 결정 + if manufacturing == "FORGED": + supplier_type = "단조 피팅 전문업체" + elif manufacturing == "CAST": + supplier_type = "주조 피팅 전문업체" + else: + supplier_type = "일반 피팅 업체" + + # 납기 추정 + if is_high_pressure_fitting(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"{fitting_type} {connection} {pressure}", + "manufacturing_note": fitting_result["manufacturing"]["characteristics"] + } diff --git a/backend/app/services/flange_classifier.py b/backend/app/services/flange_classifier.py new file mode 100644 index 0000000..d139c4f --- /dev/null +++ b/backend/app/services/flange_classifier.py @@ -0,0 +1,567 @@ +""" +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_"], + "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) -> Dict: + """ + 완전한 FLANGE 분류 + + Args: + dat_file: DAT_FILE 필드 + description: DESCRIPTION 필드 + main_nom: MAIN_NOM 필드 (주 사이즈) + red_nom: RED_NOM 필드 (축소 사이즈, REDUCING 플랜지용) + + Returns: + 완전한 플랜지 분류 결과 + """ + + # 1. 재질 분류 (공통 모듈 사용) + 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: + """플랜지 사이즈 표기 포맷팅""" + + if red_nom and red_nom.strip() and red_nom != main_nom: + return f"{main_nom} x {red_nom}" + else: + return main_nom + +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"] + } diff --git a/backend/app/services/material_classifier.py b/backend/app/services/material_classifier.py new file mode 100644 index 0000000..d07dc1f --- /dev/null +++ b/backend/app/services/material_classifier.py @@ -0,0 +1,313 @@ +""" +재질 분류를 위한 공통 함수 +materials_schema.py의 데이터를 사용하여 재질을 분류 +""" + +import re +from typing import Dict, List, Optional, Tuple +from .materials_schema import ( + MATERIAL_STANDARDS, + SPECIAL_MATERIALS, + MANUFACTURING_MATERIAL_MAP, + GENERIC_MATERIAL_KEYWORDS +) + +def classify_material(description: str) -> Dict: + """ + 공통 재질 분류 함수 + + Args: + description: 자재 설명 (DESCRIPTION 필드) + + Returns: + 재질 분류 결과 딕셔너리 + """ + + desc_upper = description.upper().strip() + + # 1단계: 특수 재질 우선 확인 (가장 구체적) + special_result = check_special_materials(desc_upper) + if special_result['confidence'] > 0.9: + return special_result + + # 2단계: ASTM/ASME 규격 확인 + astm_result = check_astm_materials(desc_upper) + if astm_result['confidence'] > 0.8: + return astm_result + + # 3단계: KS 규격 확인 + ks_result = check_ks_materials(desc_upper) + if ks_result['confidence'] > 0.8: + return ks_result + + # 4단계: JIS 규격 확인 + jis_result = check_jis_materials(desc_upper) + if jis_result['confidence'] > 0.8: + return jis_result + + # 5단계: 일반 키워드 확인 + generic_result = check_generic_materials(desc_upper) + + return generic_result + +def check_special_materials(description: str) -> Dict: + """특수 재질 확인""" + + # SUPER ALLOYS 확인 + for alloy_family, alloy_data in SPECIAL_MATERIALS["SUPER_ALLOYS"].items(): + for pattern in alloy_data["patterns"]: + match = re.search(pattern, description) + if match: + grade = match.group(1) if match.groups() else "STANDARD" + grade_info = alloy_data["grades"].get(grade, {}) + + return { + "standard": f"{alloy_family}", + "grade": f"{alloy_family} {grade}", + "material_type": "SUPER_ALLOY", + "manufacturing": alloy_data.get("manufacturing", "SPECIAL"), + "composition": grade_info.get("composition", ""), + "applications": grade_info.get("applications", ""), + "confidence": 0.95, + "evidence": [f"SPECIAL_MATERIAL: {alloy_family} {grade}"] + } + + # TITANIUM 확인 + titanium_data = SPECIAL_MATERIALS["TITANIUM"] + for pattern in titanium_data["patterns"]: + match = re.search(pattern, description) + if match: + grade = match.group(1) if match.groups() else "2" + grade_info = titanium_data["grades"].get(grade, {}) + + return { + "standard": "TITANIUM", + "grade": f"Titanium Grade {grade}", + "material_type": "TITANIUM", + "manufacturing": "FORGED_OR_SEAMLESS", + "composition": grade_info.get("composition", f"Ti Grade {grade}"), + "confidence": 0.95, + "evidence": [f"TITANIUM: Grade {grade}"] + } + + return {"confidence": 0.0} + +def check_astm_materials(description: str) -> Dict: + """ASTM/ASME 규격 확인""" + + astm_data = MATERIAL_STANDARDS["ASTM_ASME"] + + # FORGED 등급 확인 + for standard, standard_data in astm_data["FORGED_GRADES"].items(): + result = check_astm_standard(description, standard, standard_data) + if result["confidence"] > 0.8: + return result + + # WELDED 등급 확인 + for standard, standard_data in astm_data["WELDED_GRADES"].items(): + result = check_astm_standard(description, standard, standard_data) + if result["confidence"] > 0.8: + return result + + # CAST 등급 확인 + for standard, standard_data in astm_data["CAST_GRADES"].items(): + result = check_astm_standard(description, standard, standard_data) + if result["confidence"] > 0.8: + return result + + # PIPE 등급 확인 + for standard, standard_data in astm_data["PIPE_GRADES"].items(): + result = check_astm_standard(description, standard, standard_data) + if result["confidence"] > 0.8: + return result + + return {"confidence": 0.0} + +def check_astm_standard(description: str, standard: str, standard_data: Dict) -> Dict: + """개별 ASTM 규격 확인""" + + # 직접 패턴이 있는 경우 (A105 등) + if "patterns" in standard_data: + for pattern in standard_data["patterns"]: + match = re.search(pattern, description) + if match: + return { + "standard": f"ASTM {standard}", + "grade": f"ASTM {standard}", + "material_type": determine_material_type(standard, ""), + "manufacturing": standard_data.get("manufacturing", "UNKNOWN"), + "confidence": 0.9, + "evidence": [f"ASTM_{standard}: Direct Match"] + } + + # 하위 분류가 있는 경우 (A182, A234 등) + else: + for subtype, subtype_data in standard_data.items(): + for pattern in subtype_data["patterns"]: + match = re.search(pattern, description) + if match: + grade_code = match.group(1) if match.groups() else "" + grade_info = subtype_data["grades"].get(grade_code, {}) + + return { + "standard": f"ASTM {standard}", + "grade": f"ASTM {standard} {grade_code}", + "material_type": determine_material_type(standard, grade_code), + "manufacturing": subtype_data.get("manufacturing", "UNKNOWN"), + "composition": grade_info.get("composition", ""), + "applications": grade_info.get("applications", ""), + "confidence": 0.9, + "evidence": [f"ASTM_{standard}: {grade_code}"] + } + + return {"confidence": 0.0} + +def check_ks_materials(description: str) -> Dict: + """KS 규격 확인""" + + ks_data = MATERIAL_STANDARDS["KS"] + + for category, standards in ks_data.items(): + for standard, standard_data in standards.items(): + for pattern in standard_data["patterns"]: + match = re.search(pattern, description) + if match: + return { + "standard": f"KS {standard}", + "grade": f"KS {standard}", + "material_type": determine_material_type_from_description(description), + "manufacturing": standard_data.get("manufacturing", "UNKNOWN"), + "description": standard_data["description"], + "confidence": 0.85, + "evidence": [f"KS_{standard}"] + } + + return {"confidence": 0.0} + +def check_jis_materials(description: str) -> Dict: + """JIS 규격 확인""" + + jis_data = MATERIAL_STANDARDS["JIS"] + + for category, standards in jis_data.items(): + for standard, standard_data in standards.items(): + for pattern in standard_data["patterns"]: + match = re.search(pattern, description) + if match: + return { + "standard": f"JIS {standard}", + "grade": f"JIS {standard}", + "material_type": determine_material_type_from_description(description), + "manufacturing": standard_data.get("manufacturing", "UNKNOWN"), + "description": standard_data["description"], + "confidence": 0.85, + "evidence": [f"JIS_{standard}"] + } + + return {"confidence": 0.0} + +def check_generic_materials(description: str) -> Dict: + """일반 재질 키워드 확인""" + + for material_type, keywords in GENERIC_MATERIAL_KEYWORDS.items(): + for keyword in keywords: + if keyword in description: + return { + "standard": "GENERIC", + "grade": keyword, + "material_type": material_type, + "manufacturing": "UNKNOWN", + "confidence": 0.6, + "evidence": [f"GENERIC: {keyword}"] + } + + return { + "standard": "UNKNOWN", + "grade": "UNKNOWN", + "material_type": "UNKNOWN", + "manufacturing": "UNKNOWN", + "confidence": 0.0, + "evidence": ["NO_MATERIAL_FOUND"] + } + +def determine_material_type(standard: str, grade: str) -> str: + """규격과 등급으로 재질 타입 결정""" + + # 스테인리스 등급 + stainless_patterns = ["304", "316", "321", "347", "F304", "F316", "WP304", "CF8"] + if any(pattern in grade for pattern in stainless_patterns): + return "STAINLESS_STEEL" + + # 합금강 등급 + alloy_patterns = ["F1", "F5", "F11", "F22", "F91", "WP1", "WP5", "WP11", "WP22", "WP91"] + if any(pattern in grade for pattern in alloy_patterns): + return "ALLOY_STEEL" + + # 주조품 + if standard in ["A216", "A351"]: + return "CAST_STEEL" + + # 기본값은 탄소강 + return "CARBON_STEEL" + +def determine_material_type_from_description(description: str) -> str: + """설명에서 재질 타입 추정""" + + desc_upper = description.upper() + + if any(keyword in desc_upper for keyword in ["SS", "STS", "STAINLESS", "304", "316"]): + return "STAINLESS_STEEL" + elif any(keyword in desc_upper for keyword in ["ALLOY", "합금", "CR", "MO"]): + return "ALLOY_STEEL" + elif any(keyword in desc_upper for keyword in ["CAST", "주조"]): + return "CAST_STEEL" + else: + return "CARBON_STEEL" + +def get_manufacturing_method_from_material(material_result: Dict) -> str: + """재질 정보로부터 제작방법 추정""" + + if material_result.get("confidence", 0) < 0.5: + return "UNKNOWN" + + material_standard = material_result.get('standard', '') + + # 직접 매핑 + if 'A182' in material_standard or 'A105' in material_standard: + return 'FORGED' + elif 'A234' in material_standard or 'A403' in material_standard: + return 'WELDED_FABRICATED' + elif 'A216' in material_standard or 'A351' in material_standard: + return 'CAST' + elif 'A106' in material_standard or 'A312' in material_standard: + return 'SEAMLESS' + elif 'A53' in material_standard: + return 'WELDED_OR_SEAMLESS' + + # manufacturing 필드가 있으면 직접 사용 + manufacturing = material_result.get("manufacturing", "UNKNOWN") + if manufacturing != "UNKNOWN": + return manufacturing + + return "UNKNOWN" + +def get_material_confidence_factors(material_result: Dict) -> List[str]: + """재질 분류 신뢰도 영향 요소 반환""" + + factors = [] + confidence = material_result.get("confidence", 0) + + if confidence >= 0.9: + factors.append("HIGH_CONFIDENCE") + elif confidence >= 0.7: + factors.append("MEDIUM_CONFIDENCE") + else: + factors.append("LOW_CONFIDENCE") + + if material_result.get("standard") == "UNKNOWN": + factors.append("NO_STANDARD_FOUND") + + if material_result.get("manufacturing") == "UNKNOWN": + factors.append("MANUFACTURING_UNCLEAR") + + return factors diff --git a/backend/app/services/materials_schema.py b/backend/app/services/materials_schema.py new file mode 100644 index 0000000..2e8d38d --- /dev/null +++ b/backend/app/services/materials_schema.py @@ -0,0 +1,525 @@ +""" +재질 분류를 위한 공통 스키마 +모든 제품군(PIPE, FITTING, FLANGE 등)에서 공통 사용 +""" + +import re +from typing import Dict, List, Optional + +# ========== 미국 ASTM/ASME 규격 ========== +MATERIAL_STANDARDS = { + "ASTM_ASME": { + "FORGED_GRADES": { + "A182": { + "carbon_alloy": { + "patterns": [ + r"ASTM\s+A182\s+(?:GR\s*)?F(\d+)", + r"A182\s+(?:GR\s*)?F(\d+)", + r"ASME\s+SA182\s+(?:GR\s*)?F(\d+)" + ], + "grades": { + "F1": { + "composition": "0.5Mo", + "temp_max": "482°C", + "applications": "중온용" + }, + "F5": { + "composition": "5Cr-0.5Mo", + "temp_max": "649°C", + "applications": "고온용" + }, + "F11": { + "composition": "1.25Cr-0.5Mo", + "temp_max": "593°C", + "applications": "일반 고온용" + }, + "F22": { + "composition": "2.25Cr-1Mo", + "temp_max": "649°C", + "applications": "고온 고압용" + }, + "F91": { + "composition": "9Cr-1Mo-V", + "temp_max": "649°C", + "applications": "초고온용" + } + }, + "manufacturing": "FORGED" + }, + "stainless": { + "patterns": [ + r"ASTM\s+A182\s+F(\d{3}[LH]*)", + r"A182\s+F(\d{3}[LH]*)", + r"ASME\s+SA182\s+F(\d{3}[LH]*)" + ], + "grades": { + "F304": { + "composition": "18Cr-8Ni", + "applications": "일반용", + "corrosion_resistance": "보통" + }, + "F304L": { + "composition": "18Cr-8Ni-저탄소", + "applications": "용접용", + "corrosion_resistance": "보통" + }, + "F316": { + "composition": "18Cr-10Ni-2Mo", + "applications": "내식성", + "corrosion_resistance": "우수" + }, + "F316L": { + "composition": "18Cr-10Ni-2Mo-저탄소", + "applications": "용접+내식성", + "corrosion_resistance": "우수" + }, + "F321": { + "composition": "18Cr-8Ni-Ti", + "applications": "고온안정화", + "stabilizer": "Titanium" + }, + "F347": { + "composition": "18Cr-8Ni-Nb", + "applications": "고온안정화", + "stabilizer": "Niobium" + } + }, + "manufacturing": "FORGED" + } + }, + "A105": { + "patterns": [ + r"ASTM\s+A105(?:\s+(?:GR\s*)?([ABC]))?", + r"A105(?:\s+(?:GR\s*)?([ABC]))?", + r"ASME\s+SA105" + ], + "description": "탄소강 단조품", + "composition": "탄소강", + "applications": "일반 압력용 단조품", + "manufacturing": "FORGED", + "pressure_rating": "150LB ~ 9000LB" + } + }, + + "WELDED_GRADES": { + "A234": { + "carbon": { + "patterns": [ + r"ASTM\s+A234\s+(?:GR\s*)?WP([ABC])", + r"A234\s+(?:GR\s*)?WP([ABC])", + r"ASME\s+SA234\s+(?:GR\s*)?WP([ABC])" + ], + "grades": { + "WPA": { + "yield_strength": "30 ksi", + "applications": "저압용", + "temp_range": "-29°C ~ 400°C" + }, + "WPB": { + "yield_strength": "35 ksi", + "applications": "일반용", + "temp_range": "-29°C ~ 400°C" + }, + "WPC": { + "yield_strength": "40 ksi", + "applications": "고압용", + "temp_range": "-29°C ~ 400°C" + } + }, + "manufacturing": "WELDED_FABRICATED" + }, + "alloy": { + "patterns": [ + r"ASTM\s+A234\s+(?:GR\s*)?WP(\d+)", + r"A234\s+(?:GR\s*)?WP(\d+)", + r"ASME\s+SA234\s+(?:GR\s*)?WP(\d+)" + ], + "grades": { + "WP1": { + "composition": "0.5Mo", + "temp_max": "482°C" + }, + "WP5": { + "composition": "5Cr-0.5Mo", + "temp_max": "649°C" + }, + "WP11": { + "composition": "1.25Cr-0.5Mo", + "temp_max": "593°C" + }, + "WP22": { + "composition": "2.25Cr-1Mo", + "temp_max": "649°C" + }, + "WP91": { + "composition": "9Cr-1Mo-V", + "temp_max": "649°C" + } + }, + "manufacturing": "WELDED_FABRICATED" + } + }, + "A403": { + "stainless": { + "patterns": [ + r"ASTM\s+A403\s+(?:GR\s*)?WP\s*(\d{3}[LH]*)", + r"A403\s+(?:GR\s*)?WP\s*(\d{3}[LH]*)", + r"ASME\s+SA403\s+(?:GR\s*)?WP\s*(\d{3}[LH]*)" + ], + "grades": { + "WP304": { + "base_grade": "304", + "manufacturing": "WELDED", + "applications": "일반 용접용" + }, + "WP304L": { + "base_grade": "304L", + "manufacturing": "WELDED", + "applications": "저탄소 용접용" + }, + "WP316": { + "base_grade": "316", + "manufacturing": "WELDED", + "applications": "내식성 용접용" + }, + "WP316L": { + "base_grade": "316L", + "manufacturing": "WELDED", + "applications": "저탄소 내식성 용접용" + } + }, + "manufacturing": "WELDED_FABRICATED" + } + } + }, + + "CAST_GRADES": { + "A216": { + "patterns": [ + r"ASTM\s+A216\s+(?:GR\s*)?([A-Z]{2,3})", + r"A216\s+(?:GR\s*)?([A-Z]{2,3})", + r"ASME\s+SA216\s+(?:GR\s*)?([A-Z]{2,3})" + ], + "grades": { + "WCA": { + "composition": "저탄소강", + "applications": "일반주조", + "temp_range": "-29°C ~ 425°C" + }, + "WCB": { + "composition": "탄소강", + "applications": "압력용기", + "temp_range": "-29°C ~ 425°C" + }, + "WCC": { + "composition": "중탄소강", + "applications": "고강도용", + "temp_range": "-29°C ~ 425°C" + } + }, + "manufacturing": "CAST" + }, + "A351": { + "patterns": [ + r"ASTM\s+A351\s+(?:GR\s*)?([A-Z0-9]+)", + r"A351\s+(?:GR\s*)?([A-Z0-9]+)", + r"ASME\s+SA351\s+(?:GR\s*)?([A-Z0-9]+)" + ], + "grades": { + "CF8": { + "base_grade": "304", + "manufacturing": "CAST", + "applications": "304 스테인리스 주조" + }, + "CF8M": { + "base_grade": "316", + "manufacturing": "CAST", + "applications": "316 스테인리스 주조" + }, + "CF3": { + "base_grade": "304L", + "manufacturing": "CAST", + "applications": "304L 스테인리스 주조" + }, + "CF3M": { + "base_grade": "316L", + "manufacturing": "CAST", + "applications": "316L 스테인리스 주조" + } + }, + "manufacturing": "CAST" + } + }, + + "PIPE_GRADES": { + "A106": { + "patterns": [ + r"ASTM\s+A106\s+(?:GR\s*)?([ABC])", + r"A106\s+(?:GR\s*)?([ABC])", + r"ASME\s+SA106\s+(?:GR\s*)?([ABC])" + ], + "grades": { + "A": { + "yield_strength": "30 ksi", + "applications": "저압용" + }, + "B": { + "yield_strength": "35 ksi", + "applications": "일반용" + }, + "C": { + "yield_strength": "40 ksi", + "applications": "고압용" + } + }, + "manufacturing": "SEAMLESS" + }, + "A53": { + "patterns": [ + r"ASTM\s+A53\s+(?:GR\s*)?([ABC])", + r"A53\s+(?:GR\s*)?([ABC])" + ], + "grades": { + "A": {"yield_strength": "30 ksi"}, + "B": {"yield_strength": "35 ksi"} + }, + "manufacturing": "WELDED_OR_SEAMLESS" + }, + "A312": { + "patterns": [ + r"ASTM\s+A312\s+TP\s*(\d{3}[LH]*)", + r"A312\s+TP\s*(\d{3}[LH]*)" + ], + "grades": { + "TP304": { + "base_grade": "304", + "manufacturing": "SEAMLESS" + }, + "TP304L": { + "base_grade": "304L", + "manufacturing": "SEAMLESS" + }, + "TP316": { + "base_grade": "316", + "manufacturing": "SEAMLESS" + }, + "TP316L": { + "base_grade": "316L", + "manufacturing": "SEAMLESS" + } + }, + "manufacturing": "SEAMLESS" + } + } + }, + + # ========== 한국 KS 규격 ========== + "KS": { + "PIPE_GRADES": { + "D3507": { + "patterns": [r"KS\s+D\s*3507\s+SPPS\s*(\d+)"], + "description": "배관용 탄소강관", + "manufacturing": "SEAMLESS" + }, + "D3583": { + "patterns": [r"KS\s+D\s*3583\s+STPG\s*(\d+)"], + "description": "압력배관용 탄소강관", + "manufacturing": "SEAMLESS" + }, + "D3576": { + "patterns": [r"KS\s+D\s*3576\s+STS\s*(\d{3}[LH]*)"], + "description": "배관용 스테인리스강관", + "manufacturing": "SEAMLESS" + } + }, + "FITTING_GRADES": { + "D3562": { + "patterns": [r"KS\s+D\s*3562"], + "description": "탄소강 단조 피팅", + "manufacturing": "FORGED" + }, + "D3563": { + "patterns": [r"KS\s+D\s*3563"], + "description": "스테인리스강 단조 피팅", + "manufacturing": "FORGED" + } + } + }, + + # ========== 일본 JIS 규격 ========== + "JIS": { + "PIPE_GRADES": { + "G3452": { + "patterns": [r"JIS\s+G\s*3452\s+SGP"], + "description": "배관용 탄소강관", + "manufacturing": "WELDED" + }, + "G3454": { + "patterns": [r"JIS\s+G\s*3454\s+STPG\s*(\d+)"], + "description": "압력배관용 탄소강관", + "manufacturing": "SEAMLESS" + }, + "G3459": { + "patterns": [r"JIS\s+G\s*3459\s+SUS\s*(\d{3}[LH]*)"], + "description": "배관용 스테인리스강관", + "manufacturing": "SEAMLESS" + } + }, + "FITTING_GRADES": { + "B2311": { + "patterns": [r"JIS\s+B\s*2311"], + "description": "강제 피팅", + "manufacturing": "FORGED" + }, + "B2312": { + "patterns": [r"JIS\s+B\s*2312"], + "description": "스테인리스강 피팅", + "manufacturing": "FORGED" + } + } + } +} + +# ========== 특수 재질 ========== +SPECIAL_MATERIALS = { + "SUPER_ALLOYS": { + "INCONEL": { + "patterns": [r"INCONEL\s*(\d+)"], + "grades": { + "600": { + "composition": "Ni-Cr", + "temp_max": "1177°C", + "applications": "고온 산화 환경" + }, + "625": { + "composition": "Ni-Cr-Mo", + "temp_max": "982°C", + "applications": "고온 부식 환경" + }, + "718": { + "composition": "Ni-Cr-Fe", + "temp_max": "704°C", + "applications": "고온 고강도" + } + }, + "manufacturing": "FORGED_OR_CAST" + }, + "HASTELLOY": { + "patterns": [r"HASTELLOY\s*([A-Z0-9]+)"], + "grades": { + "C276": { + "composition": "Ni-Mo-Cr", + "corrosion": "최고급", + "applications": "강산성 환경" + }, + "C22": { + "composition": "Ni-Cr-Mo-W", + "applications": "화학공정" + } + }, + "manufacturing": "FORGED_OR_CAST" + }, + "MONEL": { + "patterns": [r"MONEL\s*(\d+)"], + "grades": { + "400": { + "composition": "Ni-Cu", + "applications": "해양 환경" + } + } + } + }, + "TITANIUM": { + "patterns": [ + r"TITANIUM(?:\s+(?:GR|GRADE)\s*(\d+))?", + r"Ti(?:\s+(?:GR|GRADE)\s*(\d+))?", + r"ASTM\s+B\d+\s+(?:GR|GRADE)\s*(\d+)" + ], + "grades": { + "1": { + "purity": "상업용 순티타늄", + "strength": "낮음", + "applications": "화학공정" + }, + "2": { + "purity": "상업용 순티타늄 (일반)", + "strength": "보통", + "applications": "일반용" + }, + "5": { + "composition": "Ti-6Al-4V", + "strength": "고강도", + "applications": "항공우주" + } + }, + "manufacturing": "FORGED_OR_SEAMLESS" + }, + "COPPER_ALLOYS": { + "BRASS": { + "patterns": [r"BRASS|황동"], + "composition": "Cu-Zn", + "applications": ["계기용", "선박용"] + }, + "BRONZE": { + "patterns": [r"BRONZE|청동"], + "composition": "Cu-Sn", + "applications": ["해양용", "베어링"] + }, + "CUPRONICKEL": { + "patterns": [r"Cu-?Ni|CUPRONICKEL"], + "composition": "Cu-Ni", + "applications": ["해수 배관"] + } + } +} + +# ========== 제작방법별 재질 매핑 ========== +MANUFACTURING_MATERIAL_MAP = { + "FORGED": { + "primary_standards": ["ASTM A182", "ASTM A105"], + "size_range": "1/8\" ~ 4\"", + "pressure_range": "150LB ~ 9000LB", + "characteristics": "고강도, 고압용" + }, + "WELDED_FABRICATED": { + "primary_standards": ["ASTM A234", "ASTM A403"], + "size_range": "1/2\" ~ 48\"", + "pressure_range": "150LB ~ 2500LB", + "characteristics": "대구경, 중저압용" + }, + "CAST": { + "primary_standards": ["ASTM A216", "ASTM A351"], + "size_range": "1/2\" ~ 24\"", + "applications": ["복잡형상", "밸브본체"], + "characteristics": "복잡 형상 가능" + }, + "SEAMLESS": { + "primary_standards": ["ASTM A106", "ASTM A312", "ASTM A335"], + "manufacturing": "열간압연/냉간인발", + "characteristics": "이음새 없는 관" + }, + "WELDED": { + "primary_standards": ["ASTM A53", "ASTM A312"], + "manufacturing": "전기저항용접/아크용접", + "characteristics": "용접선 있는 관" + } +} + +# ========== 일반 재질 키워드 ========== +GENERIC_MATERIAL_KEYWORDS = { + "CARBON_STEEL": [ + "CS", "CARBON STEEL", "탄소강", "SS400", "SM490" + ], + "STAINLESS_STEEL": [ + "SS", "STS", "STAINLESS", "스테인리스", "304", "316", "321", "347" + ], + "ALLOY_STEEL": [ + "ALLOY", "합금강", "CHROME MOLY", "Cr-Mo" + ], + "CAST_IRON": [ + "CAST IRON", "주철", "FC", "FCD" + ], + "DUCTILE_IRON": [ + "DUCTILE IRON", "구상흑연주철", "덕타일" + ] +} diff --git a/backend/app/services/pipe_classifier.py b/backend/app/services/pipe_classifier.py new file mode 100644 index 0000000..2a96639 --- /dev/null +++ b/backend/app/services/pipe_classifier.py @@ -0,0 +1,337 @@ +""" +PIPE 분류 전용 모듈 +재질 분류 + 파이프 특화 분류 +""" + +import re +from typing import Dict, List, Optional +from .material_classifier import classify_material, get_manufacturing_method_from_material + +# ========== PIPE 제조 방법별 분류 ========== +PIPE_MANUFACTURING = { + "SEAMLESS": { + "keywords": ["SEAMLESS", "SMLS", "심리스", "무계목"], + "confidence": 0.95, + "characteristics": "이음새 없는 고품질 파이프" + }, + "WELDED": { + "keywords": ["WELDED", "WLD", "ERW", "SAW", "용접", "전기저항용접"], + "confidence": 0.95, + "characteristics": "용접으로 제조된 파이프" + }, + "CAST": { + "keywords": ["CAST", "주조"], + "confidence": 0.85, + "characteristics": "주조로 제조된 파이프" + } +} + +# ========== PIPE 끝 가공별 분류 ========== +PIPE_END_PREP = { + "BOTH_ENDS_BEVELED": { + "codes": ["BOE", "BOTH END", "BOTH BEVELED", "양쪽개선"], + "cutting_note": "양쪽 개선", + "machining_required": True, + "confidence": 0.95 + }, + "ONE_END_BEVELED": { + "codes": ["BE", "BEV", "PBE", "PIPE BEVELED END"], + "cutting_note": "한쪽 개선", + "machining_required": True, + "confidence": 0.95 + }, + "NO_BEVEL": { + "codes": ["PE", "PLAIN END", "PPE", "평단", "무개선"], + "cutting_note": "무 개선", + "machining_required": False, + "confidence": 0.95 + } +} + +# ========== PIPE 스케줄별 분류 ========== +PIPE_SCHEDULE = { + "patterns": [ + r"SCH\s*(\d+)", + r"SCHEDULE\s*(\d+)", + r"스케줄\s*(\d+)" + ], + "common_schedules": ["10", "20", "40", "80", "120", "160"], + "wall_thickness_patterns": [ + r"(\d+(?:\.\d+)?)\s*mm\s*THK", + r"(\d+(?:\.\d+)?)\s*THK" + ] +} + +def classify_pipe(dat_file: str, description: str, main_nom: str, + length: float = None) -> Dict: + """ + 완전한 PIPE 분류 + + Args: + dat_file: DAT_FILE 필드 + description: DESCRIPTION 필드 + main_nom: MAIN_NOM 필드 (사이즈) + length: LENGTH 필드 (절단 치수) + + Returns: + 완전한 파이프 분류 결과 + """ + + # 1. 재질 분류 (공통 모듈 사용) + material_result = classify_material(description) + + # 2. 제조 방법 분류 + manufacturing_result = classify_pipe_manufacturing(description, material_result) + + # 3. 끝 가공 분류 + end_prep_result = classify_pipe_end_preparation(description) + + # 4. 스케줄 분류 + schedule_result = classify_pipe_schedule(description) + + # 5. 절단 치수 처리 + cutting_dimensions = extract_pipe_cutting_dimensions(length, description) + + # 6. 최종 결과 조합 + return { + "category": "PIPE", + + # 재질 정보 (공통 모듈) + "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) + }, + + # 파이프 특화 정보 + "manufacturing": { + "method": manufacturing_result.get('method', 'UNKNOWN'), + "confidence": manufacturing_result.get('confidence', 0.0), + "evidence": manufacturing_result.get('evidence', []) + }, + + "end_preparation": { + "type": end_prep_result.get('type', 'UNKNOWN'), + "cutting_note": end_prep_result.get('cutting_note', ''), + "machining_required": end_prep_result.get('machining_required', False), + "confidence": end_prep_result.get('confidence', 0.0) + }, + + "schedule": { + "schedule": schedule_result.get('schedule', 'UNKNOWN'), + "wall_thickness": schedule_result.get('wall_thickness', ''), + "confidence": schedule_result.get('confidence', 0.0) + }, + + "cutting_dimensions": cutting_dimensions, + + "size_info": { + "nominal_size": main_nom, + "length_mm": cutting_dimensions.get('length_mm') + }, + + # 전체 신뢰도 + "overall_confidence": calculate_pipe_confidence({ + "material": material_result.get('confidence', 0), + "manufacturing": manufacturing_result.get('confidence', 0), + "end_prep": end_prep_result.get('confidence', 0), + "schedule": schedule_result.get('confidence', 0) + }) + } + +def classify_pipe_manufacturing(description: str, material_result: Dict) -> Dict: + """파이프 제조 방법 분류""" + + desc_upper = description.upper() + + # 1. DESCRIPTION 키워드 우선 확인 + for method, method_data in PIPE_MANUFACTURING.items(): + for keyword in method_data["keywords"]: + if keyword in desc_upper: + return { + "method": method, + "confidence": method_data["confidence"], + "evidence": [f"KEYWORD: {keyword}"], + "characteristics": method_data["characteristics"] + } + + # 2. 재질 규격으로 추정 + material_manufacturing = get_manufacturing_method_from_material(material_result) + if material_manufacturing in ["SEAMLESS", "WELDED"]: + return { + "method": material_manufacturing, + "confidence": 0.8, + "evidence": [f"MATERIAL_STANDARD: {material_result.get('standard')}"], + "characteristics": PIPE_MANUFACTURING.get(material_manufacturing, {}).get("characteristics", "") + } + + # 3. 기본값 + return { + "method": "UNKNOWN", + "confidence": 0.0, + "evidence": ["NO_MANUFACTURING_INFO"] + } + +def classify_pipe_end_preparation(description: str) -> Dict: + """파이프 끝 가공 분류""" + + desc_upper = description.upper() + + # 우선순위: 양쪽 > 한쪽 > 무개선 + for prep_type, prep_data in PIPE_END_PREP.items(): + for code in prep_data["codes"]: + if code in desc_upper: + return { + "type": prep_type, + "cutting_note": prep_data["cutting_note"], + "machining_required": prep_data["machining_required"], + "confidence": prep_data["confidence"], + "matched_code": code + } + + # 기본값: 무개선 + return { + "type": "NO_BEVEL", + "cutting_note": "무 개선 (기본값)", + "machining_required": False, + "confidence": 0.5, + "matched_code": "DEFAULT" + } + +def classify_pipe_schedule(description: str) -> Dict: + """파이프 스케줄 분류""" + + desc_upper = description.upper() + + # 1. 스케줄 패턴 확인 + for pattern in PIPE_SCHEDULE["patterns"]: + match = re.search(pattern, desc_upper) + if match: + schedule_num = match.group(1) + return { + "schedule": f"SCH {schedule_num}", + "schedule_number": schedule_num, + "confidence": 0.95, + "matched_pattern": pattern + } + + # 2. 두께 패턴 확인 + for pattern in PIPE_SCHEDULE["wall_thickness_patterns"]: + match = re.search(pattern, desc_upper) + if match: + thickness = match.group(1) + return { + "schedule": f"{thickness}mm THK", + "wall_thickness": f"{thickness}mm", + "confidence": 0.9, + "matched_pattern": pattern + } + + # 3. 기본값 + return { + "schedule": "UNKNOWN", + "confidence": 0.0 + } + +def extract_pipe_cutting_dimensions(length: float, description: str) -> Dict: + """파이프 절단 치수 정보 추출""" + + cutting_info = { + "length_mm": None, + "source": None, + "confidence": 0.0, + "note": "" + } + + # 1. LENGTH 필드에서 추출 (우선) + if length and length > 0: + cutting_info.update({ + "length_mm": round(length, 1), + "source": "LENGTH_FIELD", + "confidence": 0.95, + "note": f"도면 명기 치수: {length}mm" + }) + + # 2. DESCRIPTION에서 백업 추출 + else: + desc_length = extract_length_from_description(description) + if desc_length: + cutting_info.update({ + "length_mm": desc_length, + "source": "DESCRIPTION_PARSED", + "confidence": 0.8, + "note": f"설명란에서 추출: {desc_length}mm" + }) + else: + cutting_info.update({ + "source": "NO_LENGTH_INFO", + "confidence": 0.0, + "note": "절단 치수 정보 없음 - 도면 확인 필요" + }) + + return cutting_info + +def extract_length_from_description(description: str) -> Optional[float]: + """DESCRIPTION에서 길이 정보 추출""" + + length_patterns = [ + r'(\d+(?:\.\d+)?)\s*mm', + r'(\d+(?:\.\d+)?)\s*M', + r'(\d+(?:\.\d+)?)\s*LG', + r'L\s*=\s*(\d+(?:\.\d+)?)', + r'길이\s*(\d+(?:\.\d+)?)' + ] + + for pattern in length_patterns: + match = re.search(pattern, description.upper()) + if match: + return float(match.group(1)) + + return None + +def calculate_pipe_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.4, + "manufacturing": 0.25, + "end_prep": 0.2, + "schedule": 0.15 + } + + weighted_sum = sum( + confidence_scores.get(key, 0) * weight + for key, weight in weights.items() + ) + + return round(weighted_sum, 2) + +def generate_pipe_cutting_plan(pipe_data: Dict) -> Dict: + """파이프 절단 계획 생성""" + + cutting_plan = { + "material_spec": f"{pipe_data['material']['grade']} {pipe_data['schedule']['schedule']}", + "length_mm": pipe_data['cutting_dimensions']['length_mm'], + "end_preparation": pipe_data['end_preparation']['cutting_note'], + "machining_required": pipe_data['end_preparation']['machining_required'] + } + + # 절단 지시서 생성 + if cutting_plan["length_mm"]: + cutting_plan["cutting_instruction"] = f""" +재질: {cutting_plan['material_spec']} +절단길이: {cutting_plan['length_mm']}mm +끝가공: {cutting_plan['end_preparation']} +가공여부: {'베벨가공 필요' if cutting_plan['machining_required'] else '직각절단만'} + """.strip() + else: + cutting_plan["cutting_instruction"] = "도면 확인 후 절단 치수 입력 필요" + + return cutting_plan diff --git a/backend/app/services/spool_manager.py b/backend/app/services/spool_manager.py new file mode 100644 index 0000000..8289a71 --- /dev/null +++ b/backend/app/services/spool_manager.py @@ -0,0 +1,256 @@ +""" +스풀 관리 시스템 +도면별 에리어 넘버와 스풀 넘버 관리 +""" + +import re +from typing import Dict, List, Optional, Tuple +from datetime import datetime + +# ========== 스풀 넘버링 규칙 ========== +SPOOL_NUMBERING_RULES = { + "AREA_NUMBER": { + "pattern": r"#(\d{2})", # #01, #02, #03... + "format": "#{:02d}", # 2자리 숫자 + "range": (1, 99), # 01~99 + "description": "에리어 넘버" + }, + "SPOOL_NUMBER": { + "pattern": r"([A-Z]{1,2})", # A, B, C... AA, AB... + "format": "{}", + "sequence": ["A", "B", "C", "D", "E", "F", "G", "H", "I", "J", + "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", + "U", "V", "W", "X", "Y", "Z", + "AA", "AB", "AC", "AD", "AE", "AF", "AG", "AH", "AI", "AJ"], + "description": "스풀 넘버" + } +} + +# ========== 프로젝트별 넘버링 설정 ========== +PROJECT_SPOOL_SETTINGS = { + "DEFAULT": { + "area_prefix": "#", + "area_digits": 2, + "spool_sequence": "ALPHABETIC", # A, B, C... + "separator": "-", + "format": "{dwg_name}-{area}-{spool}" # 1-IAR-3B1D0-0129-N-#01-A + }, + "CUSTOM": { + # 프로젝트별 커스텀 설정 가능 + } +} + +class SpoolManager: + """스풀 관리 클래스""" + + def __init__(self, project_id: int = None): + self.project_id = project_id + self.settings = PROJECT_SPOOL_SETTINGS["DEFAULT"] + + def parse_spool_identifier(self, spool_id: str) -> Dict: + """기존 스풀 식별자 파싱""" + + # 패턴: DWG_NAME-AREA-SPOOL + # 예: 1-IAR-3B1D0-0129-N-#01-A + + parts = spool_id.split("-") + + area_number = None + spool_number = None + dwg_base = None + + for i, part in enumerate(parts): + # 에리어 넘버 찾기 (#01, #02...) + if re.match(r"#\d{2}", part): + area_number = part + # 스풀 넘버는 다음 파트 + if i + 1 < len(parts): + spool_number = parts[i + 1] + # 도면명은 에리어 넘버 앞까지 + dwg_base = "-".join(parts[:i]) + break + + return { + "original_id": spool_id, + "dwg_base": dwg_base, + "area_number": area_number, + "spool_number": spool_number, + "is_valid": bool(area_number and spool_number), + "parsed_parts": parts + } + + def generate_spool_identifier(self, dwg_name: str, area_number: str, + spool_number: str) -> str: + """새로운 스풀 식별자 생성""" + + # 에리어 넘버 포맷 검증 + area_formatted = self.format_area_number(area_number) + + # 스풀 넘버 포맷 검증 + spool_formatted = self.format_spool_number(spool_number) + + # 조합 + return f"{dwg_name}-{area_formatted}-{spool_formatted}" + + def format_area_number(self, area_input: str) -> str: + """에리어 넘버 포맷팅""" + + # 숫자만 추출 + numbers = re.findall(r'\d+', area_input) + if numbers: + area_num = int(numbers[0]) + if 1 <= area_num <= 99: + return f"#{area_num:02d}" + + raise ValueError(f"유효하지 않은 에리어 넘버: {area_input}") + + def format_spool_number(self, spool_input: str) -> str: + """스풀 넘버 포맷팅""" + + spool_clean = spool_input.upper().strip() + + # 유효한 스풀 넘버인지 확인 + if re.match(r'^[A-Z]{1,2}$', spool_clean): + return spool_clean + + raise ValueError(f"유효하지 않은 스풀 넘버: {spool_input}") + + def get_next_spool_number(self, dwg_name: str, area_number: str, + existing_spools: List[str] = None) -> str: + """다음 사용 가능한 스풀 넘버 추천""" + + if not existing_spools: + return "A" # 첫 번째 스풀 + + # 해당 도면+에리어의 기존 스풀들 파싱 + used_spools = set() + area_formatted = self.format_area_number(area_number) + + for spool_id in existing_spools: + parsed = self.parse_spool_identifier(spool_id) + if (parsed["dwg_base"] == dwg_name and + parsed["area_number"] == area_formatted): + used_spools.add(parsed["spool_number"]) + + # 다음 사용 가능한 넘버 찾기 + sequence = SPOOL_NUMBERING_RULES["SPOOL_NUMBER"]["sequence"] + for spool_num in sequence: + if spool_num not in used_spools: + return spool_num + + raise ValueError("사용 가능한 스풀 넘버가 없습니다") + + def validate_spool_identifier(self, spool_id: str) -> Dict: + """스풀 식별자 유효성 검증""" + + parsed = self.parse_spool_identifier(spool_id) + + validation_result = { + "is_valid": True, + "errors": [], + "warnings": [], + "parsed": parsed + } + + # 도면명 확인 + if not parsed["dwg_base"]: + validation_result["is_valid"] = False + validation_result["errors"].append("도면명이 없습니다") + + # 에리어 넘버 확인 + if not parsed["area_number"]: + validation_result["is_valid"] = False + validation_result["errors"].append("에리어 넘버가 없습니다") + elif not re.match(r"#\d{2}", parsed["area_number"]): + validation_result["is_valid"] = False + validation_result["errors"].append("에리어 넘버 형식이 잘못되었습니다 (#01 형태)") + + # 스풀 넘버 확인 + if not parsed["spool_number"]: + validation_result["is_valid"] = False + validation_result["errors"].append("스풀 넘버가 없습니다") + elif not re.match(r"^[A-Z]{1,2}$", parsed["spool_number"]): + validation_result["is_valid"] = False + validation_result["errors"].append("스풀 넘버 형식이 잘못되었습니다 (A, B, AA 형태)") + + return validation_result + +def classify_pipe_with_spool(dat_file: str, description: str, main_nom: str, + length: float = None, dwg_name: str = None, + area_number: str = None, spool_number: str = None) -> Dict: + """파이프 분류 + 스풀 정보 통합""" + + # 기본 파이프 분류 (기존 함수 사용) + from .pipe_classifier import classify_pipe + pipe_result = classify_pipe(dat_file, description, main_nom, length) + + # 스풀 관리자 생성 + spool_manager = SpoolManager() + + # 스풀 정보 추가 + spool_info = { + "dwg_name": dwg_name, + "area_number": None, + "spool_number": None, + "spool_identifier": None, + "manual_input_required": True, + "validation": None + } + + # 에리어/스풀 넘버가 제공된 경우 + if area_number and spool_number: + try: + area_formatted = spool_manager.format_area_number(area_number) + spool_formatted = spool_manager.format_spool_number(spool_number) + spool_identifier = spool_manager.generate_spool_identifier( + dwg_name, area_formatted, spool_formatted + ) + + spool_info.update({ + "area_number": area_formatted, + "spool_number": spool_formatted, + "spool_identifier": spool_identifier, + "manual_input_required": False, + "validation": {"is_valid": True, "errors": []} + }) + + except ValueError as e: + spool_info["validation"] = { + "is_valid": False, + "errors": [str(e)] + } + + # 기존 결과에 스풀 정보 추가 + pipe_result["spool_info"] = spool_info + + return pipe_result + +def group_pipes_by_spool(pipes_data: List[Dict]) -> Dict: + """파이프들을 스풀별로 그룹핑""" + + spool_groups = {} + + for pipe in pipes_data: + spool_info = pipe.get("spool_info", {}) + spool_id = spool_info.get("spool_identifier", "UNGROUPED") + + if spool_id not in spool_groups: + spool_groups[spool_id] = { + "spool_identifier": spool_id, + "dwg_name": spool_info.get("dwg_name"), + "area_number": spool_info.get("area_number"), + "spool_number": spool_info.get("spool_number"), + "pipes": [], + "total_length": 0, + "pipe_count": 0 + } + + spool_groups[spool_id]["pipes"].append(pipe) + spool_groups[spool_id]["pipe_count"] += 1 + + # 길이 합계 + pipe_length = pipe.get("cutting_dimensions", {}).get("length_mm", 0) + if pipe_length: + spool_groups[spool_id]["total_length"] += pipe_length + + return spool_groups diff --git a/backend/app/services/spool_manager_v2.py b/backend/app/services/spool_manager_v2.py new file mode 100644 index 0000000..2d33836 --- /dev/null +++ b/backend/app/services/spool_manager_v2.py @@ -0,0 +1,229 @@ +""" +수정된 스풀 관리 시스템 +도면별 스풀 넘버링 + 에리어는 별도 관리 +""" + +import re +from typing import Dict, List, Optional, Tuple +from datetime import datetime + +# ========== 스풀 넘버링 규칙 ========== +SPOOL_NUMBERING_RULES = { + "SPOOL_NUMBER": { + "sequence": ["A", "B", "C", "D", "E", "F", "G", "H", "I", "J", + "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", + "U", "V", "W", "X", "Y", "Z"], + "description": "도면별 스풀 넘버" + }, + "AREA_NUMBER": { + "pattern": r"#(\d{2})", # #01, #02, #03... + "format": "#{:02d}", # 2자리 숫자 + "range": (1, 99), # 01~99 + "description": "물리적 구역 넘버 (별도 관리)" + } +} + +class SpoolManagerV2: + """수정된 스풀 관리 클래스""" + + def __init__(self, project_id: int = None): + self.project_id = project_id + + def generate_spool_identifier(self, dwg_name: str, spool_number: str) -> str: + """ + 스풀 식별자 생성 (도면명 + 스풀넘버) + + Args: + dwg_name: 도면명 (예: "A-1", "B-3") + spool_number: 스풀넘버 (예: "A", "B") + + Returns: + 스풀 식별자 (예: "A-1-A", "B-3-B") + """ + + # 스풀 넘버 포맷 검증 + spool_formatted = self.format_spool_number(spool_number) + + # 조합: {도면명}-{스풀넘버} + return f"{dwg_name}-{spool_formatted}" + + def parse_spool_identifier(self, spool_id: str) -> Dict: + """스풀 식별자 파싱""" + + # 패턴: DWG_NAME-SPOOL_NUMBER + # 예: A-1-A, B-3-B, 1-IAR-3B1D0-0129-N-A + + # 마지막 '-' 기준으로 분리 + parts = spool_id.rsplit('-', 1) + + if len(parts) == 2: + dwg_base = parts[0] + spool_number = parts[1] + + return { + "original_id": spool_id, + "dwg_name": dwg_base, + "spool_number": spool_number, + "is_valid": self.validate_spool_number(spool_number), + "format": "CORRECT" + } + else: + return { + "original_id": spool_id, + "dwg_name": None, + "spool_number": None, + "is_valid": False, + "format": "INVALID" + } + + def format_spool_number(self, spool_input: str) -> str: + """스풀 넘버 포맷팅 및 검증""" + + spool_clean = spool_input.upper().strip() + + # 유효한 스풀 넘버인지 확인 (A-Z 단일 문자) + if re.match(r'^[A-Z]$', spool_clean): + return spool_clean + + raise ValueError(f"유효하지 않은 스풀 넘버: {spool_input} (A-Z 단일 문자만 가능)") + + def validate_spool_number(self, spool_number: str) -> bool: + """스풀 넘버 유효성 검증""" + return bool(re.match(r'^[A-Z]$', spool_number)) + + def get_next_spool_number(self, dwg_name: str, existing_spools: List[str] = None) -> str: + """해당 도면의 다음 사용 가능한 스풀 넘버 추천""" + + if not existing_spools: + return "A" # 첫 번째 스풀 + + # 해당 도면의 기존 스풀들 파싱 + used_spools = set() + + for spool_id in existing_spools: + parsed = self.parse_spool_identifier(spool_id) + if parsed["dwg_name"] == dwg_name and parsed["is_valid"]: + used_spools.add(parsed["spool_number"]) + + # 다음 사용 가능한 넘버 찾기 + sequence = SPOOL_NUMBERING_RULES["SPOOL_NUMBER"]["sequence"] + for spool_num in sequence: + if spool_num not in used_spools: + return spool_num + + raise ValueError(f"도면 {dwg_name}에서 사용 가능한 스풀 넘버가 없습니다") + + def validate_spool_identifier(self, spool_id: str) -> Dict: + """스풀 식별자 전체 유효성 검증""" + + parsed = self.parse_spool_identifier(spool_id) + + validation_result = { + "is_valid": True, + "errors": [], + "warnings": [], + "parsed": parsed + } + + # 도면명 확인 + if not parsed["dwg_name"]: + validation_result["is_valid"] = False + validation_result["errors"].append("도면명이 없습니다") + + # 스풀 넘버 확인 + if not parsed["spool_number"]: + validation_result["is_valid"] = False + validation_result["errors"].append("스풀 넘버가 없습니다") + elif not self.validate_spool_number(parsed["spool_number"]): + validation_result["is_valid"] = False + validation_result["errors"].append("스풀 넘버 형식이 잘못되었습니다 (A-Z 단일 문자)") + + return validation_result + +# ========== 에리어 관리 (별도 시스템) ========== +class AreaManager: + """에리어 관리 클래스 (물리적 구역)""" + + def __init__(self, project_id: int = None): + self.project_id = project_id + + def format_area_number(self, area_input: str) -> str: + """에리어 넘버 포맷팅""" + + # 숫자만 추출 + numbers = re.findall(r'\d+', area_input) + if numbers: + area_num = int(numbers[0]) + if 1 <= area_num <= 99: + return f"#{area_num:02d}" + + raise ValueError(f"유효하지 않은 에리어 넘버: {area_input} (#01-#99)") + + def assign_drawings_to_area(self, area_number: str, drawing_names: List[str]) -> Dict: + """도면들을 에리어에 할당""" + + area_formatted = self.format_area_number(area_number) + + return { + "area_number": area_formatted, + "assigned_drawings": drawing_names, + "assignment_count": len(drawing_names), + "assignment_date": datetime.now().isoformat() + } + +def classify_pipe_with_corrected_spool(dat_file: str, description: str, main_nom: str, + length: float = None, dwg_name: str = None, + spool_number: str = None, area_number: str = None) -> Dict: + """파이프 분류 + 수정된 스풀 정보""" + + # 기본 파이프 분류 + from .pipe_classifier import classify_pipe + pipe_result = classify_pipe(dat_file, description, main_nom, length) + + # 스풀 관리자 생성 + spool_manager = SpoolManagerV2() + area_manager = AreaManager() + + # 스풀 정보 처리 + spool_info = { + "dwg_name": dwg_name, + "spool_number": None, + "spool_identifier": None, + "area_number": None, # 별도 관리 + "manual_input_required": True, + "validation": None + } + + # 스풀 넘버가 제공된 경우 + if dwg_name and spool_number: + try: + spool_identifier = spool_manager.generate_spool_identifier(dwg_name, spool_number) + + spool_info.update({ + "spool_number": spool_manager.format_spool_number(spool_number), + "spool_identifier": spool_identifier, + "manual_input_required": False, + "validation": {"is_valid": True, "errors": []} + }) + + except ValueError as e: + spool_info["validation"] = { + "is_valid": False, + "errors": [str(e)] + } + + # 에리어 정보 처리 (별도) + if area_number: + try: + area_formatted = area_manager.format_area_number(area_number) + spool_info["area_number"] = area_formatted + except ValueError as e: + spool_info["area_validation"] = { + "is_valid": False, + "errors": [str(e)] + } + + # 기존 결과에 스풀 정보 추가 + pipe_result["spool_info"] = spool_info + + return pipe_result diff --git a/backend/app/services/test_fitting_classifier.py b/backend/app/services/test_fitting_classifier.py new file mode 100644 index 0000000..768255b --- /dev/null +++ b/backend/app/services/test_fitting_classifier.py @@ -0,0 +1,184 @@ +""" +FITTING 분류 시스템 V2 테스트 (크기 정보 추가) +""" + +from .fitting_classifier import classify_fitting, get_fitting_purchase_info + +def test_fitting_classification(): + """실제 BOM 데이터로 FITTING 분류 테스트""" + + test_cases = [ + { + "name": "90도 엘보 (BW)", + "dat_file": "90L_BW", + "description": "90 LR ELL, SCH 40, ASTM A234 GR WPB, SMLS", + "main_nom": "3\"", + "red_nom": None + }, + { + "name": "리듀싱 티 (BW)", + "dat_file": "TEE_RD_BW", + "description": "TEE RED, SCH 40 x SCH 40, ASTM A234 GR WPB, SMLS", + "main_nom": "4\"", + "red_nom": "2\"" + }, + { + "name": "동심 리듀서 (BW)", + "dat_file": "CNC_BW", + "description": "RED CONC, SCH 40 x SCH 40, ASTM A234 GR WPB, SMLS", + "main_nom": "3\"", + "red_nom": "2\"" + }, + { + "name": "편심 리듀서 (BW)", + "dat_file": "ECC_BW", + "description": "RED ECC, SCH 40 x SCH 40, ASTM A234 GR WPB, SMLS", + "main_nom": "6\"", + "red_nom": "3\"" + }, + { + "name": "소켓웰드 티 (고압)", + "dat_file": "TEE_SW_3000", + "description": "TEE, SW, 3000LB, ASTM A105", + "main_nom": "1\"", + "red_nom": None + }, + { + "name": "리듀싱 소켓웰드 티 (고압)", + "dat_file": "TEE_RD_SW_3000", + "description": "TEE RED, SW, 3000LB, ASTM A105", + "main_nom": "1\"", + "red_nom": "1/2\"" + }, + { + "name": "소켓웰드 캡 (고압)", + "dat_file": "CAP_SW_3000", + "description": "CAP, NPT(F), 3000LB, ASTM A105", + "main_nom": "1\"", + "red_nom": None + }, + { + "name": "소켓오렛 (고압)", + "dat_file": "SOL_SW_3000", + "description": "SOCK-O-LET, SW, 3000LB, ASTM A105", + "main_nom": "3\"", + "red_nom": "1\"" + }, + { + "name": "동심 스웨지 (BW)", + "dat_file": "SWG_CN_BW", + "description": "SWAGE CONC, SCH 40 x SCH 80, ASTM A105 GR , SMLS BBE", + "main_nom": "2\"", + "red_nom": "1\"" + }, + { + "name": "편심 스웨지 (BW)", + "dat_file": "SWG_EC_BW", + "description": "SWAGE ECC, SCH 80 x SCH 80, ASTM A105 GR , SMLS PBE", + "main_nom": "1\"", + "red_nom": "3/4\"" + } + ] + + print("🔧 FITTING 분류 시스템 V2 테스트 시작\n") + print("=" * 80) + + for i, test in enumerate(test_cases, 1): + print(f"\n테스트 {i}: {test['name']}") + print("-" * 60) + + result = classify_fitting( + test["dat_file"], + test["description"], + test["main_nom"], + test["red_nom"] + ) + + purchase_info = get_fitting_purchase_info(result) + + print(f"📋 입력:") + print(f" DAT_FILE: {test['dat_file']}") + print(f" DESCRIPTION: {test['description']}") + print(f" SIZE: {result['size_info']['size_description']}") + + print(f"\n🔧 분류 결과:") + print(f" 재질: {result['material']['standard']} | {result['material']['grade']}") + print(f" 피팅타입: {result['fitting_type']['type']} - {result['fitting_type']['subtype']}") + print(f" 크기정보: {result['size_info']['size_description']}") # ← 추가! + if result['size_info']['reduced_size']: + print(f" 주사이즈: {result['size_info']['main_size']}") + print(f" 축소사이즈: {result['size_info']['reduced_size']}") + print(f" 연결방식: {result['connection_method']['method']}") + print(f" 압력등급: {result['pressure_rating']['rating']} ({result['pressure_rating']['common_use']})") + print(f" 제작방법: {result['manufacturing']['method']} ({result['manufacturing']['characteristics']})") + + print(f"\n📊 신뢰도:") + print(f" 전체신뢰도: {result['overall_confidence']}") + print(f" 재질: {result['material']['confidence']}") + print(f" 피팅타입: {result['fitting_type']['confidence']}") + print(f" 연결방식: {result['connection_method']['confidence']}") + print(f" 압력등급: {result['pressure_rating']['confidence']}") + + print(f"\n🛒 구매 정보:") + print(f" 공급업체: {purchase_info['supplier_type']}") + print(f" 예상납기: {purchase_info['lead_time_estimate']}") + print(f" 구매카테고리: {purchase_info['purchase_category']}") + + # 크기 정보 저장 확인 + print(f"\n💾 저장될 데이터:") + print(f" MAIN_NOM: {result['size_info']['main_size']}") + print(f" RED_NOM: {result['size_info']['reduced_size'] or 'NULL'}") + print(f" SIZE_DESCRIPTION: {result['size_info']['size_description']}") + + if i < len(test_cases): + print("\n" + "=" * 80) + +def test_fitting_edge_cases(): + """예외 케이스 테스트""" + + edge_cases = [ + { + "name": "DAT_FILE만 있는 경우", + "dat_file": "90L_BW", + "description": "", + "main_nom": "2\"", + "red_nom": None + }, + { + "name": "DESCRIPTION만 있는 경우", + "dat_file": "", + "description": "90 DEGREE ELBOW, ASTM A234 WPB", + "main_nom": "3\"", + "red_nom": None + }, + { + "name": "알 수 없는 DAT_FILE", + "dat_file": "UNKNOWN_CODE", + "description": "SPECIAL FITTING, CUSTOM MADE", + "main_nom": "4\"", + "red_nom": None + } + ] + + print("\n🧪 예외 케이스 테스트\n") + print("=" * 50) + + for i, test in enumerate(edge_cases, 1): + print(f"\n예외 테스트 {i}: {test['name']}") + print("-" * 40) + + result = classify_fitting( + test["dat_file"], + test["description"], + test["main_nom"], + test["red_nom"] + ) + + print(f"결과: {result['fitting_type']['type']} - {result['fitting_type']['subtype']}") + print(f"크기: {result['size_info']['size_description']}") # ← 추가! + print(f"신뢰도: {result['overall_confidence']}") + print(f"증거: {result['fitting_type']['evidence']}") + +if __name__ == "__main__": + test_fitting_classification() + test_fitting_edge_cases() diff --git a/backend/app/services/test_flange_classifier.py b/backend/app/services/test_flange_classifier.py new file mode 100644 index 0000000..ee53793 --- /dev/null +++ b/backend/app/services/test_flange_classifier.py @@ -0,0 +1,126 @@ +""" +FLANGE 분류 테스트 +""" + +from .flange_classifier import classify_flange, get_flange_purchase_info + +def test_flange_classification(): + """FLANGE 분류 테스트""" + + test_cases = [ + { + "name": "웰드넥 플랜지 (일반)", + "dat_file": "FLG_WN_150", + "description": "FLG WELD NECK RF SCH 40, 150LB, ASTM A105", + "main_nom": "4\"", + "red_nom": None + }, + { + "name": "슬립온 플랜지 (일반)", + "dat_file": "FLG_SO_300", + "description": "FLG SLIP ON RF, 300LB, ASTM A105", + "main_nom": "3\"", + "red_nom": None + }, + { + "name": "블라인드 플랜지 (일반)", + "dat_file": "FLG_BL_150", + "description": "FLG BLIND RF, 150LB, ASTM A105", + "main_nom": "6\"", + "red_nom": None + }, + { + "name": "소켓웰드 플랜지 (고압)", + "dat_file": "FLG_SW_3000", + "description": "FLG SW, 3000LB, ASTM A105", + "main_nom": "1\"", + "red_nom": None + }, + { + "name": "오리피스 플랜지 (SPECIAL)", + "dat_file": "FLG_ORI_150", + "description": "FLG ORIFICE RF, 150LB, 0.5 INCH BORE, ASTM A105", + "main_nom": "4\"", + "red_nom": None + }, + { + "name": "스펙터클 블라인드 (SPECIAL)", + "dat_file": "FLG_SPB_300", + "description": "FLG SPECTACLE BLIND, 300LB, ASTM A105", + "main_nom": "6\"", + "red_nom": None + }, + { + "name": "리듀싱 플랜지 (SPECIAL)", + "dat_file": "FLG_RED_300", + "description": "FLG REDUCING, 300LB, ASTM A105", + "main_nom": "6\"", + "red_nom": "4\"" + }, + { + "name": "스페이서 플랜지 (SPECIAL)", + "dat_file": "FLG_SPC_600", + "description": "FLG SPACER, 600LB, 2 INCH THK, ASTM A105", + "main_nom": "3\"", + "red_nom": None + } + ] + + print("🔩 FLANGE 분류 테스트 시작\n") + print("=" * 80) + + for i, test in enumerate(test_cases, 1): + print(f"\n테스트 {i}: {test['name']}") + print("-" * 60) + + result = classify_flange( + test["dat_file"], + test["description"], + test["main_nom"], + test["red_nom"] + ) + + purchase_info = get_flange_purchase_info(result) + + print(f"📋 입력:") + print(f" DAT_FILE: {test['dat_file']}") + print(f" DESCRIPTION: {test['description']}") + print(f" SIZE: {result['size_info']['size_description']}") + + print(f"\n🔩 분류 결과:") + print(f" 재질: {result['material']['standard']} | {result['material']['grade']}") + print(f" 카테고리: {'🌟 SPECIAL' if result['flange_category']['is_special'] else '📋 STANDARD'}") + print(f" 플랜지타입: {result['flange_type']['type']}") + print(f" 특성: {result['flange_type']['characteristics']}") + print(f" 면가공: {result['face_finish']['finish']}") + print(f" 압력등급: {result['pressure_rating']['rating']} ({result['pressure_rating']['common_use']})") + print(f" 제작방법: {result['manufacturing']['method']} ({result['manufacturing']['characteristics']})") + + if result['flange_type']['special_features']: + print(f" 특수기능: {', '.join(result['flange_type']['special_features'])}") + + print(f"\n📊 신뢰도:") + print(f" 전체신뢰도: {result['overall_confidence']}") + print(f" 재질: {result['material']['confidence']}") + print(f" 플랜지타입: {result['flange_type']['confidence']}") + print(f" 면가공: {result['face_finish']['confidence']}") + print(f" 압력등급: {result['pressure_rating']['confidence']}") + + print(f"\n🛒 구매 정보:") + print(f" 공급업체: {purchase_info['supplier_type']}") + print(f" 예상납기: {purchase_info['lead_time_estimate']}") + print(f" 구매카테고리: {purchase_info['purchase_category']}") + if purchase_info['special_requirements']: + print(f" 특수요구사항: {', '.join(purchase_info['special_requirements'])}") + + print(f"\n💾 저장될 데이터:") + print(f" MAIN_NOM: {result['size_info']['main_size']}") + print(f" RED_NOM: {result['size_info']['reduced_size'] or 'NULL'}") + print(f" FLANGE_CATEGORY: {'SPECIAL' if result['flange_category']['is_special'] else 'STANDARD'}") + print(f" FLANGE_TYPE: {result['flange_type']['type']}") + + if i < len(test_cases): + print("\n" + "=" * 80) + +if __name__ == "__main__": + test_flange_classification() diff --git a/backend/app/services/test_material_classifier.py b/backend/app/services/test_material_classifier.py new file mode 100644 index 0000000..db4e6a9 --- /dev/null +++ b/backend/app/services/test_material_classifier.py @@ -0,0 +1,53 @@ +""" +재질 분류 테스트 +""" + +from .material_classifier import classify_material, get_manufacturing_method_from_material + +def test_material_classification(): + """재질 분류 테스트""" + + test_cases = [ + { + "description": "PIPE, SMLS, SCH 80, ASTM A106 GR B BOE-POE", + "expected_standard": "ASTM A106" + }, + { + "description": "TEE, SW, 3000LB, ASTM A182 F304", + "expected_standard": "ASTM A182" + }, + { + "description": "90 LR ELL, SCH 40, ASTM A234 GR WPB, SMLS", + "expected_standard": "ASTM A234" + }, + { + "description": "FLG WELD NECK RF SCH 40, 150LB, ASTM A105", + "expected_standard": "ASTM A105" + }, + { + "description": "GATE VALVE, ASTM A216 WCB", + "expected_standard": "ASTM A216" + }, + { + "description": "SPECIAL FITTING, INCONEL 625", + "expected_standard": "INCONEL" + } + ] + + print("🔧 재질 분류 테스트 시작\n") + + for i, test in enumerate(test_cases, 1): + result = classify_material(test["description"]) + manufacturing = get_manufacturing_method_from_material(result) + + print(f"테스트 {i}:") + print(f" 입력: {test['description']}") + print(f" 결과: {result['standard']} | {result['grade']}") + print(f" 재질타입: {result['material_type']}") + print(f" 제작방법: {manufacturing}") + print(f" 신뢰도: {result['confidence']}") + print(f" 증거: {result.get('evidence', [])}") + print() + +if __name__ == "__main__": + test_material_classification() diff --git a/backend/app/services/test_pipe_classifier.py b/backend/app/services/test_pipe_classifier.py new file mode 100644 index 0000000..4ed815c --- /dev/null +++ b/backend/app/services/test_pipe_classifier.py @@ -0,0 +1,55 @@ +""" +PIPE 분류 테스트 +""" + +from .pipe_classifier import classify_pipe, generate_pipe_cutting_plan + +def test_pipe_classification(): + """PIPE 분류 테스트""" + + test_cases = [ + { + "dat_file": "PIP_PE", + "description": "PIPE, SMLS, SCH 80, ASTM A106 GR B BOE-POE", + "main_nom": "1\"", + "length": 798.1965 + }, + { + "dat_file": "NIP_TR", + "description": "NIPPLE, SMLS, SCH 80, ASTM A106 GR B PBE", + "main_nom": "1\"", + "length": 75.0 + }, + { + "dat_file": "PIPE_SPOOL", + "description": "PIPE SPOOL, WELDED, SCH 40, CS", + "main_nom": "2\"", + "length": None + } + ] + + print("🔧 PIPE 분류 테스트 시작\n") + + for i, test in enumerate(test_cases, 1): + result = classify_pipe( + test["dat_file"], + test["description"], + test["main_nom"], + test["length"] + ) + + cutting_plan = generate_pipe_cutting_plan(result) + + print(f"테스트 {i}:") + print(f" 입력: {test['description']}") + print(f" 재질: {result['material']['standard']} | {result['material']['grade']}") + print(f" 제조방법: {result['manufacturing']['method']}") + print(f" 끝가공: {result['end_preparation']['cutting_note']}") + print(f" 스케줄: {result['schedule']['schedule']}") + print(f" 절단치수: {result['cutting_dimensions']['length_mm']}mm") + print(f" 전체신뢰도: {result['overall_confidence']}") + print(f" 절단지시: {cutting_plan['cutting_instruction']}") + print() + +if __name__ == "__main__": + test_pipe_classification() diff --git a/backend/database.py b/backend/database.py new file mode 100644 index 0000000..763453d --- /dev/null +++ b/backend/database.py @@ -0,0 +1,29 @@ +아! PostgreSQL 연결 문제입니다! 🔧 현재 database.py에서 더미 연결 정보를 사용하고 있어서 발생한 문제네요. +🔍 해결 방법 선택 +방법 1: SQLite로 변경 (간단함) +bashcd TK-MP-Project/backend/app + +# database.py를 SQLite로 변경 +cat > database.py << 'EOF' +from sqlalchemy import create_engine +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker +import os + +# SQLite 사용 (로컬 개발용) +DATABASE_URL = "sqlite:///./materials.db" + +engine = create_engine( + DATABASE_URL, + connect_args={"check_same_thread": False} # SQLite 전용 설정 +) + +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) +Base = declarative_base() + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/backend/example_corrected_spool_usage.py b/backend/example_corrected_spool_usage.py new file mode 100644 index 0000000..40e6afb --- /dev/null +++ b/backend/example_corrected_spool_usage.py @@ -0,0 +1,30 @@ +""" +수정된 스풀 시스템 사용 예시 +""" + +# 시나리오: A-1 도면에서 파이프 3개 발견 +examples = [ + { + "dwg_name": "A-1", + "pipes": [ + {"description": "PIPE 1", "user_input_spool": "A"}, # A-1-A + {"description": "PIPE 2", "user_input_spool": "A"}, # A-1-A (같은 스풀) + {"description": "PIPE 3", "user_input_spool": "B"} # A-1-B (다른 스풀) + ], + "area_assignment": "#01" # 별도: A-1 도면은 #01 구역에 위치 + } +] + +# 결과: +spool_identifiers = [ + "A-1-A", # 파이프 1, 2가 속함 + "A-1-B" # 파이프 3이 속함 +] + +area_assignment = { + "#01": ["A-1"] # A-1 도면은 #01 구역에 물리적으로 위치 +} + +print("✅ 수정된 스풀 구조가 적용되었습니다!") +print(f"스풀 식별자: {spool_identifiers}") +print(f"에리어 할당: {area_assignment}") diff --git a/backend/temp_main_update.py b/backend/temp_main_update.py new file mode 100644 index 0000000..aa69547 --- /dev/null +++ b/backend/temp_main_update.py @@ -0,0 +1,5 @@ +# main.py에 추가할 import +from .api import spools + +# app.include_router 추가 +app.include_router(spools.router, prefix="/api/spools", tags=["스풀 관리"])