hoist-dataset-pipeline/pipeline.py

#!/usr/bin/env python3
"""
起重机挂钩数据集生成全流程
============================
Step 1: 视频抽帧 → Step 2: Engagement筛选 → Step 3: Ready注入
→ Step 4: 质量过滤 → Step 5: 高阈值重推理 → Step 6: 质量排序 → Step 7: CVAT导出

用法:
  python pipeline.py                          # 运行全部步骤
  python pipeline.py --skip-extract           # 跳过抽帧
  python pipeline.py --skip-extract --skip-select  # 跳过多个步骤
"""
import os
import sys
import re
import time
import shutil
import zipfile
import subprocess
import numpy as np
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

from config import *


# ============================================================
# Step 1: 视频抽帧 (FFmpeg NVDEC硬件加速)
# ============================================================
def step1_extract():
    print("=" * 60)
    print("Step 1: 视频抽帧 (FFmpeg NVDEC)")
    print("=" * 60)

    if EXTRACTED_DIR.exists():
        print(f"  已存在 {EXTRACTED_DIR}，跳过抽帧")
        return

    videos = []
    for mp4 in VIDEO_DIR.rglob('*.mp4'):
        parts = mp4.parts
        try:
            cam_idx = parts.index(VIDEO_DIR_NAME)
            cam_id = parts[cam_idx + 1]
            cam_name = f'cam{cam_id.zfill(2)}'
            date = parts[cam_idx + 2]
        except (IndexError, ValueError):
            continue
        videos.append((mp4, cam_name, date))

    print(f"  找到 {len(videos)} 个视频")
    EXTRACTED_DIR.mkdir(parents=True, exist_ok=True)
    total_frames = 0

    for i, (mp4, cam_name, date) in enumerate(videos, 1):
        out_dir = EXTRACTED_DIR / cam_name
        out_dir.mkdir(exist_ok=True)
        stem = mp4.stem
        pattern = str(out_dir / f'{cam_name}_{date}_{stem}_%06d.jpg')

        cmd = ['ffmpeg', '-hide_banner', '-loglevel', 'error',
               '-hwaccel', 'nvdec', '-c:v', 'hevc_cuvid',
               '-i', str(mp4),
               '-qmin', '1', '-qmax', '1', '-vsync', '0', pattern]

        print(f"  [{i}/{len(videos)}] {cam_name}/{date}/{stem}...", end=' ', flush=True)
        t0 = time.time()
        subprocess.run(cmd, check=True)

        n_frames = len(list(out_dir.glob(f'{cam_name}_{date}_{stem}_*.jpg')))
        for j in range(1, n_frames + 1):
            lbl = out_dir / f'{cam_name}_{date}_{stem}_{j:06d}.txt'
            if not lbl.exists():
                lbl.touch()

        elapsed = time.time() - t0
        total_frames += n_frames
        print(f"{n_frames} frames in {elapsed:.0f}s")

    print(f"\n  抽帧完成: {total_frames} 帧")


# ============================================================
# Step 2: Engagement评分 + 去重 + 排序选取
# ============================================================
def _score_and_pos(dets):
    """评分：hoist+ear共现=0.7, 仅ear=0.4, 中心距离奖励, 多hoist惩罚"""
    if len(dets) == 0:
        return 0
    has_hoist = np.any(dets[:, 0] == 0)
    has_ear = np.any(dets[:, 0] == 1)
    n_hoist = int(np.sum(dets[:, 0] == 0))

    score = 0.7 if (has_hoist and has_ear) else (0.4 if has_ear else 0.0)

    avg_xc = np.mean(dets[:, 1])
    avg_yc = np.mean(dets[:, 2])
    center_dist = ((avg_xc - 0.5)**2 + (avg_yc - 0.5)**2) ** 0.5
    score += max(0, 0.3 - center_dist) * 0.5
    if n_hoist > 1:
        score -= (n_hoist - 1) * 0.1
    return max(0, score)


def _fix_duplicates(labels):
    """去重：ear保留最小框，hoist保留最大框"""
    for cls_id, keep_func in [(1, 'min'), (0, 'max')]:
        cls_mask = labels[:, 0] == cls_id
        if np.sum(cls_mask) <= 1:
            continue
        cls_dets = labels[cls_mask]
        areas = cls_dets[:, 3] * cls_dets[:, 4]
        keep_idx = np.argmin(areas) if keep_func == 'min' else np.argmax(areas)
        keep_mask = np.ones(len(labels), dtype=bool)
        cls_indices = np.where(cls_mask)[0]
        for ri in range(len(cls_dets)):
            if ri != keep_idx:
                keep_mask[cls_indices[ri]] = False
        labels = labels[keep_mask]
    return labels


