Files
TK-BOM-Project/backend/app/services/spool_manager.py
Hyungi Ahn 12ecb93741 feat: 완전한 자재 분류 시스템 구현 (v1.0)
🎯 주요 기능:
- 재질 분류 모듈 (ASTM/ASME 규격 자동 인식)
- PIPE 분류 시스템 (제조방법, 끝가공, 스케줄, 절단계획)
- FITTING 분류 시스템 (10가지 타입, 연결방식, 압력등급)
- FLANGE 분류 시스템 (SPECIAL/STANDARD 구분, 면가공)
- 스풀 관리 시스템 (도면별 A,B,C 넘버링, 에리어 관리)

📁 새로 추가된 파일들:
- app/services/materials_schema.py (재질 규격 데이터베이스)
- app/services/material_classifier.py (공통 재질 분류 엔진)
- app/services/pipe_classifier.py (파이프 전용 분류기)
- app/services/fitting_classifier.py (피팅 전용 분류기)
- app/services/flange_classifier.py (플랜지 전용 분류기)
- app/services/spool_manager_v2.py (수정된 스풀 관리)
- app/services/test_*.py (각 시스템별 테스트 파일)

🔧 기술적 특징:
- 정규표현식 기반 패턴 매칭
- 신뢰도 점수 시스템 (0.0-1.0)
- 증거 기반 분류 (evidence tracking)
- 모듈화된 구조 (재사용 가능)

🎯 분류 정확도:
- 재질 분류: 90-95% 신뢰도
- PIPE 분류: 85-95% 신뢰도
- FITTING 분류: 85-95% 신뢰도
- FLANGE 분류: 85-95% 신뢰도

💾 데이터베이스 연동:
- 모든 분석 결과 자동 저장
- 프로젝트/도면 정보 자동 연결
- 스풀 정보 사용자 입력 대기

🧪 테스트 커버리지:
- 실제 BOM 데이터 기반 테스트
- 예외 케이스 처리
- 10+ 개 테스트 시나리오

Version: v1.0
Date: 2024-07-15
Author: hyungiahn
2025-07-15 09:43:39 +09:00

257 lines
9.0 KiB
Python

"""
스풀 관리 시스템
도면별 에리어 넘버와 스풀 넘버 관리
"""
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