track images with git-lfs

.gitignore

@@ -0,0 +1,10 @@
+Local/*
+labels/*
+Adversarial_attack.ipynb
+Drone_VisDrone.ipynb
+AdvTest.ipynb
+*.pyc
+*checkpoint*
+*checkpoint*/
+**/.ipynb_checkpoints/
+*-checkpoint.*

app.py

@@ -0,0 +1,291 @@
+import io
+import numpy as np
+from PIL import Image
+import gradio as gr
+import torch
+from ultralytics import YOLO
+import cv2
+import attacks  # 上面那个 attacks.py，确保和 app.py 在同一目录或可 import 的包路径
+import os, glob
+
+
+# MODEL_PATH = "weights/yolov8s_3cls.pt"
+MODEL_PATH = "weights/fed_model2.pt"
+MODEL_PATH_C = "weights/yolov8s_3cls.pt"
+
+names = ['car', 'van', 'truck']
+imgsz = 640
+
+SAMPLE_DIR = "./images/train"
+SAMPLE_IMAGES = sorted([
+    p for p in glob.glob(os.path.join(SAMPLE_DIR, "*"))
+    if os.path.splitext(p)[1].lower() in [".jpg", ".jpeg", ".png", ".bmp", ".webp"]
+])[:4]  # 只取前4张
+
+# Load ultralytics model (wrapper)
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+yolom = YOLO(MODEL_PATH)  # wrapper
+# yolom_c = YOLO(MODEL_PATH_C)  # wrapper
+# put underlying module to eval on correct device might be needed in attacks functions
+  
+# def run_detection_on_pil(img_pil: Image.Image, eval_model_state, conf: float = 0.45):
+#     """
+#     Use ultralytics wrapper predict to get a visualization image with boxes.
+#     This is inference-only and does not require gradient.
+#     """
+#     # ultralytics accepts numpy array (H,W,3) in RGB, we pass it directly
+#     img = np.array(img_pil)
+#     # use model.predict with verbose=False to avoid prints
+#     eva_model =  yolom if eval_model_state == "yolom" else YOLO(MODEL_PATH_C)
+#     res = eva_model.predict(source=img, conf=conf, imgsz=imgsz, save=False, verbose=False)
+#     r = res[0]
+#     im_out = img.copy()
+#     # Boxes object may be empty
+#     try:
+#         boxes = r.boxes
+#         for box in boxes:
+#             xyxy = box.xyxy[0].cpu().numpy().astype(int)
+#             x1, y1, x2, y2 = map(int, xyxy)
+#             conf_score = float(box.conf[0].cpu().numpy())
+#             cls_id = int(box.cls[0].cpu().numpy())
+#             # label = f"{cls_id}:{conf_score:.2f}"
+#             label = f"{names[cls_id]}:{conf_score:.2f}"
+#             cv2.rectangle(im_out, (x1, y1), (x2, y2), (0,255,0), 2)
+#             cv2.putText(im_out, label, (x1, max(10,y1-5)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 1)
+#     except Exception as e:
+#         # if no boxes or structure unexpected, just return original
+#         pass
+    # return Image.fromarray(im_out)
+def iou(a, b):
+    ax1, ay1, ax2, ay2 = a
+    bx1, by1, bx2, by2 = b
+    iw = max(0, min(ax2, bx2) - max(ax1, bx1))
+    ih = max(0, min(ay2, by2) - max(ay1, by1))
+    inter = iw * ih
+    if inter <= 0:
+        return 0.0
+    area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1)
+    area_b = max(0, bx2 - bx1) * max(0, by2 - by1)
+    return inter / (area_a + area_b - inter + 1e-9)
+    
+# def center_and_diag(b): #IOU足够好 未启用
+#     x1, y1, x2, y2 = b
+#     cx = 0.5 * (x1 + x2); cy = 0.5 * (y1 + y2)
+#     diag = max(1e-9, ((x2 - x1)**2 + (y2 - y1)**2)**0.5)
+#     area = max(0, (x2 - x1)) * max(0, (y2 - y1))
+#     return cx, cy, diag, area
+  
+def run_detection_on_pil(img_pil: Image.Image, eval_model_state, conf: float = 0.45, GT_boxes=None):
+    """
+    推理+可视化。GT_boxes 和返回的 preds 都是:
+      [{'xyxy': (x1,y1,x2,y2), 'cls': int, 'conf': float(optional)}]
+    """
+    import numpy as np, cv2, math
+    from ultralytics import YOLO
+
+    # ---- 1) 推理 ----
+    img = np.array(img_pil)
+    eva_model = yolom if eval_model_state == "yolom" else YOLO(MODEL_PATH_C)
+    res = eva_model.predict(source=img, conf=conf, imgsz=imgsz, save=False, verbose=False)
+    r = res[0]
+    im_out = img.copy()
+
+    # 名称表（尽量稳）
+    names = getattr(r, "names", None)
+    if names is None and hasattr(eva_model, "model") and hasattr(eva_model.model, "names"):
+        names = eva_model.model.names
+
+    # ---- 2) 规整预测框到简单结构 ----
+    preds = []
+    try:
+        bxs = r.boxes
+        if bxs is not None and len(bxs) > 0:
+            for b in bxs:
+                xyxy = b.xyxy[0].detach().cpu().numpy().tolist()
+                x1, y1, x2, y2 = [int(v) for v in xyxy]
+                cls_id = int(b.cls[0].detach().cpu().numpy())
+                conf_score = float(b.conf[0].detach().cpu().numpy())
+                preds.append({'xyxy': (x1, y1, x2, y2), 'cls': cls_id, 'conf': conf_score})
+    except Exception as e:
+        print("collect preds error:", e)
+
+    # ---- 3) IoU 匹配 + 画框 ----
+    IOU_THR = 0.3
+    # CENTER_DIST_RATIO = 0.30    # 中心点距离 / 预测框对角线 <= 0.30 即视为同一目标
+    # AREA_RATIO_THR = 0.25       # 面积比例下限：min(area_p, area_g) / max(...) >= 0.25
+    gt_used = set()
+    
+    for p in preds:
+        color = (0, 255, 0)   # 同类：绿
+        px1, py1, px2, py2 = p['xyxy']
+        pname = names[p['cls']] if (names is not None and p['cls'] in getattr(names, 'keys', lambda: [])()) else (
+                names[p['cls']] if (isinstance(names, (list, tuple)) and 0 <= p['cls'] < len(names)) else str(p['cls'])
+        )
+        label = f"{pname}:{p.get('conf', 0.0):.2f}"
+
+        if GT_boxes != None:
+            # 找 IoU 最高的未用 GT
+            best_j, best_iou = -1, 0.0
+            for j, g in enumerate(GT_boxes):
+                if j in gt_used:
+                    continue
+                i = iou(p['xyxy'], g['xyxy'])
+                if i > best_iou:
+                    best_iou, best_j = i, j
+    
+            # 颜色规则：匹配且同类=绿；匹配但异类=红；
+            if best_iou >= IOU_THR:
+                gt_used.add(best_j)
+                if p['cls'] != int(GT_boxes[best_j]['cls']):
+                    color = (255, 0, 0)   # 异类：红
+
+        cv2.rectangle(im_out, (px1, py1), (px2, py2), color, 2)
+        cv2.putText(im_out, label, (px1, max(10, py1 - 5)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 1)
+
+    return Image.fromarray(im_out), preds
+
+
+def detect_and_attack(image, eval_model_state, attack_mode, eps, alpha, iters, conf):
+    if image is None:
+        return None, None
+
+    pil = Image.fromarray(image.astype('uint8'), 'RGB')
+
+    original_vis, GT_boxes = run_detection_on_pil(pil, eval_model_state, conf=conf, GT_boxes=None) 
+
+    if attack_mode == "none":
+        return original_vis, None
+
+    try:
+        if attack_mode == "fgsm":
+            adv_pil = attacks.fgsm_attack_on_detector(yolom, pil, eps=eps, device=device, imgsz=imgsz)
+        elif attack_mode == "pgd":
+            adv_pil = attacks.pgd_attack_on_detector(yolom, pil, eps=eps, alpha=alpha, iters=iters, device=device, imgsz=imgsz)
+        else:
+            adv_pil = attacks.demo_random_perturbation(pil, eps=eps)
+    except Exception as ex:
+        print("Whitebox attack failed:", ex)
+        adv_pil = attacks.demo_random_perturbation(pil, eps=eps)
+
+    adv_vis, _ = run_detection_on_pil(adv_pil, eval_model_state, conf=conf, GT_boxes=GT_boxes)
+    return original_vis, adv_vis
+
+
+# Gradio UI
+if __name__ == "__main__":
+    title = "Federated Adversarial Attack — FGSM/PGD Demo"
+    desc_html = (
+        "Adversarial examples are generated locally using a "
+        "<strong>client-side</strong> model’s gradients (white-box), then evaluated against the "
+        "<strong>server-side aggregated (FedAvg) central model</strong>. "
+        "If the perturbation transfers, it can "
+        "degrade or alter the FedAvg model’s predictions on the same input image."
+    )
+    with gr.Blocks(title=title) as demo:
+        # 标题居中
+        gr.Markdown(f"""
+            <div>
+              <h1 style='text-align:center;margin-bottom:0.2rem'>{title}</h1>
+              <p style='opacity:0.85'>{desc_html}</p>
+            </div>""")
+
+        with gr.Row():
+            # ===== 左列：两个输入区块 =====
+            with gr.Column(scale=5):
+                # 输入区块 1：上传窗口 & 样例选择 —— 左右并列
+                with gr.Row():
+                    with gr.Column(scale=7):
+                        in_img = gr.Image(type="numpy", label="Input image")
+                    with gr.Column(scale=2):
+                        if SAMPLE_IMAGES:
+                            gr.Examples(
+                                examples=SAMPLE_IMAGES,
+                                inputs=[in_img],
+                                label=f"Select from sample images",
+                                examples_per_page=4,
+                                # run_on_click 默认为 False（只填充，不执行）
+                            )
+
+                # 输入 2：攻击与参数
+                with gr.Accordion("Attack mode", open=True):
+                    attack_mode = gr.Radio(
+                        choices=["none", "fgsm", "pgd", "random noise"],
+                        value="fgsm",
+                        label="",          
+                        show_label=False
+                    )
+                    eps   = gr.Slider(0.0, 0.3, step=0.01, value=0.0314, label="eps")
+                    alpha = gr.Slider(0.001, 0.05, step=0.001, value=0.0078, label="alpha (PGD step)")
+                    iters = gr.Slider(1, 100, step=1, value=10, label="PGD iterations")
+                    conf  = gr.Slider(0.0, 1.0, step=0.01, value=0.45, label="Confidence threshold (live)")
+
+                with gr.Row():
+                    btn_clear  = gr.ClearButton(
+                        components=[in_img, eps, alpha, iters, conf],  # 不清空 attack_mode
+                        value="Clear"
+                    )
+                    btn_submit = gr.Button("Submit", variant="primary")
+
+            # ===== 右列：两个输出区块 =====
+            with gr.Column(scale=5):
+                # 新增：评测模型选择
+                with gr.Row():
+                    eval_choice = gr.Dropdown(
+                        choices=[(f"Client model {MODEL_PATH}", "client"),
+                                 (f"Central model {MODEL_PATH_C}", "central")],
+                        value="client",                 # ★ 初始值为合法 value
+                        label="Evaluation model"
+                    )
+                    
+                    eval_model_state = gr.State(value="yolom")  
+
+                # ★ 合并后的单一回调：规范化下拉值 + 返回(更新后的下拉值, 模型对象)
+                def on_eval_change(val: str):
+                    if isinstance(val, (list, tuple)):
+                        val = val[0] if len(val) else "client"
+                    if val not in ("client", "central"):
+                        val = "client"
+                    model = "yolom" if val == "client" else "yolom_c"
+                    return gr.update(value=val), model
+
+                # 页面加载时同步一次，避免初次为空/不一致
+                demo.load(
+                    fn=on_eval_change,
+                    inputs=eval_choice,
+                    outputs=[eval_choice, eval_model_state]
+                )
+
+                # 仅这一条 change 绑定（删掉你原来那个只写 State 的 change，避免并发覆盖）
+                eval_choice.change(
+                    fn=on_eval_change,
+                    inputs=eval_choice,
+                    outputs=[eval_choice, eval_model_state]
+                )
+                out_orig = gr.Image(label="Original detection")
+                out_adv  = gr.Image(label="After attack detection")
+
+        # Submit：手动运行
+        btn_submit.click(
+            fn=detect_and_attack,
+            inputs=[in_img, eval_model_state, attack_mode, eps, alpha, iters, conf],
+            outputs=[out_orig, out_adv]
+        )
+
+        # 仅 conf 滑块“实时”
+        conf.release(
+            fn=detect_and_attack,
+            inputs=[in_img, eval_model_state, attack_mode, eps, alpha, iters, conf],
+            outputs=[out_orig, out_adv]
+        )
+
+    demo.queue(default_concurrency_limit=2, max_size=20)
+    if os.getenv("SPACE_ID"):
+        demo.launch(
+            server_name="0.0.0.0",
+            server_port=int(os.getenv("PORT", 7860)),
+            show_error=True,
+        )
+    else:
+        demo.launch(share=True)
+
+    

app_old.py

@@ -0,0 +1,197 @@
+import io
+import numpy as np
+from PIL import Image
+import gradio as gr
+import torch
+from ultralytics import YOLO
+import cv2
+import attacks  # 上面那个 attacks.py，确保和 app.py 在同一目录或可 import 的包路径
+import os, glob
+
+
+# MODEL_PATH = "weights/yolov8s_3cls.pt"
+MODEL_PATH = "weights/fed_model2.pt"
+MODEL_PATH_C = "weights/yolov8s_3cls.pt"
+
+names = ['car', 'van', 'truck']
+imgsz = 640
+
+SAMPLE_DIR = "./images/train"
+SAMPLE_IMAGES = sorted([
+    p for p in glob.glob(os.path.join(SAMPLE_DIR, "*"))
+    if os.path.splitext(p)[1].lower() in [".jpg", ".jpeg", ".png", ".bmp", ".webp"]
+])[:4]  # 只取前4张
+
+# Load ultralytics model (wrapper)
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+yolom = YOLO(MODEL_PATH)  # wrapper
+# yolom_c = YOLO(MODEL_PATH_C)  # wrapper
+# put underlying module to eval on correct device might be needed in attacks functions
+  
+def run_detection_on_pil(img_pil: Image.Image, eval_model_state, conf: float = 0.45):
+    """
+    Use ultralytics wrapper predict to get a visualization image with boxes.
+    This is inference-only and does not require gradient.
+    """
+    # ultralytics accepts numpy array (H,W,3) in RGB, we pass it directly
+    img = np.array(img_pil)
+    # use model.predict with verbose=False to avoid prints
+    eva_model =  yolom if eval_model_state == "yolom" else YOLO(MODEL_PATH_C)
+    res = eva_model.predict(source=img, conf=conf, imgsz=imgsz, save=False, verbose=False)
+    r = res[0]
+    im_out = img.copy()
+    # Boxes object may be empty
+    try:
+        boxes = r.boxes
+        for box in boxes:
+            xyxy = box.xyxy[0].cpu().numpy().astype(int)
+            x1, y1, x2, y2 = map(int, xyxy)
+            conf_score = float(box.conf[0].cpu().numpy())
+            cls_id = int(box.cls[0].cpu().numpy())
+            # label = f"{cls_id}:{conf_score:.2f}"
+            label = f"{names[cls_id]}:{conf_score:.2f}"
+            cv2.rectangle(im_out, (x1, y1), (x2, y2), (0,255,0), 2)
+            cv2.putText(im_out, label, (x1, max(10,y1-5)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 1)
+    except Exception as e:
+        # if no boxes or structure unexpected, just return original
+        pass
+    return Image.fromarray(im_out)
+
+def detect_and_attack(image, eval_model_state, attack_mode, eps, alpha, iters, conf):
+    if image is None:
+        return None, None
+
+    pil = Image.fromarray(image.astype('uint8'), 'RGB')
+
+    original_vis = run_detection_on_pil(pil, eval_model_state, conf=conf) 
+
+    if attack_mode == "none":
+        return original_vis, None
+
+    try:
+        if attack_mode == "fgsm":
+            adv_pil = attacks.fgsm_attack_on_detector(yolom, pil, eps=eps, device=device, imgsz=imgsz)
+        elif attack_mode == "pgd":
+            adv_pil = attacks.pgd_attack_on_detector(yolom, pil, eps=eps, alpha=alpha, iters=iters, device=device, imgsz=imgsz)
+        else:
+            adv_pil = attacks.demo_random_perturbation(pil, eps=eps)
+    except Exception as ex:
+        print("Whitebox attack failed:", ex)
+        adv_pil = attacks.demo_random_perturbation(pil, eps=eps)
+
+    adv_vis = run_detection_on_pil(adv_pil, eval_model_state, conf=conf)
+    return original_vis, adv_vis
+
+
+# Gradio UI
+if __name__ == "__main__":
+    title = "Federated Adversarial Attack — FGSM/PGD Demo"
+    desc_html = (
+        "Adversarial examples are generated locally using a "
+        "<strong>client-side</strong> model’s gradients (white-box), then evaluated against the "
+        "<strong>server-side aggregated (FedAvg) central model</strong>. "
+        "If the perturbation transfers, it can "
+        "degrade or alter the FedAvg model’s predictions on the same input image."
+    )
+    with gr.Blocks(title=title) as demo:
+        # 标题居中
+        gr.Markdown(f"""
+            <div>
+              <h1 style='text-align:center;margin-bottom:0.2rem'>{title}</h1>
+              <p style='opacity:0.85'>{desc_html}</p>
+            </div>""")
+
+        with gr.Row():
+            # ===== 左列：两个输入区块 =====
+            with gr.Column(scale=5):
+                # 输入区块 1：上传窗口 & 样例选择 —— 左右并列
+                with gr.Row():
+                    with gr.Column(scale=7):
+                        in_img = gr.Image(type="numpy", label="Input image")
+                    with gr.Column(scale=2):
+                        if SAMPLE_IMAGES:
+                            gr.Examples(
+                                examples=SAMPLE_IMAGES,
+                                inputs=[in_img],
+                                label=f"Select from sample images",
+                                examples_per_page=4,
+                                # run_on_click 默认为 False（只填充，不执行）
+                            )
+
+                # 输入 2：攻击与参数
+                with gr.Accordion("Attack mode", open=True):
+                    attack_mode = gr.Radio(
+                        choices=["none", "fgsm", "pgd", "random noise"],
+                        value="fgsm",
+                        label="",          
+                        show_label=False
+                    )
+                    eps   = gr.Slider(0.0, 0.3, step=0.01, value=0.0314, label="eps")
+                    alpha = gr.Slider(0.001, 0.05, step=0.001, value=0.0078, label="alpha (PGD step)")
+                    iters = gr.Slider(1, 100, step=1, value=10, label="PGD iterations")
+                    conf  = gr.Slider(0.0, 1.0, step=0.01, value=0.45, label="Confidence threshold (live)")
+
+                with gr.Row():
+                    btn_clear  = gr.ClearButton(
+                        components=[in_img, eps, alpha, iters, conf],  # 不清空 attack_mode
+                        value="Clear"
+                    )
+                    btn_submit = gr.Button("Submit", variant="primary")
+
+            # ===== 右列：两个输出区块 =====
+            with gr.Column(scale=5):
+                # 新增：评测模型选择
+                with gr.Row():
+                    eval_choice = gr.Dropdown(
+                        choices=[(f"Client model {MODEL_PATH}", "client"),
+                                 (f"Central model {MODEL_PATH_C}", "central")],
+                        value="client",                 # ★ 初始值为合法 value
+                        label="Evaluation model"
+                    )
+                    
+                    eval_model_state = gr.State(value="yolom")  
+
+                # ★ 合并后的单一回调：规范化下拉值 + 返回(更新后的下拉值, 模型对象)
+                def on_eval_change(val: str):
+                    if isinstance(val, (list, tuple)):
+                        val = val[0] if len(val) else "client"
+                    if val not in ("client", "central"):
+                        val = "client"
+                    model = "yolom" if val == "client" else "yolom_c"
+                    return gr.update(value=val), model
+
+                # 页面加载时同步一次，避免初次为空/不一致
+                demo.load(
+                    fn=on_eval_change,
+                    inputs=eval_choice,
+                    outputs=[eval_choice, eval_model_state]
+                )
+
+                # 仅这一条 change 绑定（删掉你原来那个只写 State 的 change，避免并发覆盖）
+                eval_choice.change(
+                    fn=on_eval_change,
+                    inputs=eval_choice,
+                    outputs=[eval_choice, eval_model_state]
+                )
+                out_orig = gr.Image(label="Original detection")
+                out_adv  = gr.Image(label="After attack detection")
+
+        # Submit：手动运行
+        btn_submit.click(
+            fn=detect_and_attack,
+            inputs=[in_img, eval_model_state, attack_mode, eps, alpha, iters, conf],
+            outputs=[out_orig, out_adv]
+        )
+
+        # 仅 conf 滑块“实时”
+        conf.release(
+            fn=detect_and_attack,
+            inputs=[in_img, eval_model_state, attack_mode, eps, alpha, iters, conf],
+            outputs=[out_orig, out_adv]
+        )
+
+    # demo.queue(concurrency_count=2, max_size=20)
+    demo.launch()
+    # demo.launch(server_name="0.0.0.0", server_port=7860)
+
+    

attacks.py

@@ -0,0 +1,286 @@
+"""
+attacks.py
+
+提供对检测模型（以 YOLOv8/ultralytics 为主）执行 FGSM 与 PGD 的实现。
+
+设计思路与注意事项：
+- 假定我们可以访问到底层的 torch.nn.Module（例如 ultralytics.YOLO 实例的 .model 成员）
+  并能以 tensor 输入直接跑 forward()，得到原始预测张量 (batch, N_preds, C)
+  其中通常 C = 5 + num_classes（bbox4 + obj_conf + class_logits）。
+- 计算 loss: 对每个 anchor/pred，取 obj_conf * max_class_prob 作为该预测的置信度，
+  把全局置信度求和作为被攻击的目标函数；对该目标函数**做最小化**以让检测置信下降。
+  - FGSM: x_adv = x - eps * sign(grad(loss))
+  - PGD: 多步迭代，每步做 x = x - alpha * sign(grad), 并投影到 L_inf 球体：|x-x_orig|<=eps
+- 如果你的 ultralytics 版本/模型封装与假定不同，代码会抛错并提示如何修改。
+"""
+
+from typing import Tuple, Optional
+import torch
+import torch.nn as nn
+import numpy as np
+from PIL import Image
+import torchvision.transforms as T
+import math
+import torch.nn.functional as F
+from typing import Tuple, Dict
+
+# ============= Resize image =====================
+def _get_max_stride(net) -> int:
+    s = getattr(net, "stride", None)
+    if isinstance(s, torch.Tensor):
+        return int(s.max().item())
+    try:
+        return int(max(s))
+    except Exception:
+        return 32  # 兜底
+
+def letterbox_tensor(
+    x: torch.Tensor,
+    *,
+    imgsz: int,
+    stride: int,
+    fill: float = 114.0 / 255.0,
+    scaleup: bool = True
+) -> Tuple[torch.Tensor, Dict]:
+    """
+    x: [1,3,H,W] in [0,1]
+    返回: x_lb, meta （含缩放比例与左右上下 padding 以便反映射）
+    """
+    assert x.ndim == 4 and x.shape[0] == 1
+    _, C, H, W = x.shape
+    if imgsz is None:
+        # 动态设定目标边长：把 max(H,W) 向上取整到 stride 的倍数（更贴近原生自动整形）
+        imgsz = int(math.ceil(max(H, W) / stride) * stride)
+
+    r = min(imgsz / H, imgsz / W)  # 等比缩放比例
+    if not scaleup:
+        r = min(r, 1.0)
+
+    new_w = int(round(W * r))
+    new_h = int(round(H * r))
+
+    # 先等比缩放
+    if (new_h, new_w) != (H, W):
+        x = F.interpolate(x, size=(new_h, new_w), mode="bilinear", align_corners=False)
+
+    # 再 padding 到 (imgsz, imgsz)，保持与 YOLO 一致的对称填充
+    dw = imgsz - new_w
+    dh = imgsz - new_h
+    left, right = dw // 2, dw - dw // 2
+    top, bottom = dh // 2, dh - dh // 2
+
+    x = F.pad(x, (left, right, top, bottom), mode="constant", value=fill)
+
+    meta = {
+        "ratio": r,
+        "pad": (left, top),
+        "resized_shape": (new_h, new_w),
+        "imgsz": imgsz,
+    }
+    return x, meta
+
+def unletterbox_to_original(
+    x_lb: torch.Tensor, meta: Dict, orig_hw: Tuple[int, int]
+) -> torch.Tensor:
+    """
+    把 letterboxed 张量（[1,3,imgsz,imgsz]）反映射回原始 H0,W0 尺寸（去 padding + 反缩放）
+    """
+    assert x_lb.ndim == 4 and x_lb.shape[0] == 1
+    H0, W0 = orig_hw
+    (left, top) = meta["pad"]
+    (h_r, w_r) = meta["resized_shape"]
+
+    # 去 padding（裁出等比缩放后的区域）
+    x_unpad = x_lb[..., top:top + h_r, left:left + w_r]  # [1,3,h_r,w_r]
+
+    # 反缩放到原图大小
+    x_rec = F.interpolate(x_unpad, size=(H0, W0), mode="bilinear", align_corners=False)
+    return x_rec
+
+  
+# ----- basic preprocessing / deprocessing (RGB PIL <-> torch tensor) -----
+_to_tensor = T.Compose([
+    T.ToTensor(),  # float in [0,1], shape C,H,W
+])
+
+_to_pil = T.ToPILImage()
+
+def pil_to_tensor(img_pil: Image.Image, device: torch.device) -> torch.Tensor:
+    """PIL RGB -> float tensor [1,3,H,W] on device"""
+    t = _to_tensor(img_pil).unsqueeze(0).to(device)  # 1,C,H,W
+    t.requires_grad = True
+    return t
+
+def tensor_to_pil(t: torch.Tensor) -> Image.Image:
+    """tensor [1,3,H,W] (0..1) -> PIL RGB"""
+    t = t.detach().cpu().squeeze(0).clamp(0.0, 1.0)
+    return _to_pil(t)
+
+# ----- helper to obtain underlying torch module from ultralytics YOLO wrapper -----
+def get_torch_module_from_ultralytics(model) -> nn.Module:
+    """
+    Try to retrieve an nn.Module that accepts an input tensor and returns raw preds.
+    For ultralytics.YOLO, .model is usually the underlying Detect/Model (nn.Module).
+    """
+    if hasattr(model, "model") and isinstance(model.model, nn.Module):
+        return model.model
+    # Some wrappers nest further; attempt a few common names
+    for attr in ("model", "module", "net", "model_"):
+        if hasattr(model, attr) and isinstance(getattr(model, attr), nn.Module):
+            return getattr(model, attr)
+    raise RuntimeError("无法找到底层 torch.nn.Module。请确保传入的是 ultralytics.YOLO 实例且能访问 model.model。")
+
+# ----- interpret raw model outputs to confidences -----
+def preds_to_confidence_sum(preds: torch.Tensor) -> torch.Tensor:
+    """
+    preds: tensor shape (batch, N_preds, C) or (batch, C, H, W) depending on model.
+    We support the common YOLO format where last dim: [x,y,w,h,obj_conf, class_probs...]
+    Returns scalar: sum of (obj_conf * max_class_prob) over batch and predictions.
+    """
+    if preds is None:
+        raise ValueError("preds is None")
+    # handle shape (batch, N_preds, C)
+    if preds.ndim == 3:
+        # assume last dim: 5 + num_classes
+        if preds.shape[-1] < 6:
+            # can't interpret
+            raise RuntimeError(f"preds last dim too small ({preds.shape[-1]}). Expecting >=6.")
+        obj_conf = preds[..., 4]  # (batch, N)
+        class_probs = preds[..., 5:]  # (batch, N, num_cls)
+        max_class, _ = class_probs.max(dim=-1)  # (batch, N)
+        conf = obj_conf * max_class
+        return conf.sum()
+    # some models output (batch, C, H, W) - flatten
+    if preds.ndim == 4:
+        # try to collapse so that last dim is class
+        b, c, h, w = preds.shape
+        flat = preds.view(b, c, -1).permute(0, 2, 1)  # (batch, N, C)
+        return preds_to_confidence_sum(flat)
+    raise RuntimeError(f"Unhandled preds dimensionality: {preds.shape}")
+
+# ----- core attack implementations -----
+def fgsm_attack_on_detector(
+    model,
+    img_pil: Image.Image,
+    eps: float = 0.03,
+    device: Optional[torch.device] = None,
+    imgsz: Optional[int] = None,  # None=自动对齐到 stride 倍数；也可传 640
+  
+) -> Image.Image:
+    """
+    Perform a single-step FGSM on a detection model (white-box).
+    - model: ultralytics.YOLO wrapper (or anything where get_torch_module_from_ultralytics works)
+    - img_pil: input PIL RGB
+    - eps: max per-pixel perturbation in [0,1] (L_inf)
+    Returns PIL image of adversarial example.
+    """
+    device = device or (torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu"))
+    # get torch module
+    net = get_torch_module_from_ultralytics(model)
+    net = net.to(device).eval()
+    for p in net.parameters():
+      p.requires_grad_(False)   # 建议：避免对参数求梯度
+    
+    # (a) 原图 -> [1,3,H0,W0]，随后先 detach 掉梯度
+    x_orig = pil_to_tensor(img_pil, device)  
+    H0, W0 = x_orig.shape[-2:]
+    x_orig = x_orig.detach()
+    
+    # (b) 可微 letterbox
+    s = _get_max_stride(net)
+    x_lb, meta = letterbox_tensor(x_orig, imgsz=imgsz, stride=s, fill=114/255.0)
+    x_lb = x_lb.clone().detach().requires_grad_(True)
+
+    # (c) 前向与你的损失
+    with torch.enable_grad():
+        preds = net(x_lb)
+        if isinstance(preds, (tuple, list)):
+            tensor_pred = next((p for p in preds if isinstance(p, torch.Tensor) and p.ndim >= 3), None)
+            if tensor_pred is None:
+                raise RuntimeError("模型 forward 返回了 tuple/list，但无法从中找到预测张量。")
+            preds = tensor_pred
+
+        loss = - preds_to_confidence_sum(preds)  
+        loss.backward()
+
+    # (d) FGSM 在 letterboxed 空间施扰
+    # FGSM update: x_adv = x + eps * sign(grad(loss wrt x))
+    with torch.no_grad():
+        adv_lb = (x_lb + eps * x_lb.grad.sign()).clamp(0, 1)
+
+    # 清理（单步可选；PGD循环时必做）
+    x_lb.grad = None
+    net.zero_grad(set_to_none=True)
+
+    # (e) 反映射回原图尺寸
+    adv_orig = unletterbox_to_original(adv_lb, meta, (H0, W0)).detach()
+
+    # (f) 转回 PIL
+    adv_pil = tensor_to_pil(adv_orig)
+    return adv_pil
+
+def pgd_attack_on_detector(
+    model,
+    img_pil: Image.Image,
+    eps: float = 0.03,      # L_inf 半径（输入在[0,1]域）
+    alpha: float = 0.007,   # 步长
+    iters: int = 10,
+    device: Optional[torch.device] = None,
+    imgsz: Optional[int] = None,  # None=自动对齐到 stride 倍数；也可传 640
+):
+    """
+    在 YOLO 的 letterbox 域做 PGD，
+    迭代结束后把对抗样本映回原图大小并返回 PIL。
+    依赖你已实现的: pil_to_tensor, tensor_to_pil, letterbox_tensor, unletterbox_to_original, _get_max_stride
+    """
+    device = device or (torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu"))
+    net = get_torch_module_from_ultralytics(model).to(device).eval()
+
+    # 仅对输入求梯度，冻结参数以省资源
+    for p in net.parameters():
+        p.requires_grad_(False)
+
+    # 原图 -> Tensor（[1,3,H0,W0], [0,1]）
+    x0 = pil_to_tensor(img_pil, device).detach()
+    H0, W0 = x0.shape[-2:]
+
+    # 可微 letterbox（等比缩放 + 对称 pad 到 stride 倍数）
+    s = _get_max_stride(net)
+    x_lb_orig, meta = letterbox_tensor(x0, imgsz=imgsz, stride=s, fill=114/255.0)  # [1,3,S,S]
+    x = x_lb_orig.clone().detach().requires_grad_(True)
+
+    for _ in range(iters):
+        # 前向 + 反向（需要梯度）
+        preds = net(x)
+        if isinstance(preds, (tuple, list)):
+            preds = next((p for p in preds if isinstance(p, torch.Tensor) and p.ndim >= 3), None)
+            if preds is None:
+                raise RuntimeError("模型 forward 返回 tuple/list，但未找到预测张量。")
+        loss = - preds_to_confidence_sum(preds)      # 我们希望置信度总和下��� → 最小化
+        loss.backward()
+
+        # 更新步与投影（不记录计算图）
+        with torch.no_grad():
+            x.add_(alpha * x.grad.sign())
+            # 投影到 L_inf 球: 通过裁剪 delta 更稳
+            delta = (x - x_lb_orig).clamp(-eps, eps)
+            x.copy_((x_lb_orig + delta).clamp(0.0, 1.0))
+
+        # 清理并设置下一步
+        x.grad = None
+        net.zero_grad(set_to_none=True)
+        x.requires_grad_(True)
+
+    # 反映射回原图尺寸
+    adv_orig = unletterbox_to_original(x.detach(), meta, (H0, W0)).detach()
+    return tensor_to_pil(adv_orig)
+
+
+# ----- graceful fallback / demo noise if whitebox impossible -----
+def demo_random_perturbation(img_pil: Image.Image, eps: float = 0.03) -> Image.Image:
+    """Non-gradient demo perturbation used as fallback."""
+    arr = np.asarray(img_pil).astype(np.float32) / 255.0
+    noise = np.sign(np.random.randn(*arr.shape)).astype(np.float32)
+    adv = np.clip(arr + eps * noise, 0.0, 1.0)
+    adv_img = Image.fromarray((adv * 255).astype(np.uint8))
+    return adv_img

federated_utils.py

@@ -0,0 +1,98 @@
+# federated_utils.py
+import copy
+import torch
+import tempfile
+from typing import List, Optional, Tuple
+from ultralytics import YOLO
+import os
+import numpy as np
+
+def average_state_dicts(state_dicts: List[dict]) -> dict:
+    keys = list(state_dicts[0].keys())
+    for sd in state_dicts:
+        assert list(sd.keys()) == keys, "state_dict keys mismatch"
+    avg = {}
+    for k in keys:
+        s = sum(sd[k].cpu().float() for sd in state_dicts)
+        avg[k] = (s / len(state_dicts)).to(state_dicts[0][k].device)
+    return avg
+
+def load_model_state(path: str) -> dict:
+    model = YOLO(path)
+    sd = model.model.state_dict()
+    return {k: v.cpu() for k, v in sd.items()}
+
+def save_state_dict(sd: dict, out_path: str):
+    torch.save(sd, out_path)
+
+def train_client_local(client_data_yaml: str,
+                       init_weights_path: str,
+                       epochs: int = 1,
+                       batch: int = 8,
+                       imgsz: int = 640,
+                       device: Optional[str] = None) -> dict:
+    """用 ultralytics 进行极短本地训练，返回 state_dict（CPU）。"""
+    device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+    model = YOLO(init_weights_path)
+    # 轻量训练；若你有更具体的超参可以改这里
+    model.train(data=client_data_yaml, epochs=epochs, batch=batch, imgsz=imgsz, device=device, save=False, verbose=False)
+    sd = model.model.state_dict()
+    return {k: v.cpu() for k, v in sd.items()}
+
+def run_fedavg_from_checkpoints(client_ckpt_paths: List[str], out_global_path: str) -> str:
+    sds = []
+    for p in client_ckpt_paths:
+        try:
+            tmp = torch.load(p, map_location='cpu')
+            if isinstance(tmp, dict) and 'model' in tmp:
+                sd = tmp['model']
+            else:
+                sd = tmp
+            sd_cpu = {k: v.cpu() for k, v in sd.items()}
+        except Exception:
+            # 如果直接load失败，就用YOLO包装加载
+            model = YOLO(p)
+            sd_cpu = {k: v.cpu() for k, v in model.model.state_dict().items()}
+        sds.append(sd_cpu)
+    avg = average_state_dicts(sds)
+    torch.save(avg, out_global_path)
+    return out_global_path
+
+def run_federated_simulation(client_data_yaml_list: List[str],
+                             init_global_weights: str,
+                             rounds: int = 1,
+                             local_epochs: int = 1,
+                             device: Optional[str] = None,
+                             imgsz: int = 640,
+                             batch: int = 8) -> Tuple[str, List[str]]:
+    """
+    超轻量联邦模拟（同步执行）：
+      for r in rounds:
+        对每个 client: 从当前全局权重做本地小步训练 → 得到 client state_dict
+        平均得到新的全局 → 保存 global_round_r.pt
+    返回：(初始全局路径, [每轮全局路径列表])
+    """
+    device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+    global_state = load_model_state(init_global_weights)
+    round_paths: List[str] = []
+    for r in range(rounds):
+        client_states = []
+        for i, cyaml in enumerate(client_data_yaml_list):
+            # 将当前全局state写临时文件供 YOLO 加载
+            init_tmp = f"__tmp_global_r{r}_c{i}.pt"
+            torch.save(global_state, init_tmp)
+            try:
+                sd_client = train_client_local(
+                    cyaml, init_tmp, epochs=local_epochs, batch=batch, imgsz=imgsz, device=device
+                )
+            except Exception as e:
+                print(f"[WARN] client {i} local train failed: {e}; fallback to global")
+                sd_client = load_model_state(init_tmp)
+            client_states.append(sd_client)
+        # FedAvg
+        new_global = average_state_dicts(client_states)
+        out_path = f"global_round_{r}.pt"
+        save_state_dict(new_global, out_path)
+        round_paths.append(out_path)
+        global_state = new_global
+    return init_global_weights, round_paths

requirements.txt

@@ -0,0 +1,13 @@
+# --- Core ---
+gradio>=4.44.0
+ultralytics>=8.3.0
+torch>=2.2.0
+torchvision>=0.17.0
+
+# --- Image/Array utils ---
+opencv-python
+Pillow
+numpy
+
+# （可选）日志/可视化
+tqdm

weights/fed_model2.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5556943eb4eda78916605d095385ca06dd2947727c65c7171e544ca4eceda630
+size 22515300

weights/yolov8s_3cls.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:abd83b970eba32dc547e661fe0c062a4657952334267b861c74259bd4a25cccc
+size 22513856