def step2_select():
    print("\n" + "=" * 60)
    print("Step 2: Engagement评分 + 去重 + 选取engaged帧")
    print("=" * 60)

    if OUTPUT_DIR.exists():
        shutil.rmtree(str(OUTPUT_DIR))

    cameras = sorted([d.name for d in EXTRACTED_IMG_DIR.iterdir() if d.is_dir()])
    all_kept = {}

    for cam_name in cameras:
        t0 = time.time()
        lbl_dir = EXTRACTED_LBL_DIR / cam_name

        # 读取所有标签
        labels = {}
        with os.scandir(str(lbl_dir)) as it:
            for entry in it:
                if not entry.name.endswith('.txt'):
                    continue
                with open(entry.path) as f:
                    data = f.read().strip()
                stem = entry.name[:-4]
                if data:
                    rows = []
                    for line in data.split('\n'):
                        parts = line.split()
                        if len(parts) >= 5:
                            rows.append([float(x) for x in parts[:5]])
                    labels[stem] = np.array(rows) if rows else np.zeros((0, 5))
                else:
                    labels[stem] = np.zeros((0, 5))

        # 评分
        scores = {name: _score_and_pos(dets) for name, dets in labels.items()}

        # 去重
        for name, dets in labels.items():
            if len(dets) > 0:
                labels[name] = _fix_duplicates(dets)

        # 按分数排序
        ranked = sorted(scores.keys(), key=lambda n: scores[n], reverse=True)

        # 筛选
        selected = []
        last_seq = -999
        for name in ranked:
            if scores[name] < SCORE_THRESH:
                continue
            dets = labels[name]
            if not np.any((dets[:, 0] == 0) | (dets[:, 0] == 1)):
                continue
            m = re.search(r'_(\d{6})$', name)
            seq = int(m.group(1)) if m else 0
            if abs(seq - last_seq) < SEQ_GAP or seq == last_seq:
                continue
            selected.append(name)
            last_seq = seq

        # 创建输出目录 + 符号链接
        out_img = OUTPUT_DIR / 'images' / cam_name
        out_lbl = OUTPUT_DIR / 'labels' / cam_name
        out_img.mkdir(parents=True, exist_ok=True)
        out_lbl.mkdir(parents=True, exist_ok=True)

        img_dir = EXTRACTED_IMG_DIR / cam_name
        for name in selected:
            si = img_dir / f'{name}.jpg'
            sl = EXTRACTED_LBL_DIR / cam_name / f'{name}.txt'
            di = out_img / f'czzt_{name}.jpg'
            dl = out_lbl / f'czzt_{name}.txt'
            if si.exists() and not di.exists():
                os.symlink(str(si.resolve()), str(di))
            if sl.exists() and not dl.exists():
                os.symlink(str(sl.resolve()), str(dl))

        all_kept[cam_name] = len(selected)
        print(f"  {cam_name}: {len(selected)} images in {time.time() - t0:.1f}s")

    total = sum(all_kept.values())
    print(f"\n  Step 2 完成: {total} images")
    return all_kept


# ============================================================
# Step 3: 注入 ready 标签 (YOLOv5推理)
# ============================================================
def _calc_iou(boxA, boxB):
    x1 = max(boxA[0], boxB[0]); y1 = max(boxA[1], boxB[1])
    x2 = min(boxA[2], boxB[2]); y2 = min(boxA[3], boxB[3])
    inter = max(0, x2-x1) * max(0, y2-y1)
    areaA = (boxA[2]-boxA[0]) * (boxA[3]-boxA[1])
    areaB = (boxB[2]-boxB[0]) * (boxB[3]-boxB[1])
    union = areaA + areaB - inter
    return inter / union if union > 0 else 0


