add targeted attack- classify all obj as truck

app.py

@@ -25,37 +25,7 @@ SAMPLE_IMAGES = sorted([
 # 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
@@ -158,9 +128,9 @@ def detect_and_attack(image, eval_model_state, attack_mode, eps, alpha, iters, c
 
     try:
         if attack_mode == "fgsm":
-            adv_pil = attacks.fgsm_attack_on_detector(yolom, pil, eps=eps, device=device, imgsz=imgsz)
+            adv_pil = attacks.fgsm_attack_on_detector(yolom, pil, eps=eps, device=device, imgsz=imgsz, gt_xywh=GT_boxes)
         elif attack_mode == "pgd":
-            adv_pil = attacks.pgd_attack_on_detector(yolom, pil, eps=eps, alpha=alpha, iters=iters, device=device, imgsz=imgsz)
+            adv_pil = attacks.pgd_attack_on_detector(yolom, pil, eps=eps, alpha=alpha, iters=iters, device=device, imgsz=imgsz, gt_xywh=GT_boxes)
         else:
             adv_pil = attacks.demo_random_perturbation(pil, eps=eps)
     except Exception as ex:
@@ -177,7 +147,7 @@ if __name__ == "__main__":
     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>. "
+        "<strong>server-side aggregated (FedAvg) global model</strong>. "
         "If the perturbation transfers, it can "
         "degrade or alter the FedAvg model’s predictions on the same input image."
     )
@@ -232,7 +202,7 @@ if __name__ == "__main__":
                 with gr.Row():
                     eval_choice = gr.Dropdown(
                         choices=[(f"Client model {MODEL_PATH}", "client"),
-                                 (f"Central model {MODEL_PATH_C}", "central")],
+                                 (f"Global model {MODEL_PATH_C}", "global")],
                         value="client",                 # ★ 初始值为合法 value
                         label="Evaluation model"
                     )
@@ -243,7 +213,7 @@ if __name__ == "__main__":
                 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"):
+                    if val not in ("client", "global"):
                         val = "client"
                     model = "yolom" if val == "client" else "yolom_c"
                     return gr.update(value=val), model
@@ -286,6 +256,6 @@ if __name__ == "__main__":
             show_error=True,
         )
     else:
-        demo.launch(share=True)
+        demo.launch()
 
     

attacks.py

@@ -131,32 +131,173 @@ def get_torch_module_from_ultralytics(model) -> nn.Module:
     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}")
