프로젝트 1 — 부품 입고 검사 파이프라인 완성

import hashlib
import json
import os
import time
from pathlib import Path
import cv2
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

plt.rcParams["font.family"] = "Malgun Gothic"
plt.rcParams["axes.unicode_minus"] = False

CONFIG = {
    "exts": [".png", ".jpg", ".jpeg"], "min_side": 64,
    "dark_drop": 30, "blur_min": 30, "hash_max": 2,
    "size": [320, 240], "dark_fix": 50, "gamma": 0.7, "noise_max": 3.0, "denoise_ksize": 5,
    "weights": "out_ch18_runs/parts/weights/best.pt", "conf": 0.25,
    "iou_hit": 0.5, "seed": 19,
}

def collect(folder, config):
    """수집: 확장자로 그림만 고른다."""
    return [{"file": p.name, "source": p.as_posix(), "bytes": p.stat().st_size}
            for p in sorted(Path(folder).iterdir()) if p.is_file() and p.suffix.lower() in config["exts"]]

def gray_of(img):
    return cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

def ahash(img):
    small = cv2.resize(gray_of(img), (8, 8), interpolation=cv2.INTER_AREA)
    return (small > small.mean()).flatten()

def filter_stage(items, config):
    """거르기: 읽기 → 밝기 → 흐림 → 중복. 탈락하면 이유를 기록한다."""
    passed, log, seen = [], [], []
    for it in items:
        reason, img = "", None
        if it["bytes"] == 0:
            reason = "빈 파일"
        else:
            img = cv2.imread(it["source"])
            if img is None:
                reason = "깨짐"
            elif min(img.shape[:2]) < config["min_side"]:
                reason = "너무 작음"
        if not reason:
            mean = float(gray_of(img).mean())
            blur = float(cv2.Laplacian(gray_of(img), cv2.CV_64F).var())
            if mean < config["dark_drop"]:
                reason = f"아주 어두움({mean:.1f})"
            elif blur < config["blur_min"]:
                reason = f"흐림({blur:.1f})"
        if not reason:
            with open(it["source"], "rb") as f:
                digest = hashlib.md5(f.read()).hexdigest()
            bits = ahash(img)
            for other, d, b in seen:
                if d == digest or int((b != bits).sum()) <= config["hash_max"]:
                    reason = f"중복({other})"
                    break
        if reason:
            log.append({"file": it["file"], "stage": "거르기", "result": "탈락", "reason": reason})
        else:
            seen.append((it["file"], digest, bits))
            passed.append({**it, "img": img})
    return passed, log

# 수집과 거르기만 먼저 시험한다
test_passed, test_log = filter_stage(collect("ch19_incoming", CONFIG), CONFIG)
print("통과", len(test_passed), "장 · 탈락", [(r["file"], r["reason"]) for r in test_log])

def prep_stage(items, config):
    """다듬기: 크기 → 밝기 → 잡음. 비율과 손질을 항목에 적는다."""
    w, h = config["size"]
    lut = np.array([((i / 255) ** config["gamma"]) * 255 for i in range(256)]).round().astype(np.uint8)
    out = []
    for it in items:
        img, notes = it["img"], []
        scale = w / img.shape[1]
        if img.shape[:2] != (h, w):
            img = cv2.resize(img, (w, h), interpolation=cv2.INTER_AREA)
            notes.append(f"크기 {scale:g}배")
        if float(gray_of(img).mean()) < config["dark_fix"]:
            img = cv2.LUT(img, lut)
            notes.append(f"감마 {config['gamma']}")
        g = gray_of(img)
        if float(cv2.absdiff(g, cv2.medianBlur(g, 3)).mean()) > config["noise_max"]:
            k = config["denoise_ksize"]
            img = cv2.GaussianBlur(img, (k, k), 0)
            notes.append(f"블러 {k}x{k}")
        out.append({**it, "img": img, "scale": scale, "notes": ", ".join(notes)})
    return out

