<jupyter_start><jupyter_text>ArcFaceを実装する paperhttps://arxiv.org/abs/1801.07698 参考URLhttps://qiita.com/noritsugu_yamada/items/2e049cd7a8fd77eee0f5<jupyter_code>%matplotlib inline
import keras
from keras.utils import to_categorical
from keras.datasets import mnist
import matplotlib.pyplot as plt
import numpy as np
(x_train, y_train), (x_test, y_test) = mnist.load_data()
def show_image(image_data):
    plt.figure(figsize=(16,8))
    plt.imshow(image_data)
    plt.show()
# データを正規化する
#show_image(x_train[0])
x_train = np.expand_dims(x_train,axis=3)
x_test = np.expand_dims(x_test,axis=3)
x_train = x_train.astype('float32')
x_test = x_test.astype('float32')
x_train = x_train / 255.
x_test = x_test / 255.
print(x_train[0].shape)
#show_image(x_train[0])
# one-hotベクトル化
num_classes = 10
print(y_train[0])
y_train = to_categorical(y_train,num_classes)
y_test = to_categorical(y_test,num_classes)
print(y_train[0])
from keras.models import Sequential
from keras.layers import Conv2D,MaxPooling2D,Dropout

def create_feature_extraction_model():
    inputs = Input(shape=(28,28,1)) 
    base_model = Sequential()
    base_model.add(Conv2D(32,kernel_size=(3,3),activation='relu',input_shape=(28,28,1)))
    base_model.add(Conv2D(64, (3, 3), activation='relu'))
    base_model.add(MaxPooling2D(pool_size=(2, 2)))
    base_model.add(Dropout(0.25))
    #base_model.add(Conv2D(2, (3, 3), activation='relu'))
    #base_model.compile(optimizer='adam',
    #          loss='categorical_crossentropy',
    #          metrics=['accuracy'])
    #base_model.summary()
    return base_model
# オリジナルレイヤーの定義
# https://keras.io/ja/layers/writing-your-own-keras-layers/
from keras import backend as K
from keras.engine.topology import Layer
import numpy as np
import tensorflow as tf

class ArcFace(Layer):
    def __init__(self, output_dim,margin,scale, **kwargs):
        self.output_dim = output_dim
        self.m = margin
        self.s = scale
        super(ArcFace, self).__init__(**kwargs)

    def build(self, input_shape):
        '''
        input[0] : last layer's feature
        input[1] : y_label
        '''
        # Create a trainable weight variable for this layer.
        self.kernel = self.add_weight(name='kernel',
                                      shape=(input_shape[0][1], self.output_dim),
                                      initializer='uniform',
                                      trainable=True)
        super(ArcFace, self).build(input_shape)  # Be sure to call this somewhere!

    def call(self, x):
        # 正解ラベル
        y = x[1]
        y_label = tf.argmax(y,axis=1)
        # 正規化（単位ベクトルに変換）
        x_normalize = K.l2_normalize(x[0],axis=0)
        weights_normalize = K.l2_normalize(self.kernel,axis=1)
        # サンプルデータの特徴ベクトルと各クラスの代表ベクトルとの内積を計算
        # これにより特徴ベクトルと各代表ベクトルとのcosθがわかる
        cosines = K.dot(x_normalize,weights_normalize)

        # 正解クラスに対応するcosθのみ、その値をcos(θ + m(argin))に変更する
        # 加法定理
        # cos(θ + m) = cos(θ)cos(m) - sin(θ)sin(m)
        #cos_theta = [cosines[i] for i in y if i == 1]
        #cos_theta = cosines[y_label]
        cos_m = K.cos(self.m)
        #sin_theta = K.sqrt(1 - K.square(target_cos))
        sin_theta = K.sqrt(1 - K.square(cosines))
        sin_m = K.sin(self.m)
        cos_theta_plus_m = cosines * cos_m - sin_theta * sin_m
        
        # 正解クラスに対応するcosθだけ入れ替える
        #output = [cosines[i] if y[i] == 0 else cos_theta_plus_m for i in range(len(y))]
        # y * cosines : 正解ラベルに対応するものだけ残る
        # (1 - y) * cosines : その他のラベルに対応するものはそのまま残す
        output = (y * cos_theta_plus_m) + ((1 - y) * cosines)
        
        # softmaxにかける前に出力をself.s倍する
        output = output * self.s
        
        return output

    def compute_output_shape(self, input_shape):
        return (input_shape[0][0], self.output_dim)

from keras.models import Model
from keras.layers import Input,Activation,GlobalAveragePooling2D

def create_arcface_model(base_model,n_classes,margin,scale,model_file=None):
    x = base_model.output
    yinput = Input(shape=(n_classes,)) #ArcFaceで使用
    features = GlobalAveragePooling2D()(x) 
    x = ArcFace(n_classes,margin , scale)([features,yinput]) #outputをクラス数と同じ数に
    prediction = Activation('softmax')(x)
    arcface_model = Model(inputs=[base_model.input,yinput],outputs=prediction)
    
    # 重みをloadするとき
    if model_file:
        arcface_model.load_weights(model_file)
        print('load model : '.format(model_file))
        
    return arcface_model
