<jupyter_start><jupyter_text>Install SlayerPytorch on ColabAfter the installations the runtime needs to be restarted. ```exit()```Will restart the runtime without deleting files. The runtime will automatically start. And if you press "run all" the run is not interrupted and works till the end.<jupyter_code>!git clone https://github.com/bamsumit/slayerPytorch
!pip install ninja
exit()
%cd slayerPytorch/
!python setup.py install
exit()<jupyter_output><empty_output><jupyter_text>Test to verify if everything went well with the installation<jupyter_code>%cd slayerPytorch/test/
!python -m  unittest<jupyter_output><empty_output><jupyter_text>Network Configuration<jupyter_code>!unzip -q '/content/drive/MyDrive/Test.zip' -d '/content/'
import sys, os
CURRENT_TEST_DIR = os.getcwd()
sys.path.append(CURRENT_TEST_DIR + "/../../src")

from datetime import datetime
import numpy as np
import matplotlib.pyplot as plt
from IPython.display import HTML
import torch
from torch.utils.data import Dataset, DataLoader
import slayerSNN as snn
#from learningStats import learningStats
from IPython.display import HTML
import time
import shutil

def save_ckp(state, is_best_loss, is_best_acc, checkpoint_dir, best_model_dir):
    f_path = checkpoint_dir+'checkpoint.pt'
    torch.save(state, f_path)
    if is_best_loss:
        best_fpath = best_model_dir+'best_model.pt'
        shutil.copyfile(f_path, best_fpath)
    if is_best_acc:
        acc_fpath =  best_model_dir+'best_acc.pt'
        shutil.copyfile(f_path, acc_fpath)
        
def load_ckp(checkpoint_fpath, model, optimizer):
    checkpoint = torch.load(checkpoint_fpath)
    model.load_state_dict(checkpoint['state_dict'])
    optimizer.load_state_dict(checkpoint['optimizer'])
    return model, optimizer, checkpoint['epoch']

actionName = [
    'hand_clapping',
    'right_hand_wave',
    'left_hand_wave',
    'right_arm_clockwise',
    'right_arm_counter_clockwise',
    'left_arm_clockwise', 
    'left_arm_counter_clockwise',
    'arm_roll',
    'air_drums',
    'air_guitar',
    'other_gestures',
]

# Define dataset module
class IBMGestureDataset(Dataset):
    def __init__(self, datasetPath, sampleFile, samplingTime, sampleLength):
        self.path = datasetPath 
        self.samples = np.loadtxt(sampleFile).astype('int')
        self.samplingTime = samplingTime
        self.nTimeBins    = int(sampleLength / samplingTime)

    def __getitem__(self, index):
        # Read inoput and label
        inputIndex  = self.samples[index, 0]
        classLabel  = self.samples[index, 1]
        # Read input spike
        inputSpikes = snn.io.readNpSpikes(
                        self.path + str(inputIndex.item()) + '.npy'
                        ).toSpikeTensor(torch.zeros((2,128,128,self.nTimeBins)),
                        samplingTime=self.samplingTime)
        # Create one-hot encoded desired matrix
        desiredClass = torch.zeros((11, 1, 1, 1))
        desiredClass[classLabel,...] = 1
        
        return inputSpikes, desiredClass, classLabel

    def __len__(self):
        return self.samples.shape[0]
		
# Define the network
class Network(torch.nn.Module):
    def __init__(self, netParams):
        super(Network, self).__init__()
        # initialize slayer
        slayer = snn.loihi(netParams['neuron'], netParams['simulation'])
        self.slayer = slayer
        # define network functions
        self.conv1 = slayer.conv(2, 16, 5, padding=2, weightScale=10)
        self.conv2 = slayer.conv(16, 32, 3, padding=1, weightScale=50)
        self.pool1 = slayer.pool(4)
        self.pool2 = slayer.pool(2)
        self.pool3 = slayer.pool(2)
        self.fc1   = slayer.dense((8*8*32), 512)
        self.fc2   = slayer.dense(512, 11)
        self.drop  = slayer.dropout(0.1)

    def forward(self, spikeInput):
        spike = self.slayer.spikeLoihi(self.pool1(spikeInput )) # 32, 32, 2
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.drop(spike)
        spike = self.slayer.spikeLoihi(self.conv1(spike)) # 32, 32, 16
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.slayer.spikeLoihi(self.pool2(spike)) # 16, 16, 16
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.drop(spike)
        spike = self.slayer.spikeLoihi(self.conv2(spike)) # 16, 16, 32
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.slayer.spikeLoihi(self.pool3(spike)) #  8,  8, 32
        spike = spike.reshape((spike.shape[0], -1, 1, 1, spike.shape[-1]))
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.drop(spike)
        spike = self.slayer.spikeLoihi(self.fc1  (spike)) # 512
        spike = self.slayer.delayShift(spike, 1)
        
        spike = self.slayer.spikeLoihi(self.fc2  (spike)) # 11
        spike = self.slayer.delayShift(spike, 1)
        
        return spike


