1
2
3
4
5
6
7
8
9
10
11
12
import numpy as np 
import matplotlib.pyplot as plt
from sklearn.datasets import load_digits

digits = load_digits()

plt.figure(figsize=(20,4))
for index, (image, label) in enumerate(zip(digits.data[0:5], digits.target[0:5])):
  plt.subplot(1, 5, index + 1)
  plt.imshow(np.reshape(image, (8,8)), cmap=plt.cm.gray)
  plt.title('Training: %i\n' % label, fontsize = 20)
plt.show()

1
print(np.reshape(digits.data[0], (8, 8)))

1
2
3
4
5
plt.figure(figsize=(20,4))
for index, (image, label) in enumerate(zip(digits.data[0:1], digits.target[0:5])):
  plt.subplot(1, 1, index + 1)
  plt.imshow(np.reshape(image, (8,8)), cmap=plt.cm.gray)
plt.show()

1
2
3
4
5
plt.figure(figsize=(20, 4)) 
for index, image in enumerate(partesDigito[0][1]):
  plt.subplot(1, len(partesDigito[0][1]), index + 1)
  plt.imshow(image, cmap=plt.cm.gray)
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
en = 8 # tamanho da dimensão da entrada
kn = 3 # tamanho da dimensão do kernel
k = 10 # quantidade de kernels
qtdMatrizes = (en - kn) ** 2
kernels = np.random.randn(k, kn, kn)

def partes(x):
    partesDigito = []
    entrada = np.zeros(((en - kn) ** 2, kn, kn))
    X = x.reshape(en, en) # Muda a entrada vetorial para matriz
    for ki in range(0, k): # Para cada kernel separa a entrada em partes
        index = 0;
        for i in range(0, en - kn): 
            for j in range(0, en - kn):
                entrada[index] = X[i:i+kn, j:j+kn] # Cada parte é uma submatriz da entrada
                index += 1
        partesDigito.append((ki, entrada)) # Para cada kernel gera varias partes da entrada
    return partesDigito

partesDigito = partes(digits.data[0])
print(partesDigito[0])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import math

def sigmoid(x):
  return 1 / (1 + math.exp(-x))

# t representa a dimensão de um mapa de ativação
t = en - kn + 1

# Aplica a função de ativação no resultado da multiplicação do kernel pela entrada
# Recebe como entrada o kernel e uma matriz com as partes da entrada
def ativacao(k, x):
    mapa = np.zeros((t, t)) #Mapa de ativações de um kernel vazio
    for i in range(0, len(x) - 1): #Para cada parte da matriz da entrada
        mapa[int(i/t)][i%t] = sigmoid((x[i] * k).sum())
    return mapa

print("Mapa de ativações de um kernel")
print(ativacao(kernels[0], partesDigito[0][1]))

1
2
3
4
5
6
7
8
9
10
mapas = np.zeros((k, t, t))

for i in range(0, k):
    mapas[i] = ativacao(kernels[partesDigito[i][0]], partesDigito[i][1])

plt.figure(figsize=(20, 4))
for index, image in enumerate(mapas):
  plt.subplot(1, len(mapas), index + 1)
  plt.imshow(image, cmap=plt.cm.gray)
plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
pn = 2 # tamanho do pooling
ps = 2 # tamanho do passo do pooling

# Max Pooling
def maxPooling(x):
    agrupado = np.zeros((int(t / pn), int(t / pn)))
    for i in range(0, int(t / pn)):
        for j in range(0, int(t / pn)):
            agrupado[i][j] = x[i * pn : i * pn + kn - 1, j * pn : j * pn + kn - 1].max()
    return agrupado

pooling = []
for i in range(0, len(mapas)):
    pooling.append(maxPooling(mapas[i]))
    
print(pooling[0])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# weights representa os pesos da camada densa (MLP), aqui ainda falta fazer o backpropagation para ajustar esses pesos.
weights = np.random.randn(10, 90)

# função de ativação RELU
def relu(x):
    return max(0, x)

# Multilayer Perceptron
def mlp(x):
    # A saída é um vetor que será passado para a camada de saída.
    saida = []
    for w in weights:
        saida.append(relu((w * x).sum()))
    return saida

print (mlp(np.asarray(pooling).ravel()))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# weights representa os pesos da camada de saída, aqui ainda falta fazer o backpropagation para ajustar esses pesos.
weightsOut = np.random.randn(10, 1)

# função de ativação SoftMax
def softmax(x):
    e_x = np.exp(x - np.max(x))
    return e_x / e_x.sum(axis=0)

def out(x):
    # A saída representa a probabilidade de cada número entre zero e nove.
    saida = []
    for i in range(0, len(weightsOut)):
        saida.append((weightsOut[i] * x).sum())
    return softmax(saida)

x = mlp(np.asarray(pooling).ravel())
print(out(x))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
sc = SparkContext.getOrCreate()
# Montando um RDD com todo dataset de digitos
rdd = sc.parallelize(digits.data, 4)

# Cria uma matriz com os kernels (filtros) da primeira camada convolucional
kernels1 = np.random.randn(k, kn, kn)

print(rdd
      # 1º convolucao
      .map(lambda x: partes(x)) # Separa as 25 partes da imagem para cada um dos 10 kernels
      .map(lambda x: list(map(lambda y: ativacao(kernels1[y[0]], y[1]), x))) # Gera o mapa de ativações de cada kernel
      .map(lambda x: list(map(lambda y: maxPooling(y), x))) # Aplica o max pooling na primeira camada
      # Camada Densa (MLP)
      .map(lambda x: np.asarray(x).ravel()) # Converte a matriz de saída da camada convolucional para um vetor 
      .map(lambda x: mlp(x)) # Manda os dados da última camada convolucional agrupada para a camada Densa
      .map(lambda x: out(x)) # Aplica o softmax e retorna os valores das probabilidades de cada número. 
      .take(1))