def create_model(n_classes,margin,scale):
    base_model = create_feature_extraction_model()
    arcface_model = create_arcface_model(base_model,n_classes,margin,scale)
    return arcface_model
# ArcFaceのパラメータ
n_classes = 10
margin = 0.05
scale = 30
arcface_model = create_model(n_classes,margin,scale)
arcface_model.summary()
arcface_model.compile(optimizer='adam',
          loss='categorical_crossentropy',
          metrics=['accuracy'])<jupyter_output>W0716 23:43:40.561098 4540978624 deprecation.py:506] From /anaconda3/envs/arcface/lib/python3.7/site-packages/keras/backend/tensorflow_backend.py:3445: calling dropout (from tensorflow.python.ops.nn_ops) with keep_prob is deprecated and will be removed in a future version.
Instructions for updating:
Please use `rate` instead of `keep_prob`. Rate should be set to `rate = 1 - keep_prob`.<jupyter_text>学習<jupyter_code>from keras import callbacks

# パラメータ
batch_size = 32
epochs = 20
verbose = 1
validation_split = 0.1
model_path = 'arcface_margin{}_scale{}_mnist_64features.hdf5'.format(margin,scale)

callbacks_list = [
        #バリデーションlossが改善したらモデルをsave
        callbacks.ModelCheckpoint(
        filepath=model_path,
        monitor="val_loss",
        save_weights_only=True,
        save_best_only=True),

        #バリデーションlossが改善しなくなったら学習率を変更する
        callbacks.ReduceLROnPlateau(
            monitor="val_loss",
            factor=0.8,
            patience=5,
            verbose=1)]
history = arcface_model.fit([x_train,y_train],y_train,batch_size=batch_size,epochs=epochs,verbose=verbose,validation_split=validation_split,callbacks=callbacks_list)
# 結果表示
plt.figure(figsize=(16,8))
plt.subplot(121)
plt.plot(history.history['acc'],label='acc')
plt.plot(history.history['val_acc'],label='val_acc')
plt.legend()
plt.subplot(122)
plt.plot(history.history['loss'],label='loss')
plt.plot(history.history['val_loss'],label='val_loss')
plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>推論<jupyter_code># predict用のmodel. ArcFaceLayerを除き,重みはloadして使用する
def create_predict_model(n_classes,s,m, model_file):
    base_model = create_feature_extraction_model()
    arcface_model = create_arcface_model(base_model,n_classes,m,s, model_file) #学習と同じlayer数
    predict_model = Model(arcface_model.get_layer(index=0).input, arcface_model.get_layer(index=-4).output)
    predict_model.summary()

    return predict_model
predict_model = Model(arcface_model.get_layer(index=0).input, arcface_model.get_layer(index=5).output)
predict_model.summary()
n_classes = 10
s = 30
m = 0.1
model_file = 'arcface_margin0.1_scale30_mnist_64features.hdf5'
predict_model = create_predict_model(n_classes,s,m,model_file)
#predict_model.summary()
batch_size = 32
feature_pred = predict_model.predict(x_test,batch_size=batch_size,verbose=1)
from sklearn.preprocessing import normalize
a = normalize(feature_pred,norm='l2')
np.linalg.norm(a[0],ord=2)<jupyter_output><empty_output><jupyter_text>答え合わせする<jupyter_code>from sklearn.preprocessing import normalize

def calc_confusion_matrix(pred,typical_vecs,y_test,num_classes):
    confusion_matrix = np.zeros((num_classes,num_classes))
    label_pred = []
    # l2正規化
    pred_normalize = normalize(pred,norm='l2')
    typical_vecs_normalize = normalize(typical_vecs,norm='l2')
    print(np.linalg.norm(pred_normalize[0],ord=2))
    print(np.linalg.norm(typical_vecs_normalize[0],ord=2))
    # 推論結果と各クラスの代表ベクトルとのコサイン類似度を計算してもっとも高い値のクラスとする
    init = True
    for i,pred_norm in enumerate(pred_normalize):
        #cos = np.dot(pred_norm,typical_vecs_normalize)
        cos = np.dot(typical_vecs_normalize,pred_norm)
        if init:
            print(pred_norm.shape)
            print(typical_vecs_normalize.shape)
            print(cos.shape)
            print(cos)
            print(np.argmax(y_test[i]))
            
            init = False
        cos_max = np.argmax(cos)
        label_pred.append(cos_max)
        correct_label = np.argmax(y_test[i])
        confusion_matrix[correct_label][cos_max] += 1
    return confusion_matrix,label_pred
    