if __name__ == '__main__':
	netParams = snn.params('/content/slayerPytorch/exampleLoihi/03_IBMGesture/network.yaml')
	
	# Define the cuda device to run the code on.
	device = torch.device('cuda')
	# deviceIds = [2, 3]

	# Create network instance.
	net = Network(netParams).to(device)
	# net = torch.nn.DataParallel(Network(netParams).to(device), device_ids=deviceIds)

	# Create snn loss instance.
	error = snn.loss(netParams, snn.loihi).to(device)

	# Define optimizer module.
	# optimizer = torch.optim.Adam(net.parameters(), lr = 0.01, amsgrad = True)
	optimizer = snn.utils.optim.Nadam(net.parameters(), lr = 0.01, amsgrad = True)

	#Dataset and dataLoader instances.

	testingSet = IBMGestureDataset(datasetPath ="/content/MyDrive/Test/", 
									sampleFile  = "/content/drive/MyDrive/Test/test.txt",
									samplingTime= 1.0,
									sampleLength= 1450)
	testLoader = DataLoader(dataset=testingSet, batch_size=1, shuffle=False, num_workers=1)
  
	i=0

	# Learning stats instance.
	stats = snn.utils.stats()
	ckp_path = "drive/My Drive/best/check.pt"
	net, optimizer, start_epoch = load_ckp(ckp_path, net, optimizer)
	start_epoch<jupyter_output><empty_output><jupyter_text>Adversarial AttacksCollection of 5 different Adversarial Attacks Gradient Based Attack<jupyter_code>testingSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testingSet, batch_size=1, shuffle=False, num_workers=4)

import argparse
import numpy as np
import pdb 
import torch, torchvision
from torch.utils.data import DataLoader
from torch.autograd import Variable


def update(x,y,t,s,A):
  for i in range(x-s,x+s+1):
     for j in range(y-s,y+s+1):
        if not(i==x and j==y) and i>0 and j>0 and i<128 and j<128:
	        A[i][j] = t
    

#Select the device to run the code 
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print ('Device: ' + str(device))