def step3_inject_ready(cam_name):
    import torch
    import cv2

    img_dir = OUTPUT_DIR / 'images' / cam_name
    lbl_dir = OUTPUT_DIR / 'labels' / cam_name
    if not img_dir.exists():
        return 0

    images = sorted(img_dir.glob('czzt_*.jpg'))
    if not images:
        return 0

    sys.path.insert(0, YOLOV5_DIR)
    from models.common import DetectMultiBackend
    from utils.general import non_max_suppression, scale_boxes
    from utils.torch_utils import select_device

    device = select_device('')
    model = DetectMultiBackend(MODEL_PATH, device=device)
    model.eval()

    injected = 0
    for img_path in images:
        lbl_file = lbl_dir / f'{img_path.stem}.txt'

        # 读取已有标签
        existing = []
        if lbl_file.exists():
            with open(lbl_file) as f:
                for line in f:
                    parts = line.strip().split()
                    if len(parts) >= 5:
                        existing.append([int(parts[0])] + [float(x) for x in parts[1:]])

        # 需要同时有hoist和ear才注入ready
        if not (any(r[0] == 0 for r in existing) and any(r[0] == 1 for r in existing)):
            continue

        # 推理
        img_bgr = cv2.imread(str(img_path))
        if img_bgr is None:
            continue
        img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
        h0, w0 = img_rgb.shape[:2]
        img_resized = cv2.resize(img_rgb, (IMG_WIDTH, IMG_HEIGHT))
        img_tensor = img_resized.transpose(2, 0, 1)
        img_tensor = np.ascontiguousarray(img_tensor)
        img_tensor = torch.from_numpy(img_tensor).to(device).float() / 255.0
        if img_tensor.ndimension() == 3:
            img_tensor = img_tensor.unsqueeze(0)

        pred = model(img_tensor)[0]
        pred = non_max_suppression(pred, READY_CONF_THRESH, READY_IOU_THRESH, classes=[2])

        if len(pred) == 0 or len(pred[0]) == 0:
            continue

        dets = pred[0]
        dets[:, :4] = scale_boxes(img_tensor.shape[2:], dets[:, :4], (h0, w0)).round()
        areas = (dets[:, 2] - dets[:, 0]) * (dets[:, 3] - dets[:, 1])

        # 收集ear框用于IoU过滤
        ear_boxes = []
        for r in existing:
            if r[0] == 1:
                ex, ey, ew, eh = r[1], r[2], r[3], r[4]
                ear_boxes.append([(ex-ew/2)*w0, (ey-eh/2)*h0, (ex+ew/2)*w0, (ey+eh/2)*h0])

        # 找最大的不与ear重叠的ready
        sorted_idx = torch.argsort(areas, descending=True)
        best = None
        for idx in sorted_idx:
            det = dets[idx]
            rbox = [det[0].item(), det[1].item(), det[2].item(), det[3].item()]
            if all(_calc_iou(rbox, ebox) <= READY_EAR_IOU_THRESH for ebox in ear_boxes):
                best = det
                break

        if best is None:
            continue

        xc = (best[0] + best[2]) / 2 / w0
        yc = (best[1] + best[3]) / 2 / h0
        w = (best[2] - best[0]) / w0
        h = (best[3] - best[1]) / h0

        with open(lbl_file, 'a') as f:
            f.write(f'2 {xc:.6f} {yc:.6f} {w:.6f} {h:.6f}\n')
        injected += 1

    return injected


def step3_inject():
    print("\n" + "=" * 60)
    print("Step 3: 注入 ready 标签 (YOLOv5推理)")
    print("=" * 60)

    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])
    total_injected = 0
    for cam_name in cameras:
        n = step3_inject_ready(cam_name)
        total_injected += n
        print(f"  {cam_name}: injected ready to {n} images")
    print(f"\n  Step 3 完成: {total_injected} ready labels injected")


# ============================================================
# Step 4: 质量过滤
# ============================================================
def step4_filter():
    print("\n" + "=" * 60)
    print("Step 4: 质量过滤")
    print("=" * 60)

    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])
    total_keep = 0

    for cam_name in cameras:
        img_dir = OUTPUT_DIR / 'images' / cam_name
        lbl_dir = OUTPUT_DIR / 'labels' / cam_name
        if not lbl_dir.exists():
            continue

        discard_ratio = discard_flat = discard_both = keep = 0

        for lbl_file in list(lbl_dir.glob('czzt_*.txt')):
            with open(lbl_file) as f:
                lines = f.readlines()

            hoist_areas, ready_areas, ear_aspects = [], [], []
            for line in lines:
                parts = line.strip().split()
                if len(parts) < 5:
                    continue
                cls = int(parts[0])
                w, h = float(parts[3]), float(parts[4])
                if cls == 0:
                    hoist_areas.append(w * h)
                elif cls == 2:
                    ready_areas.append(w * h)
                elif cls == 1 and h > 0:
                    ear_aspects.append(w / h)

            fail_ratio = (hoist_areas and ready_areas and
                          max(ready_areas) / max(hoist_areas) >= READY_HOIST_RATIO_THRESH)
            fail_flat = (ear_aspects and all(a >= EAR_FLAT_ASPECT_THRESH for a in ear_aspects))

            if fail_ratio and fail_flat:
                discard_both += 1
            elif fail_ratio:
                discard_ratio += 1
            elif fail_flat:
                discard_flat += 1
            else:
                keep += 1
                continue

            img_file = img_dir / (lbl_file.stem + '.jpg')
            if img_file.exists():
                img_file.unlink()
            lbl_file.unlink()

        total_keep += keep
        print(f"  {cam_name}: keep={keep}, discard_ratio={discard_ratio}, discard_flat={discard_flat}, discard_both={discard_both}")

    print(f"\n  Step 4 完成: {total_keep} images after filtering")


