Spaces:
Sleeping
Sleeping
| """ | |
| P3 (Fashion Image Segmentation · Attribute Tagging) -- 공통 유틸리티 | |
| 2단계(2-Stage) 파이프라인 전반에서 공유하는 보조 도구를 모았습니다. | |
| Stage 1 : Mask R-CNN instance segmentation (46개 카테고리) | |
| Stage 2 : 인스턴스 crop 기반 multi-label attribute tagging (294개 속성) | |
| [주요 구성] | |
| 1. 카테고리 상수 MAIN_GARMENT_NAMES(27) / GARMENT_PART_NAMES(19) | |
| 2. load_category_ids() 런타임에 COCO JSON → (MAIN_IDS, PART_IDS, id2name) | |
| 3. load_attribute_supers() 인덱스 → 속성 슈퍼카테고리 매핑 (그룹별 메트릭용) | |
| 4. set_seed() 재현성 시드 고정 | |
| 5. AverageMeter 실행 평균 추적 | |
| 6. EarlyStopping 조기 종료 (mAP@0.5 / F1 모니터링) | |
| 7. compute_segmentation_metrics() Stage 1 -- COCO mAP (Main/Part 분리) | |
| 8. compute_attribute_metrics() Stage 2 -- macro F1 + 슈퍼카테고리별 F1 | |
| 9. collate_fn_stage1() Mask R-CNN용 variable-size collate | |
| 10. setup_logger() 파일 + 콘솔 동시 로깅 | |
| 11. get_pos_weight() inverse-frequency BCE pos_weight | |
| 근거: 카테고리/속성 구조는 EDA(`004_EDA/P3_EDA_short.ipynb` cell-05)에서 검증. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import json | |
| import random | |
| import logging | |
| from collections import Counter | |
| from typing import Dict, List, Optional, Sequence, Tuple | |
| import numpy as np | |
| import torch | |
| # ============================================================================ | |
| # 1. 카테고리 상수 (EDA 결과 기반 -- 004_EDA/P3_EDA_short.ipynb cell-05) | |
| # ---------------------------------------------------------------------------- | |
| # Fashionpedia 46개 카테고리 = Main 의류 27개 + Part 부속/장식 19개. | |
| # Main 만 Stage 2 속성 태깅의 입력으로 사용한다 (EDA 가정 ③: Main 속성이 | |
| # Part 보다 압도적으로 풍부). 이름은 COCO `categories[*].name` 과 정확히 일치. | |
| # ============================================================================ | |
| #: Stage 2 입력이 되는 본체 의류 27개 (옷 + 액세서리) | |
| MAIN_GARMENT_NAMES = { | |
| # 상의 | |
| "shirt, blouse", "top, t-shirt, sweatshirt", "sweater", "cardigan", | |
| "jacket", "vest", | |
| # 하의 / 원피스류 | |
| "pants", "shorts", "skirt", "coat", "dress", "jumpsuit", "cape", | |
| # 헤드 / 액세서리 | |
| "glasses", "hat", "headband, head covering, hair accessory", | |
| "tie", "glove", "watch", "belt", | |
| # 다리 / 발 | |
| "leg warmer", "tights, stockings", "sock", "shoe", | |
| # 소품 | |
| "bag, wallet", "scarf", "umbrella", | |
| } | |
| #: 의류에 부속/장식되는 파트 19개 (Stage 2에서 제외, Stage 1에서만 검출) | |
| GARMENT_PART_NAMES = { | |
| # 구조 부위 | |
| "hood", "collar", "lapel", "epaulette", "sleeve", "pocket", "neckline", | |
| "buckle", "zipper", | |
| # 장식 요소 | |
| "applique", "bead", "bow", "flower", "fringe", "ribbon", "rivet", | |
| "ruffle", "sequin", "tassel", | |
| } | |
| def load_category_ids( | |
| coco_json_path: str, | |
| ) -> Tuple[set, set, Dict[int, str]]: | |
| """COCO JSON을 읽어 Main / Part 카테고리 ID 집합과 ID→이름 매핑을 만든다. | |
| 카테고리 ID 는 데이터셋 버전에 따라 달라질 수 있으므로 하드코딩하지 않고 | |
| 런타임에 이름(MAIN_GARMENT_NAMES / GARMENT_PART_NAMES)으로 매핑한다. | |
| Args: | |
| coco_json_path (str): instances_attributes_*.json 경로 | |
| Returns: | |
| Tuple[set, set, Dict[int, str]]: | |
| - MAIN_IDS : Main 카테고리 ID 집합 (27개) | |
| - PART_IDS : Part 카테고리 ID 집합 (19개) | |
| - id2name : {category_id: name} 전체 매핑 (46개) | |
| Raises: | |
| ValueError: JSON 의 카테고리 이름이 정의된 상수와 어긋날 때 | |
| (Main+Part 합이 46이 아니거나 분류되지 않은 이름이 있을 때). | |
| """ | |
| with open(coco_json_path, "r") as f: | |
| raw = json.load(f) | |
| id2name = {c["id"]: c["name"] for c in raw["categories"]} | |
| main_ids, part_ids, unknown = set(), set(), [] | |
| for cid, name in id2name.items(): | |
| if name in MAIN_GARMENT_NAMES: | |
| main_ids.add(cid) | |
| elif name in GARMENT_PART_NAMES: | |
| part_ids.add(cid) | |
| else: | |
| unknown.append(name) | |
| if unknown: | |
| raise ValueError( | |
| f"Main/Part 어느 쪽에도 속하지 않는 카테고리: {unknown}. " | |
| "MAIN_GARMENT_NAMES / GARMENT_PART_NAMES 상수를 확인하세요." | |
| ) | |
| if len(main_ids) != len(MAIN_GARMENT_NAMES) or len(part_ids) != len(GARMENT_PART_NAMES): | |
| raise ValueError( | |
| f"카테고리 개수 불일치: Main {len(main_ids)}/{len(MAIN_GARMENT_NAMES)}, " | |
| f"Part {len(part_ids)}/{len(GARMENT_PART_NAMES)}" | |
| ) | |
| return main_ids, part_ids, id2name | |
| def load_attribute_supers( | |
| coco_json_path: str, | |
| num_attrs: int = 294, | |
| ) -> Tuple[List[str], Dict[int, int]]: | |
| """속성 인덱스 → 슈퍼카테고리 매핑을 만든다 (슈퍼카테고리별 F1 메트릭용). | |
| Fashionpedia 속성 ID 는 0~340 범위에 비연속적으로 분포(총 294개)하므로, | |
| 학습용 multi-hot 벡터의 인덱스는 **속성 ID 오름차순 순서**로 0..293에 매핑한다. | |
| preprocess 의 multi-hot 생성과 반드시 동일한 정렬 규칙을 사용해야 한다. | |
| Args: | |
| coco_json_path (str): instances_attributes_*.json 경로 | |
| num_attrs (int): 속성 개수 (기본 294) | |
| Returns: | |
| Tuple[List[str], Dict[int, int]]: | |
| - idx2super : 길이 num_attrs 리스트. idx2super[i] = i번째 속성의 슈퍼카테고리 | |
| - attr_id2idx : {원본 attribute_id: 0..num_attrs-1 인덱스} | |
| Note: | |
| Fashionpedia 데이터의 실제 슈퍼카테고리는 11종이다 | |
| (animal, leather, length, neckline type, nickname, non-textile material type, | |
| opening type, silhouette, textile finishing, textile pattern, waistline). | |
| """ | |
| with open(coco_json_path, "r") as f: | |
| raw = json.load(f) | |
| attrs = sorted(raw["attributes"], key=lambda a: a["id"]) | |
| if len(attrs) != num_attrs: | |
| raise ValueError(f"속성 개수 불일치: JSON {len(attrs)} != num_attrs {num_attrs}") | |
| idx2super = [a["supercategory"] for a in attrs] | |
| attr_id2idx = {a["id"]: i for i, a in enumerate(attrs)} | |
| return idx2super, attr_id2idx | |
| # ============================================================================ | |
| # 2. 재현성 | |
| # ============================================================================ | |
| def set_seed(seed: int = 42) -> None: | |
| """random / numpy / torch / cudnn 시드를 모두 고정해 재현성을 확보한다. | |
| Args: | |
| seed (int): 랜덤 시드 (기본 42) | |
| """ | |
| os.environ["PYTHONHASHSEED"] = str(seed) | |
| random.seed(seed) | |
| np.random.seed(seed) | |
| torch.manual_seed(seed) | |
| torch.cuda.manual_seed_all(seed) | |
| # 완전 결정론적 동작 (속도보다 재현성 우선) | |
| torch.backends.cudnn.deterministic = True | |
| torch.backends.cudnn.benchmark = False | |
| # ============================================================================ | |
| # 3. 학습 보조 클래스 | |
| # ============================================================================ | |
| class AverageMeter: | |
| """실행 평균과 합계를 추적하는 누적기. | |
| 배치 단위 loss / 메트릭의 에폭 평균을 구할 때 사용한다. | |
| 사용 예: | |
| meter = AverageMeter("loss") | |
| for loss, bs in batches: | |
| meter.update(loss, n=bs) | |
| print(meter.avg) | |
| """ | |
| def __init__(self, name: str = "metric") -> None: | |
| self.name = name | |
| self.reset() | |
| def reset(self) -> None: | |
| """모든 통계를 0으로 초기화.""" | |
| self.val = 0.0 | |
| self.avg = 0.0 | |
| self.sum = 0.0 | |
| self.count = 0 | |
| def update(self, val: float, n: int = 1) -> None: | |
| """새 값으로 통계를 갱신한다. | |
| Args: | |
| val (float): 현재 배치의 (평균) 값 | |
| n (int): 표본 수 (배치 크기). 가중 평균에 사용 | |
| """ | |
| self.val = val | |
| self.sum += val * n | |
| self.count += n | |
| self.avg = self.sum / self.count if self.count > 0 else 0.0 | |
| def __repr__(self) -> str: | |
| return f"{self.name}: {self.avg:.4f} (n={self.count})" | |
| class EarlyStopping: | |
| """검증 메트릭이 patience 에폭 동안 개선되지 않으면 학습 중단 신호를 준다. | |
| Stage 1 은 mAP@0.5 (mode='max'), Stage 2 는 macro F1 (mode='max') 을 모니터링. | |
| loss 처럼 작아야 좋은 지표는 mode='min' 으로 사용. | |
| """ | |
| def __init__( | |
| self, | |
| patience: int = 5, | |
| mode: str = "max", | |
| delta: float = 0.0, | |
| ) -> None: | |
| """초기화. | |
| Args: | |
| patience (int): 개선 없이 허용할 최대 에폭 수 | |
| mode (str): "max"(클수록 좋음) 또는 "min"(작을수록 좋음) | |
| delta (float): 개선으로 인정할 최소 변화량 | |
| """ | |
| if mode not in ("max", "min"): | |
| raise ValueError(f"mode 는 'max' 또는 'min' 이어야 합니다 (받은 값: {mode})") | |
| self.patience = patience | |
| self.mode = mode | |
| self.delta = delta | |
| self.counter = 0 | |
| self.best_score: Optional[float] = None | |
| self.early_stop = False | |
| def _is_improved(self, score: float) -> bool: | |
| if self.best_score is None: | |
| return True | |
| if self.mode == "max": | |
| return score > self.best_score + self.delta | |
| return score < self.best_score - self.delta | |
| def __call__(self, metric: float) -> bool: | |
| """현재 에폭 메트릭을 받아 학습 중단 여부를 반환한다. | |
| Args: | |
| metric (float): 현재 에폭의 검증 메트릭 | |
| Returns: | |
| bool: True 이면 학습을 멈춰야 함 | |
| """ | |
| if self._is_improved(metric): | |
| self.best_score = metric | |
| self.counter = 0 | |
| else: | |
| self.counter += 1 | |
| if self.counter >= self.patience: | |
| self.early_stop = True | |
| return self.early_stop | |
| # ============================================================================ | |
| # 4. collate_fn (Stage 1) | |
| # ============================================================================ | |
| def collate_fn_stage1(batch: List[Tuple]) -> Tuple[List, List]: | |
| """Mask R-CNN용 collate 함수. | |
| torchvision detection 모델은 서로 다른 크기의 이미지를 하나의 텐서로 | |
| stack 하지 않고 **리스트** 그대로 입력받는다. 기본 collate 의 자동 stacking | |
| 을 우회하기 위해 (이미지 리스트, 타겟 리스트) 로만 분리한다. | |
| Args: | |
| batch (List[Tuple]): [(image_tensor, target_dict), ...] | |
| Returns: | |
| Tuple[List, List]: (images, targets) -- 둘 다 길이 B 리스트 | |
| """ | |
| images, targets = zip(*batch) | |
| return list(images), list(targets) | |
| # ============================================================================ | |
| # 5. Stage 1 메트릭 -- COCO mAP | |
| # ============================================================================ | |
| def compute_segmentation_metrics( | |
| coco_gt, | |
| coco_dt, | |
| main_ids: Optional[set] = None, | |
| part_ids: Optional[set] = None, | |
| iou_type: str = "segm", | |
| ) -> Dict[str, float]: | |
| """pycocotools COCOeval 로 instance segmentation mAP 를 계산한다. | |
| Args: | |
| coco_gt: GT COCO 객체 (`pycocotools.coco.COCO`) | |
| coco_dt: 예측 결과 COCO 객체. 보통 `coco_gt.loadRes(results)` 로 생성하며, | |
| results 는 [{"image_id", "category_id", "segmentation"(RLE), | |
| "score"}, ...] 형식의 리스트. | |
| main_ids (set, optional): Main 카테고리 ID 집합. 주면 Main mAP 분리 계산. | |
| part_ids (set, optional): Part 카테고리 ID 집합. 주면 Part mAP 분리 계산. | |
| iou_type (str): "segm"(마스크) 또는 "bbox". | |
| Returns: | |
| Dict[str, float]: { | |
| "mAP_50": mAP @ IoU=0.50, | |
| "mAP_50_95": mAP @ IoU=0.50:0.95 (COCO primary), | |
| "main_mAP_50": (main_ids 제공 시) Main 카테고리 mAP@0.5, | |
| "part_mAP_50": (part_ids 제공 시) Part 카테고리 mAP@0.5, | |
| } | |
| Note: | |
| COCOeval.stats[0] = mAP@[.5:.95], stats[1] = mAP@0.5. | |
| 값이 -1 이면 해당 카테고리에 대한 예측/GT 가 없다는 뜻. | |
| """ | |
| from pycocotools.cocoeval import COCOeval | |
| def _run(cat_ids: Optional[Sequence[int]] = None) -> Tuple[float, float]: | |
| evaluator = COCOeval(coco_gt, coco_dt, iouType=iou_type) | |
| if cat_ids is not None: | |
| evaluator.params.catIds = list(cat_ids) | |
| evaluator.evaluate() | |
| evaluator.accumulate() | |
| evaluator.summarize() | |
| # stats[0]=mAP@[.5:.95], stats[1]=mAP@0.5 | |
| return float(evaluator.stats[1]), float(evaluator.stats[0]) | |
| map_50, map_50_95 = _run() | |
| metrics: Dict[str, float] = {"mAP_50": map_50, "mAP_50_95": map_50_95} | |
| if main_ids: | |
| metrics["main_mAP_50"] = _run(main_ids)[0] | |
| if part_ids: | |
| metrics["part_mAP_50"] = _run(part_ids)[0] | |
| return metrics | |
| # ============================================================================ | |
| # 6. Stage 2 메트릭 -- multi-label attribute F1 | |
| # ============================================================================ | |
| def compute_attribute_metrics( | |
| logits, | |
| targets, | |
| threshold: float = 0.5, | |
| attr_supers: Optional[Sequence[str]] = None, | |
| ) -> Dict[str, float]: | |
| """multi-label 속성 분류 메트릭 (macro F1/precision/recall + 슈퍼카테고리별 F1). | |
| Args: | |
| logits: 모델 출력 [N, num_attrs]. **raw logit** 으로 보고 내부에서 | |
| sigmoid 를 적용한다 (torch.Tensor / np.ndarray 모두 허용). | |
| targets: multi-hot 정답 [N, num_attrs] (0/1). | |
| threshold (float): sigmoid 확률을 1로 판정할 임계값. | |
| attr_supers (Sequence[str], optional): 길이 num_attrs. 각 속성 인덱스의 | |
| 슈퍼카테고리 이름 (load_attribute_supers 로 얻음). 주면 그룹별 | |
| macro F1 을 함께 반환한다. | |
| Returns: | |
| Dict[str, float]: { | |
| "f1_macro", "precision_macro", "recall_macro", | |
| "f1_micro", | |
| "f1_super/<supercategory>": ... (attr_supers 제공 시 그룹별) | |
| } | |
| """ | |
| from sklearn.metrics import f1_score, precision_score, recall_score | |
| probs = _to_numpy(logits) | |
| y_true = _to_numpy(targets).astype(int) | |
| y_pred = (_sigmoid(probs) >= threshold).astype(int) | |
| metrics = { | |
| "f1_macro": float(f1_score(y_true, y_pred, average="macro", zero_division=0)), | |
| "precision_macro": float(precision_score(y_true, y_pred, average="macro", zero_division=0)), | |
| "recall_macro": float(recall_score(y_true, y_pred, average="macro", zero_division=0)), | |
| "f1_micro": float(f1_score(y_true, y_pred, average="micro", zero_division=0)), | |
| } | |
| # 슈퍼카테고리(9~11개 그룹)별 macro F1 | |
| if attr_supers is not None: | |
| if len(attr_supers) != y_true.shape[1]: | |
| raise ValueError( | |
| f"attr_supers 길이({len(attr_supers)})가 속성 차원({y_true.shape[1]})과 다릅니다." | |
| ) | |
| attr_supers = np.asarray(attr_supers) | |
| for group in sorted(set(attr_supers.tolist())): | |
| cols = np.where(attr_supers == group)[0] | |
| f1_g = f1_score( | |
| y_true[:, cols], y_pred[:, cols], average="macro", zero_division=0 | |
| ) | |
| metrics[f"f1_super/{group}"] = float(f1_g) | |
| return metrics | |
| # ============================================================================ | |
| # 7. 로깅 | |
| # ============================================================================ | |
| def setup_logger(log_dir: str, name: str = "train") -> logging.Logger: | |
| """파일과 콘솔에 동시 출력하는 로거를 만든다. | |
| Args: | |
| log_dir (str): 로그 파일을 저장할 디렉토리 (없으면 생성) | |
| name (str): 로거 이름 겸 로그 파일명 (`<name>.log`) | |
| Returns: | |
| logging.Logger: 설정된 로거 | |
| """ | |
| os.makedirs(log_dir, exist_ok=True) | |
| logger = logging.getLogger(name) | |
| logger.setLevel(logging.INFO) | |
| # 중복 핸들러 방지 (재호출 시 로그가 여러 번 찍히는 문제) | |
| if logger.handlers: | |
| return logger | |
| logger.propagate = False | |
| fmt = logging.Formatter( | |
| "[%(asctime)s] %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" | |
| ) | |
| file_handler = logging.FileHandler(os.path.join(log_dir, f"{name}.log")) | |
| file_handler.setFormatter(fmt) | |
| logger.addHandler(file_handler) | |
| stream_handler = logging.StreamHandler() | |
| stream_handler.setFormatter(fmt) | |
| logger.addHandler(stream_handler) | |
| return logger | |
| # ============================================================================ | |
| # 8. BCE pos_weight (속성 long-tail 보정 -- EDA 가정 ⑤) | |
| # ============================================================================ | |
| def get_pos_weight( | |
| attr_counter: Counter, | |
| num_attrs: int = 294, | |
| max_weight: float = 10.0, | |
| attr_ids: Optional[Sequence[int]] = None, | |
| ) -> torch.Tensor: | |
| """속성 빈도의 역수로 BCEWithLogitsLoss 의 pos_weight 를 계산한다. | |
| 희귀 속성(long-tail)일수록 양성 신호가 약하므로 손실에서 더 크게 가중한다. | |
| 가장 흔한 속성을 기준(1.0)으로 한 inverse-frequency 가중을 사용하고, | |
| 극단적 가중을 막기 위해 max_weight 로 clipping 한다. | |
| pos_weight[i] = clip(max(count) / count[i], 1.0, max_weight) | |
| Args: | |
| attr_counter (Counter): {속성키: 등장횟수}. 키는 attr_ids 와 같은 체계여야 함. | |
| num_attrs (int): 속성 개수 (출력 벡터 길이, 기본 294) | |
| max_weight (float): pos_weight 상한 (기본 10.0) | |
| attr_ids (Sequence[int], optional): 인덱스 i 에 대응하는 원본 속성 ID 목록. | |
| Fashionpedia 처럼 ID 가 비연속(0~340)일 때 정렬된 ID 리스트를 넘긴다. | |
| None 이면 키를 0..num_attrs-1 인덱스로 간주(연속 가정). | |
| Returns: | |
| torch.Tensor: shape (num_attrs,), dtype float32. | |
| """ | |
| keys = list(attr_ids) if attr_ids is not None else list(range(num_attrs)) | |
| if len(keys) != num_attrs: | |
| raise ValueError(f"attr_ids 길이({len(keys)})가 num_attrs({num_attrs})와 다릅니다.") | |
| # 등장 0회는 1로 바닥 처리 (0 나눗셈 방지) | |
| counts = np.array([max(attr_counter.get(k, 0), 1) for k in keys], dtype=np.float64) | |
| pos_weight = counts.max() / counts | |
| pos_weight = np.clip(pos_weight, 1.0, max_weight) | |
| return torch.tensor(pos_weight, dtype=torch.float32) | |
| # ============================================================================ | |
| # 내부 헬퍼 | |
| # ============================================================================ | |
| def _to_numpy(x) -> np.ndarray: | |
| """torch.Tensor / np.ndarray / list 를 numpy 배열로 변환.""" | |
| if isinstance(x, torch.Tensor): | |
| return x.detach().cpu().numpy() | |
| return np.asarray(x) | |
| def _sigmoid(x: np.ndarray) -> np.ndarray: | |
| """수치 안정적인 sigmoid.""" | |
| return np.where(x >= 0, 1.0 / (1.0 + np.exp(-x)), np.exp(x) / (1.0 + np.exp(x))) | |