def calc_gradients(
		test_dataloader,
		model,
		max_iter,
		learning_rate,
    samples,
		filter_on,
		s,
		T,
		targets=None,
		weight_loss2=1,
		batch_size=1,
		seq_len=40,):
	
	#Define the modifier and the optimizer

  min_loss = 1e-5
  prev_loss = 1e-5
  correct = np.zeros(max_iter)	
  for batch_index, (input_image, input_target, input_label) in enumerate(test_dataloader):

    modif = torch.Tensor(1, seq_len, 2, 128, 128).fill_(1).to(device)
    modifier = torch.nn.Parameter(modif, requires_grad=True)
    optimizer = torch.optim.Adam([modifier], lr=learning_rate)

    if batch_index>samples:
      break
    model.eval()
    with torch.no_grad():	
      input_image, input_label = input_image.to(device), input_label.to(device)

    #Clean video prediction 
    print(f'Batch Number: {batch_index}/{len(test_dataloader)}')
    print('------------------prediction for clean video-------------------')
    input_image = Variable(input_image, requires_grad=True)
    output = model.forward(input_image)
    pre_label= snn.predict.getClass(output)
    p = torch.nn.Softmax(dim=1)		
    print (f'Prediction: {pre_label}, Original_label: {input_label.cpu().numpy()}')

    print('------------------prediction for adversarial video-------------------')

    min_in = input_image.min().detach() #0
    max_in = input_image.max().detach() #1

    for iiter in range(max_iter):
      with open ('/content/modifier.txt','w') as file:
        file.write(str(modifier))      
      print(modifier[0,0,0,0,0])			
      input_image = Variable(input_image, requires_grad=True)
      #model.lstm.reset_hidden_state()
      
      #Frames to be perturbed
      indicator = [0]*1450
      for i,x in enumerate(indicator):
        if (i>399 and i<510) or (i>724 and i<870) or (i>1199 and i< 1275):
          x=1
    
      #Perturbating the frames

      input_image=torch.reshape(input_image, (1,1450,2,128,128)) ###
      true_image = modifier[0,0,:,:,:]+input_image[0,0,:,:,:]
      true_image = torch.unsqueeze(true_image, 0)
      
      for ll in range(seq_len-1): #seq_len =1450
        if indicator[ll+1] == 1:
          mask_temp = modifier[0,ll+1,:,:,:]+input_image[0,ll+1,:,:,:]
        else:
          mask_temp = input_image[0,ll+1,:,:,:]
        mask_temp = torch.unsqueeze(mask_temp,0)
        true_image = torch.cat((true_image, mask_temp),0)

      #######		 Filter section 	######
      if filter_on :
        img = torch.reshape(true_image, (2,128,128,1450)).cpu().detach()
        TD = snn.io.spikeArrayToEvent(img.numpy(), samplingTime=1)
        snn.io.encodeNpSpikes("/content/drive/My Drive/temp/img.npy", TD)
        data= np.load('/content/drive/My Drive/temp/img.npy'.format(i))
        data= data[data[:,3].argsort()] #sort elements by timestamp
        temp=np.zeros((128,128))
        real=[]
        for d in data:
          update(int(d[0]),int(d[1]),d[3],s,temp)
          if d[3]-temp[int(d[0])][int(d[1])]<T:
            real.append(d)
        real=np.stack(real, axis=0 )
        with open("/content/drive/My Drive/temp/fil.npy", "wb") as f:
          np.save(f,real)
        true_image = snn.io.readNpSpikes(
            "/content/drive/My Drive/temp/fil.npy"
            ).toSpikeTensor(torch.zeros((2,128,128,1450)),
            samplingTime=1.0)
        true_image =  true_image.to(device)
      true_image = torch.reshape(true_image, (1,1450,2,128,128))			
      #######		 End Filter Section 	 #######		
     
      #Prediction on the adversarial video
      output = model.forward(torch.reshape(true_image, (1,2,128,128,1450)))
      pre_label = snn.predict.getClass(output)
      numSpikes = torch.sum(output, 4, keepdim=True)
      print (f'numSpikes: {numSpikes[0,:,0,0]}, maxSpikes : {torch.max(numSpikes)}')
      numSpikes= torch.div(numSpikes,torch.max(numSpikes))	
      probs= p(numSpikes.reshape((numSpikes.shape[0], -1)))	

      #extracting the probability of true label 
      zero_array = torch.zeros(11).to(device) 
      zero_array[input_label.cpu()] = 1
      true_label_onehot = probs*zero_array
      true_label_prob = torch.sum(true_label_onehot, 1)

      if pre_label.numpy()==input_label.cpu().numpy() : correct[iiter]+=1

      #Loss
      if targets is None:
        loss1 = -torch.log(1 - true_label_prob + 1e-6)
      else:
        loss1 = -torch.log(true_label_prob + 1e-6)
      loss1 = torch.mean(loss1)
      input_image=torch.reshape(input_image, (1,1450,2,128,128))
      loss2 =  weight_loss2*torch.sum(torch.sqrt(torch.mean(torch.pow((true_image-input_image+1e-21), 2), dim=0).mean(dim=2).mean(dim=2).mean(dim=1)))
      loss = loss1 +  loss2
      optimizer.zero_grad()
      loss.backward() 
      optimizer.step()
      if True: #iiter % max_iter == 0: 
        print (f'Probability for ground truth label : {true_label_prob.detach().cpu().numpy()}')
        if prev_loss < loss : 
          print(f'Iteration: [{iiter+1}/{max_iter}], Loss: {loss}(\u25b2), Loss1: {loss1}, Loss2: {loss2}\n\n')
        elif prev_loss > loss: 
          print(f'Iteration: [{iiter+1}/{max_iter}], Loss: {loss}(\u25bc), Loss1: {loss1}, Loss2: {loss2}\n\n')
        else: 
          print(f'Iteration: [{iiter+1}/{max_iter}], Loss: {loss}, Loss1: {loss1}, Loss2: {loss2}\n\n')
      prev_loss = loss

      break_condition = False
      if loss < min_loss:
        if torch.abs(loss-min_loss) < 0.0001:
            break_condition = True
            print ('Aborting early!')
        min_loss = loss

      # Empty cache
      if torch.cuda.is_available():
        torch.cuda.empty_cache()
    # Save adversarial dataset as spike events
    image = torch.reshape(true_image, (2,128,128,1450)).cpu().detach()
    TD = snn.io.spikeArrayToEvent(image.numpy(), samplingTime=1)
    snn.io.encodeNpSpikes("/content/prova/{}.npy".format(batch_index), TD)
    if torch.cuda.is_available():
        torch.cuda.empty_cache()	
  print(correct)