# ============================================================
# Step 5: 重新YOLO推理 (高置信度阈值, 清除重叠框)
# ============================================================
def step5_reinfer():
    import torch
    import cv2

    print("\n" + "=" * 60)
    print(f"Step 5: 重新YOLO推理 (conf>={REINFER_CONF}, iou={REINFER_IOU})")
    print("=" * 60)

    sys.path.insert(0, YOLOV5_DIR)
    from models.common import DetectMultiBackend
    from utils.general import non_max_suppression, scale_boxes
    from utils.torch_utils import select_device

    device = select_device('')
    model = DetectMultiBackend(MODEL_PATH, device=device)
    model.eval()

    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])
    total_reinferred = 0

    for cam_name in cameras:
        img_dir = OUTPUT_DIR / 'images' / cam_name
        lbl_dir = OUTPUT_DIR / 'labels' / cam_name
        if not img_dir.exists():
            continue

        images = sorted(img_dir.glob('czzt_*.jpg'))
        if not images:
            continue

        reinferred = 0
        for img_path in images:
            img_bgr = cv2.imread(str(img_path))
            if img_bgr is None:
                continue
            img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
            h0, w0 = img_rgb.shape[:2]
            img_resized = cv2.resize(img_rgb, (IMG_WIDTH, IMG_HEIGHT))
            img_tensor = img_resized.transpose(2, 0, 1)
            img_tensor = np.ascontiguousarray(img_tensor)
            img_tensor = torch.from_numpy(img_tensor).to(device).float() / 255.0
            if img_tensor.ndimension() == 3:
                img_tensor = img_tensor.unsqueeze(0)

            pred = model(img_tensor)[0]
            pred = non_max_suppression(pred, REINFER_CONF, REINFER_IOU)

            lbl_file = lbl_dir / f'{img_path.stem}.txt'

            if len(pred) == 0 or len(pred[0]) == 0:
                lbl_file.write_text('')
                reinferred += 1
                continue

            dets = pred[0]
            dets[:, :4] = scale_boxes(img_tensor.shape[2:], dets[:, :4], (h0, w0)).round()

            with open(lbl_file, 'w') as f:
                for det in dets:
                    cls_id = int(det[5])
                    if cls_id not in CLASSES:
                        continue
                    xc = (det[0] + det[2]) / 2 / w0
                    yc = (det[1] + det[3]) / 2 / h0
                    w = (det[2] - det[0]) / w0
                    h = (det[3] - det[1]) / h0
                    f.write(f'{cls_id} {xc:.6f} {yc:.6f} {w:.6f} {h:.6f}\n')
            reinferred += 1

        total_reinferred += reinferred
        print(f"  {cam_name}: reinferred {reinferred} images")

    print(f"\n  Step 5 完成: {total_reinferred} images reinferred")


# ============================================================
# Step 6: 质量排序取top MAX_PER_CAM
# ============================================================
def step6_select_top():
    print("\n" + "=" * 60)
    print(f"Step 6: 质量排序取top {MAX_PER_CAM}")
    print("=" * 60)

    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])

    for cam_name in cameras:
        img_dir = OUTPUT_DIR / 'images' / cam_name
        lbl_dir = OUTPUT_DIR / 'labels' / cam_name
        if not img_dir.exists():
            continue

        images = sorted(img_dir.glob('czzt_*.jpg'))
        if len(images) <= MAX_PER_CAM:
            print(f"  {cam_name}: {len(images)} images (<= {MAX_PER_CAM}, no trimming)")
            continue

        def compute_quality(img_path):
            try:
                import cv2
                img = cv2.imread(str(img_path), cv2.IMREAD_GRAYSCALE)
                if img is None:
                    return 0
                brightness = np.mean(img) / 255.0
                laplacian = cv2.Laplacian(img, cv2.CV_64F)
                sharpness = laplacian.var() / 1000.0
                return brightness * QUALITY_BRIGHTNESS_WEIGHT + min(sharpness, 1.0) * QUALITY_SHARPNESS_WEIGHT
            except Exception:
                return 0

        quality = {}
        with ThreadPoolExecutor(max_workers=16) as pool:
            for img_path in images:
                quality[img_path.name] = compute_quality(img_path)

        ranked = sorted(quality.keys(), key=lambda n: quality[n], reverse=True)
        keep_names = set(ranked[:MAX_PER_CAM])

        removed = 0
        for img_path in images:
            if img_path.name not in keep_names:
                img_path.unlink()
                lbl = lbl_dir / (img_path.stem + '.txt')
                if lbl.exists():
                    lbl.unlink()
                removed += 1

        print(f"  {cam_name}: kept {len(keep_names)}, removed {removed}")


