2. 作物病害数据集清洗与类别平衡

清洗目标与判定口径

本案例处理的是“已采集但未清洗”的作物病害图像集。假设原始目录结构为 raw/类别名/*.jpg，类别包括 健康、早疫病、晚疫病、叶霉病。清洗需要同时完成三件事：去除重复图像、过滤模糊图像、统计并平衡类别分布。为了让结果可复现，先固定判定口径：

• 重复判定：对图像计算 64 位感知哈希（dHash），汉明距离 ≤ 5 视为重复，保留文件体积更大的一张。
• 模糊判定：灰度图拉普拉斯方差低于阈值 80 视为模糊；阈值需按实际设备分辨率校准，本例为示例值。
• 平衡目标：以清洗后最大类别样本数为基准，各类别允许偏差 ±10%；不足者过采样（复制并加轻微增强），超出者欠采样（随机抽取）。

清洗脚本实现

下面脚本只依赖 opencv-python、numpy、pandas，按“扫描 → 判重 → 判模糊 → 统计 → 平衡”顺序执行，并把每一步的剔除原因写入报告，便于回溯。示例参数均为教学假设值，实际使用前应先用少量样本校准模糊阈值。

import os, shutil, random, csv
from pathlib import Path
import cv2
import numpy as np
import pandas as pd

RAW_DIR = Path("raw")
CLEAN_DIR = Path("clean")
REPORT_DIR = Path("reports")
BLUR_THRESHOLD = 80.0      # 示例阈值，需按设备校准
HAMMING_MAX = 5            # 判重距离
SEED = 42
random.seed(SEED)

def dhash(img, size=8):
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    resized = cv2.resize(gray, (size + 1, size))
    diff = resized[:, 1:] > resized[:, :-1]
    bits = np.packbits(diff.flatten())
    return bits

def hamming(a, b):
    return int(np.unpackbits(a ^ b).sum())

def laplacian_var(img):
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    return float(cv2.Laplacian(gray, cv2.CV_64F).var())

records = []
for cls_dir in sorted(RAW_DIR.iterdir()):
    if not cls_dir.is_dir():
        continue
    for img_path in sorted(cls_dir.glob("*.jpg")):
        img = cv2.imread(str(img_path))
        if img is None:
            records.append({"path": str(img_path), "class": cls_dir.name,
                            "status": "unreadable", "reason": "decode_fail"})
            continue
        records.append({
            "path": str(img_path), "class": cls_dir.name,
            "size": img_path.stat().st_size,
            "hash": dhash(img), "blur": laplacian_var(img),
            "status": "keep", "reason": ""
        })

df = pd.DataFrame(records)
df = df[df["status"] == "keep"].copy()

# 1) 重复检测：同类内比较，保留体积更大者
for cls, group in df.groupby("class"):
    kept = []
    for idx, row in group.iterrows():
        dup = False
        for k_idx in kept:
            if hamming(row["hash"], df.at[k_idx, "hash"]) <= HAMMING_MAX:
                dup = True
                if row["size"] > df.at[k_idx, "size"]:
                    df.at[k_idx, "status"] = "drop"
                    df.at[k_idx, "reason"] = "duplicate"
                    kept.remove(k_idx)
                    kept.append(idx)
                else:
                    df.at[idx, "status"] = "drop"
                    df.at[idx, "reason"] = "duplicate"
                break
        if not dup:
            kept.append(idx)

# 2) 模糊过滤
mask_blur = (df["status"] == "keep") & (df["blur"] < BLUR_THRESHOLD)
df.loc[mask_blur, "status"] = "drop"
df.loc[mask_blur, "reason"] = "blur"

# 3) 统计
kept_df = df[df["status"] == "keep"]
counts = kept_df["class"].value_counts().sort_index()
print("清洗后类别分布：\n", counts)

# 4) 平衡：以最大类为基准，±10%
target = int(counts.max())
low, high = int(target * 0.9), int(target * 1.1)
plan = []
for cls, cnt in counts.items():
    if cnt < low:
        plan.append((cls, "oversample", low - cnt))
    elif cnt > high:
        plan.append((cls, "undersample", cnt - high))
    else:
        plan.append((cls, "keep", 0))

# 5) 落盘
if CLEAN_DIR.exists():
    shutil.rmtree(CLEAN_DIR)
for cls in counts.index:
    (CLEAN_DIR / cls).mkdir(parents=True, exist_ok=True)

for _, row in kept_df.iterrows():
    shutil.copy2(row["path"], CLEAN_DIR / row["class"] / Path(row["path"]).name)

for cls, action, delta in plan:
    files = sorted((CLEAN_DIR / cls).glob("*.jpg"))
    if action == "oversample":
        for i in range(delta):
            src = files[i % len(files)]
            img = cv2.imread(str(src))
            aug = cv2.flip(img, 1) if i % 2 == 0 else cv2.convertScaleAbs(img, alpha=1.1, beta=10)
            cv2.imwrite(str(CLEAN_DIR / cls / f"aug_{i:04d}_{src.name}"), aug)
    elif action == "undersample":
        drop = random.sample(files, delta)
        for f in drop:
            f.unlink()

REPORT_DIR.mkdir(exist_ok=True)
df.to_csv(REPORT_DIR / "clean_report.csv", index=False)
pd.DataFrame(plan, columns=["class", "action", "delta"]).to_csv(
    REPORT_DIR / "balance_plan.csv", index=False)
print("平衡方案：\n", plan)

结果判读与验收

脚本运行后应检查三份产物：reports/clean_report.csv 记录每张图的 status 与 reason；reports/balance_plan.csv 记录每个类别的处理动作与调整量；clean/ 为最终训练集。验收标准如下：