def main(net):

  args = {
      'description': 'Sparse Adversarial Perturbations',
      'num_iter': 5,
      'learning_rate': 1,
      'samples': 264,
			'filter_on': False,
			's': 2,
			'T': 5,
      'target': None,
      'weight_loss2': 0,
      'split_path': None,
      'split_number': 1,
      'img_dim': 128,
      'channels': 2,
			}
  seq_len = 1450 #number of frames in a video
  batch_size = 1
  targets = None
  image_shape = (args['channels'], args['img_dim'], args['img_dim'])
  net = net.to(device)
  net.train()
  print('Model Loaded Successfully!')
	#Call the function to generate the adversarial videos
  calc_gradients(
	testLoader,
	net,
	args['num_iter'],
	args['learning_rate'],
  args['samples'],
	args['filter_on'],
	args['s'],
	args['T'],
	targets,
	args['weight_loss2'],
	batch_size,
	seq_len,
	)

if __name__ == '__main__':
	main(net)<jupyter_output><empty_output><jupyter_text>Frame Attack<jupyter_code>import numpy as np
import torch
import time

def cornerGen(x=0,y=0, N=128,T=1450):
  corner = torch.zeros((N,N,T))
  for i in range(N):
    for j in range(N):
      if i==x or i == N-1-x or j==y or j== N-1-y :
        corner[i,j,:]=1       

  return corner.expand(2,N,N,T)

testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testSet, batch_size=1, shuffle=False, num_workers=4) 

samples = [i for i in range(264)]
stats.testing.reset()

N=128
x=0
y=0

start=time.time()

temp= list(samples)
corner= cornerGen(x,y)
for i  in temp:

  input,target,label=testSet[i]
  
  adv= input.unsqueeze(dim=0)+corner
  
  # Save adversarial dataset as spike events
  image = torch.reshape(adv, (2,N,N,1450))
  TD = snn.io.spikeArrayToEvent(image.numpy(), samplingTime=1)
  snn.io.encodeNpSpikes("/content/prova/{}.npy".format(i), TD)
print(time.time()-start)<jupyter_output><empty_output><jupyter_text>Corner Attack<jupyter_code>import numpy as np
import torch
import time

def cornerGen(x=0,y=0,left=False, N=128,T=1450):
  corner = torch.zeros((N,N,T))
  for i in range(N):
    for j in range(N):
      if left:
        if (i==x and j<y):
          corner[i,j,:]=1
      else:
        if (i==x and j>=N-y):
          corner[i,j,:]=1        

  return corner.expand(2,N,N,T)

testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testSet, batch_size=1, shuffle=False, num_workers=4) 

samples = [i for i in range(264)]
stats.testing.reset()
it=0
N=128
flag=0
left = 1
x=0
y=2
max_y=2
start=time.time()
while samples and x<N and y<N:
  temp= list(samples)
  corner= cornerGen(x,y,left)
  for i  in temp:

    input,target,label=testSet[i]
    
    adv= input.unsqueeze(dim=0)+corner
    with torch.no_grad():
      adv  = adv.to(device)
      target = target.to(device) 

    output = net.forward(adv)
    if (snn.predict.getClass(output) != label):
      #print(f' Sample {i} , {label}')
      try: 
        samples.remove(i)
      except:
        print(f'{i} not in list')
      # Save adversarial dataset as spike events
      image = torch.reshape(adv, (2,N,N,1450)).cpu().detach()
      TD = snn.io.spikeArrayToEvent(image.numpy(), samplingTime=1)
      snn.io.encodeNpSpikes("/content/prova/{}.npy".format(i), TD)
  print(f'remaining samples {len(samples)}, x = {x}, y={y}', 'left' if left else 'right')

  if flag:
    x= 0
    flag^=1
    left^=1
    if left:
      if y==2:
        y=5
      else:
        y+=4
  else:
    x= N-1
    flag^=1

print(time.time()-start)<jupyter_output><empty_output><jupyter_text>Dash Attack<jupyter_code>import numpy as np
import torch
import time

def cornerGen(x=0,y=0,left=False, N=128,T=1450):
  corner = torch.zeros((N,N,T))
  for i in range(N):
    for j in range(N):
      if left:
        if (i==x) and (j==y or j== y-1):
          corner[i,j,:]=1
      else:
        if (i==x) and (j==N-y or j== N-y+1):
          corner[i,j,:]=1        

  return corner.expand(2,N,N,T)

testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testSet, batch_size=1, shuffle=False, num_workers=4) 

samples = [i for i in range(264)]
stats.testing.reset()
it=0
N=128
flag=0
left = 1
x=0
y=2
max_y=2
start=time.time()
while samples and x<N and y<N:
  temp= list(samples)
  corner= cornerGen(x,y,left)
  for i  in temp:

    input,target,label=testSet[i]
    
    adv= input.unsqueeze(dim=0)+corner
    with torch.no_grad():
      adv  = adv.to(device)
      target = target.to(device) 

    output = net.forward(adv)
    if (snn.predict.getClass(output) != label):
      #print(f' Sample {i} , {label}')
      try: 
        samples.remove(i)
      except:
        print(f'{i} not in list')
      # Save adversarial dataset as spike events
      image = torch.reshape(adv, (2,N,N,1450)).cpu().detach()
      TD = snn.io.spikeArrayToEvent(image.numpy(), samplingTime=1)
      snn.io.encodeNpSpikes("/content/prova/{}.npy".format(i), TD)
  print(f'remaining samples {len(samples)}, x = {x}, y={y}', 'left' if left else 'right')

  if flag:
    x= 0
    flag^=1
    left^=1
    if left:
      if y==2:
        y=5
      else:
        y+=4
  else:
    x= N-1
    flag^=1

print(time.time()-start)<jupyter_output><empty_output><jupyter_text>MF-Aware Dash Attack<jupyter_code>import numpy as np
import torch
import time


def dashGen(x=0,y=2,left=False,th0=10, N=128,T=1450,):
  th=th0
  dash = torch.zeros((2,N,N,T))
  for t in range(T):
    for i in range(N):
      for j in range(N):
        if left:
          if (i==x  and (j==y or j== y-1) and t<th):
            dash[0,i,j,t]=1
        else:
          if (i==x  and (j==N-y or j== N-y-1) and t<th):
            dash[1,i,j,t]=1
    if t==th:
      th+=th0
      y+=2

  return dash


testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testSet, batch_size=1, shuffle=False, num_workers=4)


#samples = [i for i in range(264)]
samples = [0,178]
stats.testing.reset()

N=128
flag=0
left = 1
min_x=0
x=min_x
y=2
th0=250
start=time.time()
while samples and min_x<12:
  temp= list(samples)
  corner= dashGen(y,x,left,th0)
  for i  in temp:

    input,target,label=testSet[i]
    
    adv= torch.clamp(input.unsqueeze(dim=0)+corner,0,1)
    with torch.no_grad():
      adv  = adv.to(device)
      target = target.to(device) 

    output = net.forward(adv)
    if (snn.predict.getClass(output) != label):
      #print(f' Sample {i} , {label}')
      try: 
        samples.remove(i)
      except:
        print(f'{i} not in list')
      # Save adversarial dataset as spike events
      image = torch.reshape(adv, (2,N,N,1450)).cpu().detach()
      TD = snn.io.spikeArrayToEvent(image.numpy(), samplingTime=1)
      snn.io.encodeNpSpikes("/content/prova/{}.npy".format(i), TD)
  print(f'remaining samples {len(samples)}, x = {x}, y={y}', 'left' if left else 'right')

  if flag:
    x= min_x
    flag^=1
    left^=1
    if left:
      y+=4
  else:
    x= N-1-min_x
    flag^=1
  
  if y>N/2:
    y=2
    min_x+=4
    x=min_x
    flag=0
    left = 1


print(time.time()-start)<jupyter_output><empty_output><jupyter_text>Test Choose the desired Dataset (Normal, Attacked, Filtered) to verify the results.<jupyter_code>testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=filSet, batch_size=1, shuffle=False, num_workers=4)

for epoch in range(1):
    stats.testing.reset()
    tSt = datetime.now()
    # Testing loop.
    for i, (input, target, label) in enumerate(testLoader, 0):
      net.eval()
      with torch.no_grad():
        input  = input.to(device)
        target = target.to(device) 
    
      output = net.forward(input)
      stats.testing.correctSamples += torch.sum( snn.predict.getClass(output) == label ).data.item()
      stats.testing.numSamples     += len(label)

      loss = error.numSpikes(output, target)
      stats.testing.lossSum += loss.cpu().data.item()
      stats.print(epoch, i)<jupyter_output><empty_output><jupyter_text>BAF Analysis on clean input (original Dataset without perturbations)<jupyter_code>import numpy as np