+
+
+def _ensure_bcn(preds):
+    assert preds.ndim == 3
+    B, C1, C2 = preds.shape
+    if C1 - 4 > 0 and C2 >= 1000:   # [B, 4+nc, N]
+        return preds
+    if C2 - 4 > 0 and C1 >= 1000:   # [B, N, 4+nc]
+        return preds.permute(0, 2, 1).contiguous()
+    return preds
+
+def _xywh_to_xyxy(xywh):
+    x,y,w,h = xywh.unbind(-1)
+    return torch.stack([x-w/2, y-h/2, x+w/2, y+h/2], dim=-1)
+
+def _xyxy_to_xywh(xyxy):
+    x1,y1,x2,y2 = xyxy.unbind(-1)
+    cx = (x1+x2)/2; cy = (y1+y2)/2
+    w  = (x2-x1).clamp(min=0); h = (y2-y1).clamp(min=0)
+    return torch.stack([cx,cy,w,h], dim=-1)
+
+def _map_xyxy_to_letterbox(xyxy_tensor, meta):
+    if meta is None:
+        return xyxy_tensor
+    r = meta.get('ratio', meta.get('scale', (1.0, 1.0)))
+    p = meta.get('pad', (0.0, 0.0))
+    if isinstance(r, (int, float)):
+        r = (float(r), float(r))
+    rx, ry = float(r[0]), float(r[1])
+    px, py = float(p[0]), float(p[1])
+    x1 = xyxy_tensor[:, 0] * rx + px
+    y1 = xyxy_tensor[:, 1] * ry + py
+    x2 = xyxy_tensor[:, 2] * rx + px
+    y2 = xyxy_tensor[:, 3] * ry + py
+    return torch.stack([x1, y1, x2, y2], dim=-1)
+
+def _iou_xyxy(b_xyxy, g_xyxy):
+    N, M = b_xyxy.size(0), g_xyxy.size(0)
+    b = b_xyxy[:, None, :].expand(N, M, 4)
+    g = g_xyxy[None, :, :].expand(N, M, 4)
+    inter_x1 = torch.maximum(b[...,0], g[...,0])
+    inter_y1 = torch.maximum(b[...,1], g[...,1])
+    inter_x2 = torch.minimum(b[...,2], g[...,2])
+    inter_y2 = torch.minimum(b[...,3], g[...,3])
+    inter_w  = (inter_x2 - inter_x1).clamp(min=0)
+    inter_h  = (inter_y2 - inter_y1).clamp(min=0)
+    inter    = inter_w * inter_h
+    area_b   = (b[...,2]-b[...,0]).clamp(min=0) * (b[...,3]-b[...,1]).clamp(min=0)
+    area_g   = (g[...,2]-g[...,0]).clamp(min=0) * (g[...,3]-g[...,1]).clamp(min=0)
+    return inter / (area_b + area_g - inter + 1e-9)
+
+def _gt_list_to_xyxy_tensor(gt_list, device, meta=None):
+    if not gt_list:
+        return torch.empty(0, 4, device=device, dtype=torch.float32)
+    xyxy = torch.tensor([b['xyxy'] for b in gt_list], dtype=torch.float32, device=device)
+    return _map_xyxy_to_letterbox(xyxy, meta)
+
+def preds_to_targeted_loss(
+    preds,                  # [B,4+nc,N] 或 [B,N,4+nc]；类别部分最好是 logits
+    target_cls: int,
+    gt_xywh,                # 这里直接支持 list[{'xyxy':..., 'cls':..., 'conf':...}]
+    topk: int = 20,
+    kappa: float = 0.1,
+    lambda_margin: float = 1.0,
+    lambda_keep: float = 0.2,
+    lambda_target: float = 0.0,    # 新增：恢复 -p_t.mean() 这项
+    debug: bool = False,
+    meta: dict | None = None,      # 若 GT 是原图坐标，传入 letterbox 的 meta
+):
+    preds = _ensure_bcn(preds)
+    B, C, N = preds.shape
+    nc = C - 4
+    assert 0 <= target_cls < nc
+
+    # 解析 GT（list -> tensor in letterbox coords）
+    gt_xyxy_lb = _gt_list_to_xyxy_tensor(gt_xywh, preds.device, meta=meta)  # [M,4]
+
+    boxes_bxn4 = preds[:, :4, :].permute(0, 2, 1)   # [B,N,4] (xywh, letterbox)
+    logits_bxcn = preds[:, 4:, :]                   # [B,nc,N]
+
+    # 若类别部分像概率(0~1)，转为 logits
+    zmin, zmax = logits_bxcn.min().item(), logits_bxcn.max().item()
+    if 0.0 <= zmin and zmax <= 1.0:
+        p = logits_bxcn.clamp(1e-6, 1-1e-6)
+        logits_bxcn = torch.log(p) - torch.log1p(-p)
+
+    # 选与 GT 最相关的候选 idx（batch=0）
+    b_xyxy = _xywh_to_xyxy(boxes_bxn4[0])          # [N,4]
+    if gt_xyxy_lb.numel() > 0:
+        iou = _iou_xyxy(b_xyxy, gt_xyxy_lb)        # [N,M]
+        best_per_gt = iou.argmax(dim=0)            # [M]
+        idx = torch.unique(best_per_gt, sorted=False)
+        if idx.numel() < topk:
+            topvals = iou.max(dim=1).values
+            topidx2 = torch.topk(topvals, k=min(topk, N)).indices
+            idx = torch.unique(torch.cat([idx, topidx2], 0), sorted=False)[:topk]
+    else:
+        # 没 GT 就按当前最大类别置信度取 topk
+        z = logits_bxcn[0]                         # [nc,N]
+        pmax = z.softmax(dim=0).max(dim=0).values
+        idx = torch.topk(pmax, k=min(topk, N)).indices
+
+    if idx.numel() == 0:
+        idx = torch.arange(min(topk, N), device=preds.device)
+
+    # 取这些候选的类别 logits：[K,nc]
+    z = logits_bxcn[0, :, idx].T                   # [K,nc]
+
+    # 1) CW-style margin
+    mask = torch.ones(nc, device=z.device, dtype=torch.bool)
+    mask[target_cls] = False
+    z_t   = z[:, target_cls]
+    z_oth = z[:, mask].max(dim=1).values
+    loss_margin = F.relu(kappa + z_oth - z_t).mean()
+
+    # 2) keep（KL >= 0）
+    with torch.no_grad():
+        p_clean = z.detach().softmax(dim=1)
+    logp_adv = z.log_softmax(dim=1)
+    loss_keep = F.kl_div(logp_adv, p_clean, reduction="batchmean")
+
+    # 3) 你的旧项：直接推高目标类 logit
+    loss_target = -z_t.mean()
+
+    loss = (
+        lambda_margin * loss_margin
+        + lambda_keep * loss_keep
+        + lambda_target * loss_target
+    )
+
+    if debug:
+        same_ratio = (z.argmax(dim=1) == target_cls).float().mean().item()
+        print(
+            f"[dbg] K={idx.numel()} nc={nc} target={target_cls} "
+            f"margin={loss_margin.item():.6f} keep={loss_keep.item():.6f} "
+            f"targ={loss_target.item():.6f} same_ratio={same_ratio:.3f} "
+            f"z_t_mean={z_t.mean().item():.3f} z_oth_mean={z_oth.mean().item():.3f}"
+        )
+    return loss
+
+
+# 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(
@@ -165,7 +306,7 @@ def fgsm_attack_on_detector(
     eps: float = 0.03,
     device: Optional[torch.device] = None,
     imgsz: Optional[int] = None,  # None=自动对齐到 stride 倍数；也可传 640
-  
+    gt_xywh: torch.Tensor | None = None  # letterbox坐标系下的目标框（可选）
 ) -> Image.Image:
     """
     Perform a single-step FGSM on a detection model (white-box).
@@ -200,7 +341,21 @@ def fgsm_attack_on_detector(
                 raise RuntimeError("模型 forward 返回了 tuple/list，但无法从中找到预测张量。")
             preds = tensor_pred
 
-        loss = - preds_to_confidence_sum(preds)  
+        target_cls = 2
+        loss = - preds_to_targeted_loss(
+            preds,
+            target_cls=target_cls,
+            gt_xywh=gt_xywh,    # 直接传你的 list[dict]
+            topk=20,
+            kappa=0.1,
+            lambda_margin=1.0,
+            lambda_keep=0.2,
+            lambda_target=0.0,   # 恢复 -p_t.mean() 的影响
+            debug=False,
+            meta=meta            # 若 GT 是原图坐标，务必传 meta
+        )
+
+        # loss = - preds_to_confidence_sum(preds)  
         loss.backward()
 
     # (d) FGSM 在 letterboxed 空间施扰
@@ -227,6 +382,7 @@ def pgd_attack_on_detector(
     iters: int = 10,
     device: Optional[torch.device] = None,
     imgsz: Optional[int] = None,  # None=自动对齐到 stride 倍数；也可传 640
+    gt_xywh: torch.Tensor | None = None  # letterbox坐标系下的目标框（可选）
 ):
     """
     在 YOLO 的 letterbox 域做 PGD，
@@ -249,6 +405,9 @@ def pgd_attack_on_detector(
     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)
 
+    targeted = True
+    sign = -1.0 if targeted else 1.0    # 定向取负号，非定向取正号
+    target_cls = 2
     for _ in range(iters):
         # 前向 + 反向（需要梯度）
         preds = net(x)
@@ -256,7 +415,21 @@ def pgd_attack_on_detector(
             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 = - preds_to_targeted_loss(
+            preds,
+            target_cls=target_cls,
+            gt_xywh=gt_xywh,    # 直接传你的 list[dict]
+            topk=20,
+            kappa=0.1,
+            lambda_margin=1.0,
+            lambda_keep=0.2,
+            lambda_target=0.0,   # 恢复 -p_t.mean() 的影响
+            debug=False,
+            meta=meta            # 若 GT 是原图坐标，务必传 meta
+        )
+        
+        # loss = - preds_to_confidence_sum(preds)      # 我们希望置信度总和下降 → 最小化
         loss.backward()
 
         # 更新步与投影（不记录计算图）