def infer_stage(items, model, config):
    """추론: 그림마다 predict 하고 상자를 원래 좌표로 되돌린 물체 표를 돌려준다."""
    found = []
    for it in items:
        r = model.predict(it["img"], imgsz=config["size"][0], conf=config["conf"], device="cpu", verbose=False)[0]
        for (x1, y1, x2, y2), c, k in zip(r.boxes.xyxy.cpu().numpy(), r.boxes.conf.cpu().numpy(),
                                          r.boxes.cls.cpu().numpy().astype(int)):
            s = it["scale"]
            found.append({"file": it["file"], "class": r.names[int(k)], "conf": round(float(c), 3),
                          "x1": float(x1) / s, "y1": float(y1) / s, "x2": float(x2) / s, "y2": float(y2) / s})
    return pd.DataFrame(found, columns=["file", "class", "conf", "x1", "y1", "x2", "y2"])

# 다듬기만 시험한다(추론은 4단계에서 모델을 불러 함께 돈다)
test_prepped = prep_stage(test_passed, CONFIG)
print("손질:", {it["file"]: it["notes"] for it in test_prepped if it["notes"]})

def iou(a, b):
    """두 상자 (x1, y1, x2, y2) 의 겹친 넓이 ÷ 합친 넓이(18차시)."""
    w = max(0.0, min(a[2], b[2]) - max(a[0], b[0]))
    h = max(0.0, min(a[3], b[3]) - max(a[1], b[1]))
    inter = w * h
    union = (a[2] - a[0]) * (a[3] - a[1]) + (b[2] - b[0]) * (b[3] - b[1]) - inter
    return inter / union if union > 0 else 0.0

# 손계산: (0,0,10,10) 과 (5,0,15,10) 은 겹친 넓이 50, 합친 넓이 100 + 100 - 50 = 150
print("IoU 손계산:", round(iou((0, 0, 10, 10), (5, 0, 15, 10)), 3))

def match_count(preds, truths, thr):
    """같은 그림의 예측과 정답을 짝지어 맞힌 수를 센다. 신뢰도 높은 예측부터, 정답 하나에 예측 하나."""
    used, hits = set(), 0
    for p in sorted(preds, key=lambda d: -d["conf"]):
        best, best_j = 0.0, None
        for j, t in enumerate(truths):
            if j in used or t["class"] != p["class"]:
                continue
            v = iou((p["x1"], p["y1"], p["x2"], p["y2"]), (t["x1"], t["y1"], t["x2"], t["y2"]))
            if v > best:
                best, best_j = v, j
        if best_j is not None and best >= thr:
            used.add(best_j)
            hits += 1
    return hits

# 작은 시험: 예측 둘(하나는 클래스가 틀림), 정답 하나
p = [{"class": "nut", "conf": 0.9, "x1": 0, "y1": 0, "x2": 10, "y2": 10},
     {"class": "bolt", "conf": 0.8, "x1": 0, "y1": 0, "x2": 10, "y2": 10}]
t = [{"class": "nut", "x1": 1, "y1": 0, "x2": 11, "y2": 10}]
print("맞힌 수:", match_count(p, t, 0.5))

def run_pipeline(folder, config):
    """수집 → 거르기 → 다듬기 → 추론 → 채점·보고. 결과 표들을 사전으로 돌려준다."""
    times = {}
    Path("out_ch19_config.json").write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8")

    start = time.perf_counter()
    items = collect(folder, config)
    times["수집"] = time.perf_counter() - start

    start = time.perf_counter()
    passed, log = filter_stage(items, config)
    times["거르기"] = time.perf_counter() - start

    start = time.perf_counter()
    prepped = prep_stage(passed, config)
    times["다듬기"] = time.perf_counter() - start

    start = time.perf_counter()
    os.environ["YOLO_VERBOSE"] = "False"
    from ultralytics import YOLO
    if not Path(config["weights"]).exists():
        raise SystemExit("best.pt 가 없습니다. 18차시 실습 2를 먼저 실행하세요")
    model = YOLO("out_ch18_runs/parts/weights/best.pt")
    det = infer_stage(prepped, model, config)
    times["추론"] = time.perf_counter() - start

    # 채점: 통과한 그림의 정답만 센다
    answer_path = Path(folder) / "answer.csv"
    answer = pd.read_csv(answer_path) if answer_path.exists() else pd.DataFrame(columns=["filename"])
    rows = list(log)
    for it in prepped:
        preds = det[det["file"] == it["file"]].to_dict("records")
        truths = answer[answer["filename"] == it["file"]].to_dict("records")
        rows.append({"file": it["file"], "stage": "추론", "result": "통과", "reason": "", "notes": it["notes"],
                     "found": len(preds), "truth": len(truths),
                     "hits": match_count(preds, truths, config["iou_hit"])})
    report = pd.DataFrame(rows).sort_values("file").reset_index(drop=True)
    # 탈락한 그림은 개수 칸이 비어 소수(1.0)로 바뀌므로, 빈 칸을 허용하는 정수형(Int64)으로 되돌린다
    report[["found", "truth", "hits"]] = report[["found", "truth", "hits"]].astype("Int64")
    report.to_csv("out_ch19_report.csv", index=False, encoding="utf-8-sig")
    return {"items": items, "prepped": prepped, "log": log, "det": det, "report": report, "times": times}