# 答え合わせ
typical_vecs = np.array(arcface_model.get_layer(index=-2).get_weights())[0]
confusion_matrix,label_pred = calc_confusion_matrix(feature_pred,typical_vecs,y_test,num_classes)
confusion_matrix[0]<jupyter_output><empty_output><jupyter_text>検証<jupyter_code>from sklearn.preprocessing import normalize
# l2正規化
feature_pred_normalize = normalize(feature_pred,norm='l2')
#feature_pred_normalize = np.l2_normalize(feature_pred)<jupyter_output><empty_output><jupyter_text>特徴量が２次元（または３次元）の場合は可視化できるので可視化したい２次元の場合は円周上にクラス毎に分かれてプロットされることを期待している  ３次元の場合は球上にクラス毎に分かれてプロットされることを期待している<jupyter_code>import matplotlib.cm as cm
# 可視化（６４次元なので意味なし）
y_test_label = np.argmax(y_test,axis=1)
plt.figure(figsize=(10,10))
plt.scatter(feature_pred_normalize[:,0],feature_pred_normalize[:,1],c=y_test_label,cmap=cm.seismic)
plt.legend()
plt.show()<jupyter_output>W0717 19:19:21.714380 4540978624 legend.py:1282] No handles with labels found to put in legend.<jupyter_text>６４次元のある単位ベクトルeとテストデータのコサイン類似度を調べる所属するクラス毎にeとのコサイン類似度を調べてヒストグラムで表示<jupyter_code>base_vec = np.array([1 if i == 0 else 0 for i in range(feature_pred.shape[1])])
y_test_label = np.argmax(y_test,axis=1)
cosines = []
for i,feature in enumerate(feature_pred_normalize):
    cos = np.dot(base_vec,feature)
    y = y_test_label[i]
    cosines.append([y,cos])
    
hist = []
num_classes = 10
for i in range(num_classes):
    targets = [v[1] for v in cosines if v[0] == i]
    hist.append(targets)<jupyter_output><empty_output><jupyter_text>クラスごとに結果表示<jupyter_code># クラス別にある単位ベクトルeとのコサイン類似度のヒストグラムを表示
import seaborn as sns
rows = num_classes / 2
cols = 2
plt.figure(figsize=(16,8))
plt.subplots_adjust(wspace=0.4, hspace=0.6)
for i in range(num_classes):
    plt.subplot(rows,cols,i+1)
    sns.distplot(hist[i], kde=False, rug=False, bins=20) 
    plt.title(i)
plt.show()<jupyter_output><empty_output><jupyter_text>全クラス重ねて表示ほぼ0付近なので今回のベースモデルとはまったく近くないということになる<jupyter_code># 全クラス重ねて表示してみる
import seaborn as sns
plt.figure(figsize=(16,8))
for i in range(num_classes):
    sns.distplot(hist[i], kde=False, rug=False, bins=20,label=i) 
plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>各クラスの代表ベクトル間のコサイン類似度を計算してみる<jupyter_code># 各クラスの代表ベクトル間のコサイン類似度を計算してみる
from sklearn.preprocessing import normalize
typical_vecs = arcface_model.get_layer(name='arc_face_2').get_weights()
typical_vecs_normalize = normalize(typical_vecs[0],norm='l2')
num_classes = 10
result = []
for i in range(num_classes):
    vec_i_normalize = typical_vecs[0][:,i] / np.linalg.norm(typical_vecs[0][:,i],ord=2)
    vec_i = vec_i_normalize
    for j in range(num_classes):
        vec_j_normalize = typical_vecs[0][:,j] / np.linalg.norm(typical_vecs[0][:,j],ord=2)
        vec_j = vec_j_normalize
        result.append(np.dot(vec_i,vec_j))
result_2d = np.reshape(np.array(result),(10,10))
plt.figure(figsize=(8,7))
plt.title('cosine similarity between typical vecs(MNIST)')
sns.heatmap(result_2d)<jupyter_output><empty_output><jupyter_text>knnグラフ構築してみる<jupyter_code>def save_graph_to_csv(G,csv_file_path):
    with open(csv_file_path,'w') as f:
        array = G.toarray()
        for i in range(array.shape[0]):
            content = str(i)
            for connected in np.where(array[i] == 1)[0]:
                content += ',{}'.format(connected)
            content += '\n'
            f.write(content)
from sklearn.neighbors import kneighbors_graph
neighbors = 3
G = kneighbors_graph(feature_pred_normalize,neighbors)
save_graph_to_csv(G,'./graph_csv/arcface_knn3.csv')<jupyter_output><empty_output><jupyter_text>radius neighbors graph作ってみる<jupyter_code>from sklearn.neighbors import radius_neighbors_graph
radius = 0.05
G = radius_neighbors_graph(feature_pred_normalize,radius)
save_graph_to_csv(G,'./graph_csv/arcface_rn05.csv')
a
typical_vecs[0][:,1]<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11891059414768582, 'fraction_numerical': 0.02439815190078625, 'mean_word_length': 4.552655265526552, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '199172', 'license': 'MIT', 'n_tokens_mistral': 5221, 'n_tokens_neox': 5027, 'n_words': 663, 'path': 'notebook/ArcFace_feature64d.ipynb', 'repo_name': 'takecx/ArcFace-Keras', 'size': '12337'}