# ============================================================
# Step 7: 生成 CVAT ZIP
# ============================================================
def step7_generate_cvat():
    print("\n" + "=" * 60)
    print("Step 7: 生成 CVAT ZIP")
    print("=" * 60)

    CVAT_ZIP_DIR.mkdir(parents=True, exist_ok=True)
    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])

    for cam_name in cameras:
        img_dir = OUTPUT_DIR / 'images' / cam_name
        lbl_dir = OUTPUT_DIR / 'labels' / cam_name

        img_names = sorted([f.name for f in img_dir.glob('czzt_*.jpg')])
        if not img_names:
            print(f"  {cam_name}: 0 images, skipping")
            continue

        zip_path = CVAT_ZIP_DIR / f'{cam_name}.zip'
        obj_names_str = '\n'.join(CVAT_CLASSES) + '\n'
        obj_data = f'classes = {len(CVAT_CLASSES)}\ntrain = data/train.txt\nnames = data/obj.names\nbackup = backup/\n'
        train_lines = '\n'.join(f'data/obj_train_data/data/obj_train_data/{n}' for n in img_names) + '\n'

        with zipfile.ZipFile(str(zip_path), 'w', zipfile.ZIP_STORED) as zf:
            zf.writestr('obj.names', obj_names_str)
            zf.writestr('obj.data', obj_data)
            zf.writestr('train.txt', train_lines)
            for img_name in img_names:
                zf.write(str(img_dir / img_name), f'obj_train_data/data/obj_train_data/{img_name}')
                stem = img_name[:-4]
                lbl_file = lbl_dir / f'{stem}.txt'
                if lbl_file.exists():
                    zf.write(str(lbl_file), f'obj_train_data/data/obj_train_data/{stem}.txt')

        print(f"  {cam_name}: {len(img_names)} images -> {zip_path.name}")

    print(f"\n  Step 7 完成: CVAT zips in {CVAT_ZIP_DIR}")


# ============================================================
# 最终统计
# ============================================================
def print_stats():
    print("\n" + "=" * 60)
    print("最终统计")
    print("=" * 60)

    cameras = sorted([d.name for d in (OUTPUT_DIR / 'images').iterdir() if d.is_dir()])
    total = 0
    label_counts = {0: 0, 1: 0, 2: 0}

    for cam_name in cameras:
        img_dir = OUTPUT_DIR / 'images' / cam_name
        lbl_dir = OUTPUT_DIR / 'labels' / cam_name
        n_img = len(list(img_dir.glob('czzt_*.jpg'))) if img_dir.exists() else 0
        n_lbl = {0: 0, 1: 0, 2: 0}
        if lbl_dir.exists():
            for lf in lbl_dir.glob('czzt_*.txt'):
                with open(lf) as f:
                    for line in f:
                        parts = line.strip().split()
                        if len(parts) >= 5:
                            c = int(parts[0])
                            if c in n_lbl:
                                n_lbl[c] += 1
                                label_counts[c] += 1
        total += n_img
        print(f"  {cam_name}: {n_img} images, hoist={n_lbl[0]}, ear={n_lbl[1]}, ready={n_lbl[2]}")

    print(f"\n  总计: {total} images")
    print(f"  hoist(0): {label_counts[0]}")
    print(f"  ear(1): {label_counts[1]}")
    print(f"  ready(2): {label_counts[2]}")


# ============================================================
# Main
# ============================================================
def main():
    skip = set(sys.argv[1:])

    if '--skip-extract' not in skip:
        step1_extract()
    if '--skip-select' not in skip:
        step2_select()
    if '--skip-inject' not in skip:
        step3_inject()
    if '--skip-filter' not in skip:
        step4_filter()
    if '--skip-reinfer' not in skip:
        step5_reinfer()
    if '--skip-select-top' not in skip:
        step6_select_top()
    if '--skip-zip' not in skip:
        step7_generate_cvat()

    print_stats()


if __name__ == '__main__':
    main()