out = run_pipeline("ch19_incoming", CONFIG)
print("단계별 시간(초):", {k: round(v, 3) for k, v in out["times"].items()})
print("저장: out_ch19_config.json, out_ch19_report.csv")

report, det = out["report"], out["det"]
# 깔때기: 탈락 이유의 첫 낱말로 단계를 나눈다
gates = [("읽기", ("빈 파일", "깨짐", "너무 작음")), ("밝기", ("아주 어두움",)), ("흐림", ("흐림",)), ("중복", ("중복",))]
reasons = report["reason"].fillna("")
funnel, left = [("수집", len(out["items"]))], len(out["items"])
for name, heads in gates:
    left -= int(reasons.str.startswith(heads).sum())
    funnel.append((name, left))
funnel.append(("다듬기", len(out["prepped"])))
print("깔때기:", funnel)

counts = det["class"].value_counts().reindex(["bolt", "nut", "washer"], fill_value=0)
print("클래스별 찾은 개수:", counts.to_dict())

found, truth, hits = int(report["found"].sum()), int(report["truth"].sum()), int(report["hits"].sum())
precision = hits / found if found else 0.0
recall = hits / truth if truth else 0.0
print(f"찾음 {found} · 정답 {truth} · 맞힘 {hits} → 정밀도 {precision:.3f}, 재현율 {recall:.3f}")
missed = report[(report["stage"] == "추론") & (report["hits"] < report["truth"])]["file"].tolist()
print("놓친 물체가 있는 그림:", missed)

# 요약 그림: 위 줄에 상자 그림 넷(씨앗으로 고름), 아래 줄에 클래스 막대와 깔때기
COLORS = {"bolt": (76, 139, 95), "nut": (38, 93, 254), "washer": (190, 125, 142)}
rng = np.random.default_rng(CONFIG["seed"])
picks = sorted(rng.choice([it["file"] for it in out["prepped"]], 4, replace=False).tolist())
fig = plt.figure(figsize=(14, 7.5), layout="constrained")
gs = fig.add_gridspec(2, 4)
for k, name in enumerate(picks):
    it = next(p for p in out["prepped"] if p["file"] == name)
    canvas = it["img"].copy()
    for b in det[det["file"] == name].itertuples():
        p1 = (round(b.x1 * it["scale"]), round(b.y1 * it["scale"]))
        p2 = (round(b.x2 * it["scale"]), round(b.y2 * it["scale"]))
        cv2.rectangle(canvas, p1, p2, COLORS[b[2]], 2)
        cv2.putText(canvas, b[2], (p1[0], max(p1[1] - 4, 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.4, COLORS[b[2]], 1, cv2.LINE_AA)
    ax = fig.add_subplot(gs[0, k])
    ax.imshow(cv2.cvtColor(canvas, cv2.COLOR_BGR2RGB))
    ax.set_title(name)
    ax.axis("off")
ax = fig.add_subplot(gs[1, :2])
ax.bar(counts.index, counts.values)
ax.set_title(f"클래스별 찾은 개수 (정밀도 {precision:.3f}, 재현율 {recall:.3f})")
ax = fig.add_subplot(gs[1, 2:])
ax.bar([n for n, _ in funnel], [v for _, v in funnel])
for x, (_, v) in enumerate(funnel):
    ax.text(x, v + 0.3, str(v), ha="center")
ax.set_ylim(0, funnel[0][1] * 1.2)
ax.set_title("단계별 남은 그림 수")
fig.savefig("out_ch19_report.png", dpi=80)
plt.show()
print("저장: out_ch19_report.png")