基于AidLux平台的智慧交通中对抗防御监测应用评测

 pip install scikit-image -i https://pypi.tuna.tsinghua.edu.cn/simple some-package  
    
 pip install torch==1.8.1 -i https://pypi.tuna.tsinghua.edu.cn/simple some-package 
    
 pip install torchvision==0.9.1 -i https://pypi.tuna.tsinghua.edu.cn/simple some-package

 import os

    
 import torch
    
 import torch.nn as nn
    
 from torchvision.models import mobilenet_v2
    
 from advertorch.utils import predict_from_logits
    
 from advertorch.utils import NormalizeByChannelMeanStd
    
  
    
 from advertorch.attacks import LinfPGDAttack
    
 from advertorch_examples.utils import ImageNetClassNameLookup
    
 from advertorch_examples.utils import bhwc2bchw
    
 from advertorch_examples.utils import bchw2bhwc
    
  
    
 device = "cuda" if torch.cuda.is_available() else "cpu"
    
  
    
 ### 读取图片
    
 def get_image():
    
     img_path = os.path.join("./Lesson4_code/adv_code/images", "mountain_bike.png")
    
     img_url = "https://farm1.static.flickr.com/230/524562325_fb0a11d1e1.jpg"
    
  
    
     def _load_image():
    
     from skimage.io import imread
    
     return imread(img_path) / 255.
    
  
    
     if os.path.exists(img_path):
    
     return _load_image()
    
     else:
    
     import urllib
    
     urllib.request.urlretrieve(img_url, img_path)
    
     return _load_image()
    
  
    
 def tensor2npimg(tensor):
    
     return bchw2bhwc(tensor[0].cpu().numpy())
    
  
    
 ### 展示攻击结果
    
 def show_images(model, img, advimg, enhance=127):
    
     np_advimg = tensor2npimg(advimg)
    
     np_perturb = tensor2npimg(advimg - img)
    
  
    
     pred = imagenet_label2classname(predict_from_logits(model(img)))
    
     advpred = imagenet_label2classname(predict_from_logits(model(advimg)))
    
  
    
     import matplotlib.pyplot as plt
    
  
    
     plt.figure(figsize=(10, 5))
    
     plt.subplot(1, 3, 1)
    
     plt.imshow(np_img)
    
  
    
     plt.axis("off")
    
     plt.title("original image\n prediction: {}".format(pred))
    
     plt.subplot(1, 3, 2)
    
     plt.imshow(np_perturb * enhance + 0.5)
    
  
    
     plt.axis("off")
    
     plt.title("the perturbation,\n enhanced {} times".format(enhance))
    
     plt.subplot(1, 3, 3)
    
     plt.imshow(np_advimg)
    
     plt.axis("off")
    
     plt.title("perturbed image\n prediction: {}".format(advpred))
    
     plt.show()
    
  
    
  
    
 normalize = NormalizeByChannelMeanStd(
    
     mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
  
    
 ### 常规模型加载
    
 model = mobilenet_v2(pretrained=True)
    
 model.eval()
    
 model = nn.Sequential(normalize, model)
    
 model = model.to(device)
    
  
    
 ### 数据预处理
    
 np_img = get_image()
    
 img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
    
 imagenet_label2classname = ImageNetClassNameLookup()
    
  
    
 ### 测试模型输出结果
    
 pred = imagenet_label2classname(predict_from_logits(model(img)))
    
 print("test output:", pred)
    
  
    
 ### 输出原label
    
 pred_label = predict_from_logits(model(img))
    
  
    
 ### 对抗攻击：PGD攻击算法
    
 adversary = LinfPGDAttack(
    
     model, eps=8 / 255, eps_iter=2 / 255, nb_iter=80,
    
     rand_init=True)
    
  
    
 ### 完成攻击，输出对抗样本
    
 advimg = adversary.perturb(img, pred_label)
    
  
    
 ### 展示源图片，对抗扰动，对抗样本以及模型的输出结果
    
 show_images(model, img, advimg)

 ### GCM模块

    
 robust_mode = GradientConcealment()
    
  
    
 ### 常规模型+GCM模块
    
 class Model(nn.Module):
    
     def __init__(self, l=290):
    
     super(Model, self).__init__()
    
  
    
     self.l = l
    
     self.gcm = GradientConcealment()
    
     # model = resnet18(pretrained=True)
    
     model = mobilenet_v2(pretrained=True)
    
  
    
     # pth_path = "/Users/rocky/Desktop/训练营/model/mobilenet_v2-b0353104.pth"
    
     # print(f'Loading pth from {pth_path}')
    
     # state_dict = torch.load(pth_path, map_location='cpu')
    
     # is_strict = False
    
     # if 'model' in state_dict.keys():
    
     #    model.load_state_dict(state_dict['model'], strict=is_strict)
    
     # else:
    
     #    model.load_state_dict(state_dict, strict=is_strict)
    
  
    
     normalize = NormalizeByChannelMeanStd(
    
         mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
     self.model = nn.Sequential(normalize, model)
    
  
    
     def load_params(self):
    
     pass
    
  
    
     def forward(self, x):
    
     x = self.gcm(x)
    
     # x = ResizedPaddingLayer(self.l)(x)
    
     out = self.model(x)
    
     return out
    
  
    
 ### 常规模型+GCM模块 加载
    
 model_defense = Model().eval().to(device)
    
  
    
 ### 数据预处理
    
 np_img = get_image()
    
 img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
    
 imagenet_label2classname = ImageNetClassNameLookup()
    
  
    
 ### 测试模型输出结果
    
 pred_defense = imagenet_label2classname(predict_from_logits(model_defense(img)))
    
 print("test output:", pred_defense)
    
  
    
 ### 输出原label
    
 pre_label = predict_from_logits(model_defense(img))

    export OMP_NUM_THREADS=1

 ### 常规模型加载

    
 model = mobilenet_v2(pretrained=True)
    
 model.eval()
    
 model = nn.Sequential(normalize, model)
    
 model = model.to(device)
    
  
    
 ### 替身模型加载
    
 model_su = resnet18(pretrained=True)
    
 model_su.eval()
    
 model_su = nn.Sequential(normalize, model_su)
    
 model_su = model_su.to(device)
    
  
    
 ### 数据预处理
    
 np_img = get_image()
    
 img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
    
 imagenet_label2classname = ImageNetClassNameLookup()
    
  
    
 ### 测试模型输出结果
    
 pred = imagenet_label2classname(predict_from_logits(model(img)))
    
 print("test output:", pred)
    
  
    
 ### 输出原label
    
 pred_label = predict_from_logits(model_su(img))
    
  
    
 ### 对抗攻击：PGD攻击算法
    
 adversary = LinfPGDAttack(
    
    model_su, eps=8/255, eps_iter=2/255, nb_iter=80,
    
    rand_init=True, targeted=False)
    
  
    
 ### 完成攻击，输出对抗样本
    
 advimg = adversary.perturb(img, pred_label)
    
  
    
 ### 展示源图片，对抗扰动，对抗样本以及模型的输出结果
    
 a_type = 'PGD'
    
 show_images(a_type, model, img, advimg)
    
  
    
 ### 迁移攻击样本保存
    
 save_path = "./Lesson5_code/adv_code/adv_results/" + img_path.split("/")[-1]
    
 # save_path = r"D:\code\aidlux_Rocky\Lesson5_code\adv_code\adv_results"
    
 torchvision.utils.save_image(advimg.cpu().data, save_path + "_adv_image_"+a_type+".png")

 ### 对抗攻击监测模型

    
 class Detect_Model(nn.Module):
    
     def __init__(self, num_classes=2):
    
     super(Detect_Model, self).__init__()
    
     self.num_classes = num_classes
    
     #model = create_model('mobilenetv3_large_075', pretrained=False, num_classes=num_classes)
    
     model = create_model('resnet50', pretrained=False, num_classes=num_classes)
    
     
    
  
    
     # self.multi_PreProcess = multi_PreProcess()
    
     pth_path = os.path.join("./Lesson5_code/model", 'track2_resnet50_ANT_best_albation1_64_checkpoint.pth')
    
     #pth_path = os.path.join("./Lesson5_code/model/", "track2_tf_mobilenetv3_large_075_64_checkpoint.pth")
    
     state_dict = torch.load(pth_path, map_location='cpu')
    
     is_strict = False
    
     if 'model' in state_dict.keys():
    
         model.load_state_dict(state_dict['model'], strict=is_strict)
    
     else:
    
         model.load_state_dict(state_dict, strict=is_strict)
    
     normalize = NormalizeByChannelMeanStd(
    
         mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
     # self.model = nn.Sequential(normalize, self.multi_PreProcess, model)
    
     self.model = nn.Sequential(normalize, model)
    
  
    
     def load_params(self):
    
     pass
    
  
    
     def forward(self, x):
    
     # x = x[:,:,32:193,32:193]
    
     # x = F.interpolate(x, size=(224,224), mode="bilinear", align_corners=True)
    
     # x = self.multi_PreProcess.forward(x)
    
     out = self.model(x)
    
     if self.num_classes == 2:
    
         out = out.softmax(1)
    
         #return out[:,1:]
    
         return out[:,1:]
    
  
    
 detect_model = Detect_Model().eval().to(device)
    
  
    
 ### 对抗攻击监测
    
 detect_pred = detect_model(img)
    
 print(detect_pred)

 if detect_pred > 0.5:

    
     print("出现可疑对抗样本。")

 ### 对抗攻击：PGD攻击算法

    
 adversary = LinfPGDAttack(
    
    model_su, eps=8/255, eps_iter=2/255, nb_iter=80,
    
    rand_init=True, targeted=False)
    
  
    
 # ### 对抗攻击：FGSM攻击算法 (eps = 0.5/255, 2/255, 8/255)S
    
 # adversary = FGSM(
    
 #    model_su, eps=8/255,
    
 #    targeted=False)
    
  
    
 # ### 对抗攻击：L1PGD攻击算法 (eps = 100, 400, 1600)
    
 # adversary = L1PGDAttack(
    
 #    model_su, eps=1600, eps_iter=2/255, nb_iter=80,
    
 #    rand_init=True, targeted=False)
    
  
    
 # ### 对抗攻击：L2PGD攻击算法 (eps = 0.5, 2, 8)
    
 # adversary = L2PGDAttack(
    
 #    model_su, eps=8, eps_iter=2/255, nb_iter=80,
    
 #    rand_init=True, targeted=False)
    
  
    
 # ### 对抗攻击：LinfMomentumIterativeAttack攻击算法 (eps = 0.5/255, 2/255, 8/255)
    
 # adversary = LinfMomentumIterativeAttack(
    
 #    model_su, eps=8/255

 img_path = "Lesson5_code/adv_code/orig_images/vid_5_27620.jpg_0.jpg"

    
 # img_path = "./Lesson5_code/adv_code/orig_images/vid_5_31560.jpg_0.jpg"

 #%%

    
 ## 1.车辆检测+检测框小图提取。
    
  
    
 # aidlux相关
    
 from cvs import *
    
 import aidlite_gpu
    
 from utils import detect_postprocess, preprocess_img, draw_detect_res, extract_detect_res
    
  
    
 import time
    
 import cv2
    
  
    
 # AidLite初始化：调用AidLite进行AI模型的加载与推理，需导入aidlite
    
 aidlite = aidlite_gpu.aidlite()
    
 # Aidlite模型路径
    
 model_path = '/home/Lesson5_code/yolov5_code/models/yolov5_car_best-fp16.tflite'
    
 # 定义输入输出shape
    
 in_shape = [1 * 640 * 640 * 3 * 4]
    
 out_shape = [1 * 25200 * 6 * 4]
    
 # 加载Aidlite检测模型：支持tflite, tnn, mnn, ms, nb格式的模型加载
    
 aidlite.ANNModel(model_path, in_shape, out_shape, 4, 0)
    
  
    
 # 读取图片进行推理
    
 # 设置测试集路径
    
 source = "/home/Lesson5_code/yolov5_code/data/images/test1"
    
 images_list = os.listdir(source)
    
 print(images_list)
    
 frame_id = 0
    
 # 读取数据集
    
 for image_name in images_list:
    
     frame_id += 1
    
     print("frame_id:", frame_id)
    
     image_path = os.path.join(source, image_name)
    
     frame = cvs.imread(image_path)
    
  
    
     # 预处理
    
     img = preprocess_img(frame, target_shape=(640, 640), div_num=255, means=None, stds=None)
    
     # 数据转换：因为setTensor_Fp32()需要的是float32类型的数据，所以送入的input的数据需为float32,大多数的开发者都会忘记将图像的数据类型转换为float32
    
     aidlite.setInput_Float32(img, 640, 640)
    
     # 模型推理API
    
     aidlite.invoke()
    
     # 读取返回的结果
    
     pred = aidlite.getOutput_Float32(0)
    
     # 数据维度转换
    
     pred = pred.reshape(1, 25200, 6)[0]
    
     # 模型推理后处理
    
     pred = detect_postprocess(pred, frame.shape, [640, 640, 3], conf_thres=0.25, iou_thres=0.45)
    
     # 绘制推理结果
    
     res_img = draw_detect_res(frame, pred)
    
     # cvs.imshow(res_img)
    
  
    
     # 测试结果展示停顿
    
     #time.sleep(5)
    
  
    
     # 图片裁剪，提取车辆目标区域
    
     extract_detect_res(frame, pred, image_name)
    
  
    
     # cvs.imshow(cut_img)
    
     # cap.release()
    
     # cv2.destroyAllWindows()
    
  
    
 #%%
    
 # ## 2.使用对抗样本
    
 # from cvs import os
    
  
    
 # rec_path = "/home/Lesson5_code/yolov5_code/aidlux/extract_results/"
    
 # recs_list = os.listdir(rec_path)
    
 # print(recs_list)
    
  
    
 # ## 使用原样本
    
 # img_path = os.path.join(rec_path, recs_list[1])
    
  
    
 # ## 替换对抗样本
    
 # rec_su_path = "/home/Lesson5_code/yolov5_code/aidlux/extract_results_su/"
    
 # recs_su_list = []
    
 # for rec_su in recs_list:
    
 #     recs_su_list.append(rec_su.split('.jpg_0.jpg')[0]+'.jpg_0.png')
    
 # img_path = os.path.join(rec_su_path, recs_su_list[1])
    
  
    
  
    
 #%%
    
 ## 3.AI安全监测与告警功能
    
 ## 
    
  
    
 import os
    
 import torch
    
 import requests
    
 import time
    
 import torch.nn as nn
    
 from torchvision.models import mobilenet_v2,resnet18
    
 from advertorch.utils import predict_from_logits
    
 from advertorch.utils import NormalizeByChannelMeanStd
    
 from robust_layer import GradientConcealment, ResizedPaddingLayer
    
 from timm.models import create_model
    
  
    
 from advertorch.attacks import LinfPGDAttack
    
 from advertorch_examples.utils import ImageNetClassNameLookup
    
 from advertorch_examples.utils import bhwc2bchw
    
 from advertorch_examples.utils import bchw2bhwc
    
  
    
 device = "cuda" if torch.cuda.is_available() else "cpu"
    
  
    
  
    
 ### 读取图片
    
 def get_image(img_path):
    
     # img_path = os.path.join("/home/Lesson5_code/adv_code/adv_results/", "adv_image.png")  ## Cars28.png
    
     print('img_path',img_path)
    
     img_url = "https://farm1.static.flickr.com/230/524562325_fb0a11d1e1.jpg"
    
  
    
     def _load_image():
    
     from skimage.io import imread
    
     return imread(img_path) / 255.
    
  
    
     if os.path.exists(img_path):
    
     return _load_image()
    
     else:
    
     import urllib
    
     urllib.request.urlretrieve(img_url, img_path)
    
     return _load_image()
    
  
    
  
    
 def tensor2npimg(tensor):
    
     return bchw2bhwc(tensor[0].cpu().numpy())
    
  
    
  
    
 normalize = NormalizeByChannelMeanStd(
    
     mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
  
    
  
    
 ### 常规模型加载
    
 class Model(nn.Module):
    
     def __init__(self, l=290):
    
     super(Model, self).__init__()
    
  
    
     self.l = l
    
     self.gcm = GradientConcealment()
    
     #model = resnet18(pretrained=True)
    
     model = mobilenet_v2(pretrained=True)
    
  
    
     # pth_path = "/home/Lesson5_code/model/mobilenet_v2-b0353104.pth"
    
     # print(f'Loading pth from {pth_path}')
    
     # state_dict = torch.load(pth_path, map_location='cpu')
    
     # is_strict = False
    
     # if 'model' in state_dict.keys():
    
     #    model.load_state_dict(state_dict['model'], strict=is_strict)
    
     # else:
    
     #    model.load_state_dict(state_dict, strict=is_strict)
    
  
    
     normalize = NormalizeByChannelMeanStd(
    
         mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
     self.model = nn.Sequential(normalize, model)
    
  
    
     def load_params(self):
    
     pass
    
  
    
     def forward(self, x):
    
     #x = self.gcm(x)
    
     #x = ResizedPaddingLayer(self.l)(x)
    
     out = self.model(x)
    
     return out
    
  
    
  
    
 ### 对抗攻击监测模型
    
 class Detect_Model(nn.Module):
    
     def __init__(self, num_classes=2):
    
     super(Detect_Model, self).__init__()
    
     self.num_classes = num_classes
    
     #model = create_model('mobilenetv3_large_075', pretrained=False, num_classes=num_classes)
    
     model = create_model('resnet50', pretrained=False, num_classes=num_classes)
    
  
    
     # self.multi_PreProcess = multi_PreProcess()
    
     pth_path = os.path.join("/home/Lesson5_code/model", 'track2_resnet50_ANT_best_albation1_64_checkpoint.pth')
    
     #pth_path = os.path.join("/home/Lesson5_code/model", "track2_tf_mobilenetv3_large_075_64_checkpoint.pth")
    
     state_dict = torch.load(pth_path, map_location='cpu')
    
     is_strict = False
    
     if 'model' in state_dict.keys():
    
         model.load_state_dict(state_dict['model'], strict=is_strict)
    
     else:
    
         model.load_state_dict(state_dict, strict=is_strict)
    
     normalize = NormalizeByChannelMeanStd(
    
         mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    
     # self.model = nn.Sequential(normalize, self.multi_PreProcess, model)
    
     self.model = nn.Sequential(normalize, model)
    
  
    
     def load_params(self):
    
     pass
    
  
    
     def forward(self, x):
    
     # x = x[:,:,32:193,32:193]
    
     # x = F.interpolate(x, size=(224,224), mode="bilinear", align_corners=True)
    
     # x = self.multi_PreProcess.forward(x)
    
     out = self.model(x)
    
     if self.num_classes == 2:
    
         out = out.softmax(1)
    
         #return out[:,1:]
    
         return out[:,1:]
    
  
    
  
    
 def main(img_path):
    
  
    
     np_img = get_image(img_path)
    
     img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
    
     imagenet_label2classname = ImageNetClassNameLookup()
    
  
    
     model = Model().eval().to(device)
    
  
    
     detect_model = Detect_Model().eval().to(device)
    
  
    
  
    
     ### 对抗攻击监测
    
     detect_pred = detect_model(img)
    
     print(detect_pred)
    
  
    
     if detect_pred > 0.5:
    
     print("出现可疑对抗样本。")
    
     pred = imagenet_label2classname(predict_from_logits(model(img)))
    
     print(pred)
    
     id = 't5Su94C'
    
     # 填写喵提醒中，发送的消息，这里放上前面提到的图片外链
    
     text = "出现可疑对抗样本！！" +str(img_path)
    
     ts = str(time.time())  # 时间戳
    
     type = 'json'  # 返回内容格式
    
     request_url = "http://miaotixing.com/trigger?"
    
  
    
     headers = {
    
         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.67 Safari/537.36 Edg/87.0.664.47'}
    
  
    
     result = requests.post(request_url + "id=" + id + "&text=" + text + "&ts=" + ts + "&type=" + type,
    
                         headers=headers)
    
     else:
    
     print("图片正常。")
    
     pred = imagenet_label2classname(predict_from_logits(model(img)))
    
     print(pred)
    
  
    
  
    
 # %%
    
  
    
 if __name__ == '__main__':
    
     ## 使用原样本
    
     img_path_ = "Lesson5_code/adv_code/orig_images/vid_5_26660.jpg_0.jpg"
    
     main(img_path_)
    
  
    
     ## 替换对抗样本
    
     img_path_ = "Lesson5_code/adv_code/adv_results/vid_5_26660.jpg_0.png"
    
     main(img_path_)