def update(x,y,t,A,s):
  for i in range(x-s,x+s+1):
     for j in range(y-s,y+s+1):
        if not(i==x and j==y) and i>0 and j>0 and i<128 and j<128:
	        A[i][j] = t

testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
testLoader = DataLoader(dataset=testSet, batch_size=1, shuffle=False, num_workers=4)

for s in [1,2,3,4]:
  stats1 = snn.utils.stats()
  stats5 = snn.utils.stats()
  stats10 = snn.utils.stats()
  stats20 = snn.utils.stats()          

  for i in range(264):     
    input,target,label = testSet[i]    
    data= np.load('/content/Test/{}.npy'.format(i))
    data= data[data[:,3].argsort()] #sort elements by timestamp
    temp=np.zeros((128,128))
    real1=[]
    real5=[]
    real10=[]
    real20=[]
    for d in data:
      update(int(d[0]),int(d[1]),d[3],temp,s)
      if d[3]-temp[int(d[0])][int(d[1])]<1:
        real1.append(d)

      if d[3]-temp[int(d[0])][int(d[1])]<5:
        real5.append(d)   

      if d[3]-temp[int(d[0])][int(d[1])]<10:
        real10.append(d)

      if d[3]-temp[int(d[0])][int(d[1])]<20:
        real20.append(d)  

    real1=np.stack(real1, axis=0 )
    with open("/content/filtro/1.npy", "wb") as f:
      np.save(f,real1)
    input1 = snn.io.readNpSpikes(
        "/content/filtro/1.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input1 =  input1.to(device)
    output1 = net.forward(torch.unsqueeze(input1,dim=0))
    stats1.testing.correctSamples += torch.sum( snn.predict.getClass(output1) == label ).data.item()
    stats1.testing.numSamples     += 1
    #stats1.print(1,i)

    real5=np.stack(real5, axis=0 )
    with open("/content/filtro/5.npy", "wb") as f:
      np.save(f,real5)
    input5 = snn.io.readNpSpikes(
        "/content/filtro/5.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input5 =  input5.to(device)
    output5 = net.forward(torch.unsqueeze(input5,dim=0))
    stats5.testing.correctSamples += torch.sum( snn.predict.getClass(output5) == label ).data.item()
    stats5.testing.numSamples     += 1
    #stats5.print(1,i)

    real10=np.stack(real10, axis=0 )
    with open("/content/filtro/10.npy", "wb") as f:
      np.save(f,real10)
    input10 = snn.io.readNpSpikes(
        "/content/filtro/10.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input10 =  input10.to(device)
    output10 = net.forward(torch.unsqueeze(input10,dim=0))
    stats10.testing.correctSamples += torch.sum( snn.predict.getClass(output10) == label ).data.item()
    stats10.testing.numSamples     += 1
    #stats10.print(1,i)    

    real20=np.stack(real20, axis=0 )
    with open("/content/filtro/20.npy", "wb") as f:
      np.save(f,real20)
    input20= snn.io.readNpSpikes(
        "/content/filtro/20.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input20 =  input20.to(device)
    output20 = net.forward(torch.unsqueeze(input20,dim=0))
    stats20.testing.correctSamples += torch.sum( snn.predict.getClass(output20) == label ).data.item()
    stats20.testing.numSamples     += 1
  stats1.print(1,i)
  stats5.print(1,i)
  stats10.print(1,i)
  stats20.print(1,i)<jupyter_output><empty_output><jupyter_text>Adversarial examples tested on the SNN, after being filtered by several BAFs<jupyter_code>import numpy as np

def update(x,y,t,A,s):
  for i in range(x-s,x+s+1):
     for j in range(y-s,y+s+1):
        if not(i==x and j==y) and i>=0 and j>=0 and i<128 and j<128:
	        A[i][j] = t

stats1 = snn.utils.stats()
stats5 = snn.utils.stats()
stats10 = snn.utils.stats()
stats20 = snn.utils.stats()          
for s in [1,2,3]:
  for i in range(264):     
    input,target,label = advSet[i]    
    data= np.load('/content/prova/{}.npy'.format(i))
    data= data[data[:,3].argsort()] #sort elements by timestamp
    temp=np.zeros((128,128))
    real1=[]
    real5=[]
    real10=[]
    real20=[]
    for d in data:
      update(int(d[0]),int(d[1]),d[3],temp,s)
      if d[3]-temp[int(d[0])][int(d[1])]<1:
        real1.append(d)

      if d[3]-temp[int(d[0])][int(d[1])]<5:
        real5.append(d)   

      if d[3]-temp[int(d[0])][int(d[1])]<10:
        real10.append(d)

      if d[3]-temp[int(d[0])][int(d[1])]<20:
        real20.append(d)  

    real1=np.stack(real1, axis=0 )
    with open("/content/filtro/1.npy", "wb") as f:
      np.save(f,real1)
    input1 = snn.io.readNpSpikes(
        "/content/filtro/1.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input1 =  input1.to(device)
    output1 = net.forward(torch.unsqueeze(input1,dim=0))
    stats1.testing.correctSamples += torch.sum( snn.predict.getClass(output1) == label ).data.item()
    stats1.testing.numSamples     += 1
    stats1.print(1,i)

    real5=np.stack(real5, axis=0 )
    with open("/content/filtro/5.npy", "wb") as f:
      np.save(f,real5)
    input5 = snn.io.readNpSpikes(
        "/content/filtro/5.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input5 =  input5.to(device)
    output5 = net.forward(torch.unsqueeze(input5,dim=0))
    stats5.testing.correctSamples += torch.sum( snn.predict.getClass(output5) == label ).data.item()
    stats5.testing.numSamples     += 1
    stats5.print(1,i)

    real10=np.stack(real10, axis=0 )
    with open("/content/filtro/10.npy", "wb") as f:
      np.save(f,real10)
    input10 = snn.io.readNpSpikes(
        "/content/filtro/10.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input10 =  input10.to(device)
    output10 = net.forward(torch.unsqueeze(input10,dim=0))
    stats10.testing.correctSamples += torch.sum( snn.predict.getClass(output10) == label ).data.item()
    stats10.testing.numSamples     += 1
    stats10.print(1,i)    

    real20=np.stack(real20, axis=0 )
    with open("/content/filtro/20.npy", "wb") as f:
      np.save(f,real20)
    input20= snn.io.readNpSpikes(
        "/content/filtro/20.npy").toSpikeTensor(torch.zeros((2,128,128,1450)), samplingTime=1.0)
    input20 =  input20.to(device)
    output20 = net.forward(torch.unsqueeze(input20,dim=0))
    stats20.testing.correctSamples += torch.sum( snn.predict.getClass(output20) == label ).data.item()
    stats20.testing.numSamples     += 1
    stats20.print(1,i)<jupyter_output><empty_output><jupyter_text>Mask Filter Analysis on clean input (original Dataset without perturbations)<jupyter_code>testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)

stats.testing.reset()
tSt = datetime.now()
th = [25,50,75,100,125,150,175,200]
for T in th:
  for i in range(264):
    input, target, label = testSet[i]
    activity= torch.sum(input, dim=3)
    activity= torch.sum(activity, dim=0)
    mask=torch.where((activity>T),0,1)
    mask=mask.expand(2,128,128)
    inp= input[:,:,:,:]*mask.unsqueeze(dim=3)
    net.eval()
    with torch.no_grad():
      inp  = inp.to(device)
      target = target.to(device) 

    output = net.forward(torch.unsqueeze(inp,dim=0))
    stats.testing.correctSamples += torch.sum( snn.predict.getClass(output) == label ).data.item()
    stats.testing.numSamples     += 1

    stats.print(1,i)
  MF = stats.testing.accuracy()
  print(T,MF)<jupyter_output><empty_output><jupyter_text>Adversarial examples tested on the SNN, after being filtered by several MFS<jupyter_code>advSet = IBMGestureDataset(datasetPath  =  '/content/prova/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)
advLoader = DataLoader(dataset=advSet, batch_size=1, shuffle=False, num_workers=4)

th = [25,50,75,100,125,150,175,200]
for T in th:
  stats.testing.reset()
  tSt = datetime.now()
  for i in range(264):
    input, target, label = advSet[i]
    activity= torch.sum(input, dim=3)
    activity= torch.sum(activity, dim=0)
    mask=torch.where((activity>T),0,1)
    mask=mask.expand(2,128,128)
    inp= input[:,:,:,:]*mask.unsqueeze(dim=3)
    net.eval()
    with torch.no_grad():
      inp  = inp.to(device)
      target = target.to(device) 

    output = net.forward(torch.unsqueeze(inp,dim=0))
    stats.testing.correctSamples += torch.sum( snn.predict.getClass(output) == label ).data.item()
    stats.testing.numSamples     += 1

    stats.print(1,i)
  MF = stats.testing.accuracy()
  print(T,MF)<jupyter_output><empty_output><jupyter_text>VISUALIZE<jupyter_code># Visualize the input spikes (first five samples).
testSet = IBMGestureDataset(datasetPath  =  '/content/Test/', 
                              sampleFile  ='/content/Test/test.txt',
                              samplingTime=1.0,
                              sampleLength=1450)

from IPython.display import HTML
for i in range(1):
    input, target, label = testSet[i]
    snn.io.showTD(snn.io.spikeArrayToEvent(input.reshape((2, 128, 128, -1)).cpu().data.numpy()))
anim = snn.io.animTD(snn.io.spikeArrayToEvent(input.reshape((2, 128, 128, -1)).cpu().data.numpy()))
HTML(anim.to_jshtml())<jupyter_output><empty_output><jupyter_text>Normal noise injection + BAF<jupyter_code>def update(x,y,t,A,s):
  for i in range(x-s,x+s+1):
     for j in range(y-s,y+s+1):
        if not(i==x and j==y) and i>0 and j>0 and i<127 and j<127:
	        A[i][j] = t

np.random.seed(7)
mask = np.where(np.random.uniform(size=(1,2,128,128,1450))<0.01,1,0)
np.random.seed(1)
normal = np.random.normal(size=(1,2,128,128,1450))
noise= torch.Tensor(mask*abs(normal))
amp=[1 ,0.85, 0.70, 0.55, 0.40, 0.25, 0.10]
for a in amp:
  net.eval()
  stats.testing.reset()
  tSt = datetime.now()
  print(tSt,a)
  # Testing loop.
  for i, (input, target, label) in enumerate(testLoader, 0):
    input= torch.clamp(input+a*noise,0,1)
    #print(i)
    input=torch.reshape(input,(1450,128,128,2))
    index=np.nonzero(input)
    temp= np.zeros((128,128))
    for i in index.numpy():
      update(i[1],i[2],i[0],temp)
      if i[0]-temp[i[1]][i[2]]>=5:
        input[i[0],i[1],i[2],i[3]]= 0.
    with torch.no_grad():
      input  = input.to(device)
      target = target.to(device)
    input=torch.reshape(input,(1,2,128,128,1450))
    output = net.forward(input)
    # Stats updating		
    stats.testing.correctSamples += torch.sum(snn.predict.getClass(output) == label).data.item()
    stats.testing.numSamples     += len(label)
    loss = error.numSpikes(output, target)
    stats.testing.lossSum += loss.cpu().data.item()
  tEnd=datetime.now()
  print(tEnd)
  print(a,stats.testing.accuracy())<jupyter_output><empty_output><jupyter_text>Normal noise injection + MF<jupyter_code>np.random.seed(7)
mask = np.where(np.random.uniform(size=(1,2,128,128,1450))<0.01,1,0)
np.random.seed(1)
normal = np.random.normal(size=(1,2,128,128,1450))
noise= torch.Tensor(mask*abs(normal))
amp=[1 ,0.85, 0.70, 0.55, 0.40, 0.25, 0.10]
th = [25,50,75,100,125,150,175,200]
for T in th:
  for a in amp:
    net.eval()
    stats.testing.reset()
    tSt = datetime.now()
    print(tSt,a)
    # Testing loop.
    for i, (input, target, label) in enumerate(testLoader, 0):
      input= torch.clamp(input+a*noise,0,1)
      activity= torch.sum(input, dim=3)
      activity= torch.sum(activity, dim=0)
      mask=torch.where((activity>T),0,1)
      mask=mask.expand(2,128,128)
      inp= input[:,:,:,:]*mask.unsqueeze(dim=3)
      net.eval()
      with torch.no_grad():
        inp  = inp.to(device)
        target = target.to(device) 

      output = net.forward(torch.unsqueeze(inp,dim=0))
      stats.testing.correctSamples += torch.sum( snn.predict.getClass(output) == label ).data.item()
      stats.testing.numSamples     += 1

      stats.print(1,i)
    MF = stats.testing.accuracy()
    print(T,a,MF)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15427757500222558, 'fraction_numerical': 0.04026825721831508, 'mean_word_length': 2.9479850046860356, 'pattern_counts': {'":': 0, '<': 98, '<?xml version=': 0, '>': 84, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '519193', 'license': 'MIT', 'n_tokens_mistral': 13292, 'n_tokens_neox': 12484, 'n_words': 2276, 'path': 'DVS128Gesture/DVS128GestureAttacks.ipynb', 'repo_name': 'albertomarchisio/DVS-Attacks', 'size': '33699'}