# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 2
#     language: python
#     name: python2
# ---

# +
import caffe
import cv2
import numpy as np
import os
import utils


def get_witnesses(dir_path, layers):
    witnesses = {}
    for f in os.listdir(dir_path):
        for line in open(dir_path + f, 'r'):
            line = line.strip().split(',')
            layer = layers.index(line.pop(0))
            if layer in witnesses:
                witnesses[layer].extend(map(int, line))
            else:
                witnesses[layer] = map(int, line)
    return witnesses


def weaken(x):
    return np.exp(-x/100)


def strengthen(x):
    return 2.15 - np.exp(-x/60)


def attribute_model(net, img_path):
    pre = 'data'
    net.blobs[pre].data[...] = utils.get_data(img_path)

    for idx in witnesses:
        curr = vgg_layers[idx]
        post = vgg_layers[idx + 1]

        if pre == 'data':
            net.forward(end=post)
        else:
            net.forward(start=pre, end=post)

        ### attribute witnesses for the current layer ###
        neurons = witnesses[idx]

        attri_data = []
        for i in neurons:
            attri_data.append(np.sum(net.blobs[curr].data[0][i]))

        ### calculate mean and standard devation of attribute witness activations ###
        attri_mean = np.mean(attri_data)
        attri_std  = np.std(attri_data)

        ### neuron weakening ###
        for i in range(len(net.blobs[curr].data[0])):
            if i not in neurons:
                other_data = np.sum(net.blobs[curr].data[0][i])

                if other_data > attri_mean:
                    deviation = 0
                    if attri_std != 0:
                        deviation = (other_data - attri_mean) / attri_std

                    if 'pool3' in curr:
                        tmp = net.blobs[curr].data[0][i]
                        h, w = tmp.shape
                        tmp = tmp[2:h-2, 2:w-2]
                        tmp = cv2.resize(tmp, (h,w), interpolation=cv2.INTER_CUBIC)
                        net.blobs[curr].data[0][i] = tmp

                    net.blobs[curr].data[0][i] *= weaken(deviation)

        ### neuron strengthening ###
        for i in neurons:
            deviation = 0
            if attri_std != 0:
                deviation = abs(np.sum(net.blobs[curr].data[0][i]) - np.min(attri_data)) / attri_std
            net.blobs[curr].data[0][i] *= strengthen(deviation)

        pre = post

    net.forward(start=pre)
    return net.blobs['prob'].data[0].copy()


# -

caffe.set_mode_cpu()
vgg_root   = '../data/vgg_face_caffe/'
vgg_deploy = vgg_root + 'VGG_FACE_deploy.prototxt'
vgg_weight = vgg_root + 'VGG_FACE.caffemodel'
vgg_net    = caffe.Net(vgg_deploy, vgg_weight, caffe.TEST)
vgg_names  = utils.read_list(vgg_root + 'names.txt')
vgg_layers = utils.get_layers(vgg_net)
witnesses  = get_witnesses('../data/witnesses/', vgg_layers)

# +
attack_path = '../data/attacks/patch_first/'

img_count = 0
adv_count = 0

for img_name in os.listdir(attack_path):
    img_name  = img_name.strip()
    img_path  = attack_path + img_name
    img_count += 1

    prob_original  = utils.get_prob(vgg_net, img_path)
    prob_attribute = attribute_model(vgg_net, img_path)
    id_original    = np.argmax(prob_original)
    id_attribute   = np.argmax(prob_attribute)

    ### consistency observer for detecting adversarial samples ###
    if id_original != id_attribute:
        adv_count += 1

    id_gold = utils.get_identity(img_name, vgg_names)

    print('{:3} Gold-Original-Attribute: {:4}-{:4}-{:4}  |  Adversary: {:.4f} [{:3}/{:3}]'
          .format(img_count, id_gold, id_original, id_attribute, 
                  1.0*adv_count/img_count, adv_count, img_count))

StarcoderdataJupyterScriptsDedupFiltered