TUKRの実装@德永

.gitignore

@@ -132,5 +132,6 @@ dmypy.json
 *.DS_Store
 
 .idea/
+
 face_project/datastore/
 Lecture_UKR/fukunaga/datestore/

Lecture_TUKR/tokunaga/cross_test.py

@@ -0,0 +1,46 @@
+#sampleフォルダのukr_torchとクロステストするプログラム
+
+import numpy as np
+from Lecture_UKR.tokunaga.ukr import UKR as ukr
+from Lecture_UKR.sample.ukr_torch import UKR as ukr_sample
+from Lecture_UKR.tokunaga.data import create_kura
+
+class TestUKR():
+    def test_UKR(self):
+        epoch = 100
+        sigma = 0.2
+        eta = 100
+        latent_dim = 2
+        nb_samples = 100
+        seed = 4
+        np.random.seed(seed)
+
+        X = create_kura(nb_samples)
+        Zinit = 0.2 * sigma * np.random.rand(nb_samples, latent_dim) - 0.1 * sigma
+
+        #サンプルプログラム
+        ukr_sam = ukr_sample(X, latent_dim, sigma, prior='random', Zinit=Zinit)
+        ukr_sam.fit(epoch, eta)
+
+        #作ったプログラム
+        ukr_mine = ukr(X, latent_dim, sigma, prior='random', Zinit=Zinit)
+        ukr_mine.fit(epoch, eta)
+
+        print("You are executing ", __file__)
+
+        np.testing.assert_allclose(
+            ukr_sam.history['z'],
+            ukr_mine.history['z'],
+            rtol=1e-05,
+            atol=1e-08)
+        np.testing.assert_allclose(
+            ukr_sam.history['f'],
+            ukr_mine.history['f'],
+            rtol=1e-05,
+            atol=1e-08)
+
+
+if __name__ == "__main__":
+    test_ukr = TestUKR()
+    test_ukr.test_UKR()
+    #unittest.main()

Lecture_TUKR/tokunaga/data.py

@@ -0,0 +1,103 @@
+import numpy as np
+import matplotlib.pyplot as plt
+
+def load_kura_tsom(xsamples, ysamples, missing_rate=None,retz=None):
+    z1 = np.linspace(-1, 1, xsamples)
+    z2 = np.linspace(-1, 1, ysamples)
+
+    #z1 = np.random.uniform(-1, 1, xsamples)
+    #z2 = np.random.uniform(-1, 1, ysamples)
+
+    z1_repeated, z2_repeated = np.meshgrid(z1, z2, indexing='ij')
+    x1 = z1_repeated
+    x2 = z2_repeated
+    x3 = 1 * (x1**2 - x2**2) + np.random.uniform(-0.5, 0.5, xsamples*ysamples).reshape(xsamples, ysamples)
+    #ノイズを加えたい時はここをいじる,locがガウス分布の平均、scaleが分散,size何個ノイズを作るか
+    #このノイズを加えることによって三次元空間のデータ点は上下に動く
+    x = np.concatenate([x1[:, :, None], x2[:, :, None], x3[:, :, None]], axis=2)
+    truez = np.concatenate([x1[:, None], x2[:, None]], axis=2)
+
+    if missing_rate == 0 or missing_rate == None:
+        if retz:
+            return x, truez
+        else:
+            return x
+
+def load_kura_list(xsamples, ysamples, missing_rate=None, retz=None):
+    z1 = np.random.uniform(-1, 1, xsamples)
+    z2 = np.random.uniform(-1, 1, ysamples)
+    z1_num = np.arange(xsamples)
+    z2_num = np.arange(ysamples)
+
+    z1_repeated, z2_repeated = np.meshgrid(z1, z2)
+    z1_num_repeated, z2_num_repeated = np.meshgrid(z1_num, z2_num)
+    x1 = z1_repeated.reshape(-1)
+    x2 = z2_repeated.reshape(-1)
+    x1_num = z1_num_repeated.reshape(-1)
+    x2_num = z2_num_repeated.reshape(-1)
+    x3 = 1.0*(x1**2 - x2**2)
+    #ノイズを加えたい時はここをいじる,locがガウス分布の平均、scaleが分散,size何個ノイズを作るか
+    #このノイズを加えることによって三次元空間のデータ点は上下に動く
+
+    x = np.concatenate([x1[:, None], x2[:, None], x3[:, None]], axis=1)
+    x = x + np.random.normal(0, 0.1, x.shape)
+    x_num = np.concatenate([x1_num[:, None], x2_num[:, None]], axis=1)
+
+    if missing_rate == 0 or missing_rate == None:
+        if retz:
+            return x
+        else:
+            return x, x_num
+
+def load_kura_lost_list(xsamples, ysamples, missing_rate=None, retz=None):
+    z1 = np.random.uniform(-1, 1, xsamples)
+    z2 = np.random.uniform(-1, 1, ysamples)
+    z1_num = np.arange(xsamples)
+    z2_num = np.arange(ysamples)
+
+    z1_repeated, z2_repeated = np.meshgrid(z1, z2)
+    z1_num_repeated, z2_num_repeated = np.meshgrid(z1_num, z2_num)
+    x1 = z1_repeated.reshape(-1)
+    x2 = z2_repeated.reshape(-1)
+    x1_num = z1_num_repeated.reshape(-1)
+    x2_num = z2_num_repeated.reshape(-1)
+    x3 = 0.5*(x1**2 - x2**2)
+    #ノイズを加えたい時はここをいじる,locがガウス分布の平均、scaleが分散,size何個ノイズを作るか
+    #このノイズを加えることによって三次元空間のデータ点は上下に動く
+
+    x = np.concatenate([x1[:, None], x2[:, None], x3[:, None]], axis=1)
+    #x = np.concatenate([x1, x2, x3])
+    x = x + np.random.normal(0, 0.1, (xsamples * ysamples, 3))
+    x_num = np.concatenate([x1_num[:, None], x2_num[:, None]], axis=1)
+
+    #del_am = 10
+    #del_am = 2*(xsamples+ysamples)
+    #del_am = xsamples*ysamples
+    del_num = np.arange(xsamples*ysamples)
+    np.random.shuffle(del_num)
+    #del_num = np.random.randint(0, xsamples*ysamples)
+    del_am = xsamples*ysamples - 50
+    x = np.delete(x, del_num[: del_am], 0)
+    x_num = np.delete(x_num, del_num[: del_am], 0)
+
+    if missing_rate == 0 or missing_rate == None:
+        if retz:
+            return x
+        else:
+            return x, x_num
+
+if __name__ == '__main__':
+    import matplotlib.pyplot as plt
+    from mpl_toolkits.mplot3d import Axes3D
+
+    xsamples = 10
+    ysamples = 15
+
+    x = load_kura_tsom(xsamples, ysamples)
+    #x = load_kura_lost_list(xsamples, ysamples)[0]
+    #print(plt.rcParams['image.cmap'])
+    fig = plt.figure(figsize=[5, 5])
+    ax_x = fig.add_subplot(projection='3d')
+    ax_x.scatter(x[:, :, 0], x[:, :, 1], x[:, :, 2])
+    ax_x.set_title('Generated three-dimensional data')
+    plt.show()

Lecture_TUKR/tokunaga/l_visualizer.py

@@ -0,0 +1,102 @@
+import numpy as np
+from matplotlib import pyplot as plt
+from matplotlib.animation import FuncAnimation
+
+STEP = 150
+
+def visualize_history(X, Y_history, U_history, V_history, error_history, save_gif=False, filename="tmp"):
+    input_dim, latent_dim1, latent_dim2 = X.shape[1], U_history[0].shape[1], V_history[0].shape[1]
+    input_projection_type = '3d' if input_dim > 2 else 'rectilinear'
+
+    fig = plt.figure(figsize=(10, 8))
+    gs = fig.add_gridspec(3, 2)
+    input_ax = fig.add_subplot(gs[0:2, 0], projection=input_projection_type)
+    latent1_ax = fig.add_subplot(gs[0:1, 1], aspect='equal')
+    latent2_ax = fig.add_subplot(gs[1:2, 1], aspect='equal')
+    error_ax = fig.add_subplot(gs[2, :])
+    num_epoch = len(Y_history)
+
+    if input_dim == 3 and latent_dim1 == 2 and latent_dim2 == 2:
+        resolution = int(np.sqrt(Y_history.shape[1]))
+        if Y_history.shape[1] == resolution ** 2:
+            Y_history = np.array(Y_history).reshape((num_epoch, resolution, resolution, input_dim))
+
+    observable_drawer = [None, draw_observable_1D, draw_observable_2D, draw_observable_3D][input_dim]
+    latent1_drawer = [None, draw_latent_1D, draw_latent_2D][latent_dim1]
+    latent2_drawer = [None, draw_latent_1D, draw_latent_2D][latent_dim2]
+
+    ani = FuncAnimation(
+        fig,
+        update_graph,
+        frames=num_epoch,  # // STEP,
+        repeat=True,
+        interval=50,
+        fargs=(observable_drawer, latent1_drawer, latent2_drawer, X, Y_history, U_history, V_history, error_history, fig,
+               input_ax, latent1_ax, latent2_ax, error_ax, num_epoch))
+    plt.show()
+    if save_gif:
+        ani.save(f"{filename}.mp4", writer='ffmpeg')
+
+def update_graph(epoch, observable_drawer, latent1_drawer, latent2_drawer, X, Y_history,
+                 U_history, V_history, error_history, fig, input_ax, latent1_ax, latent2_ax, error_ax, num_epoch):
+    fig.suptitle(f"epoch: {epoch}")
+    input_ax.cla()
+    #  input_ax.view_init(azim=(epoch * 400 / num_epoch), elev=30)
+    latent1_ax.cla()
+    latent2_ax.cla()
+    error_ax.cla()
+
+    Y, U, V = Y_history[epoch], U_history[epoch], V_history[epoch]
+    colormapx = X[:, 0]
+    #colormap1 = X[X_num[:, 1] == 0][:, 0]
+    #colormap2 = X[X_num[:, 0] == 0][:, 1]
+    colormap1 = 'r'
+    colormap2 = 'b'
+        #X[:, 0]
+
+    observable_drawer(input_ax, X, Y, colormapx)
+    latent1_drawer(latent1_ax, U, colormap1)
+    latent2_drawer(latent2_ax, V, colormap2)
+    draw_error(error_ax, error_history, epoch)
+
+
+def draw_observable_3D(ax, X, Y, colormap):
+    ax.scatter(X[:, 0], X[:, 1], X[:, 2], c=colormap)
+    # ax.set_zlim(-1, 1)
+    if len(Y.shape) == 3:
+        ax.plot_wireframe(Y[:, :, 0], Y[:, :, 1], Y[:, :, 2], color='black')
+        # ax.scatter(Y[:, :, 0], Y[:, :, 1], Y[:, :, 2], color='black')
+    else:
+        ax.plot(Y[:, 0], Y[:, 1], Y[:, 2], color='black')
+# ax.plot(Y[:, 0], Y[:, 1], Y[:, 2], color='black')
+# ax.plot_wireframe(Y[:, :, 0], Y[:, :, 1], Y[:, :, 2], color='black')
+
+
+def draw_observable_2D(ax, X, Y, colormap):
+    ax.scatter(X[:, :, 0], X[:, :, 1], c=colormap)
+    ax.plot(Y[:, :, 0], Y[:, :, 1], c='black')
+
+def draw_observable_1D(ax, X, Y, colormap):
+    ax.scatter(X[:, :, 0], np.zeros(X.shape), c=colormap)
+    ax.plot(Y[:, :, 0], np.zeros((17, 21)), c='black')
+
+def draw_latent_2D(ax, Z, colormap):
+    # Z_max=np.amax(Z, axis=0)
+    # Z_min=np.amin(Z, axis=0)
+    # ax.set_xlim(Z_min[0] - 1, Z_max[0] + 1)
+    # ax.set_ylim(Z_min[1] - 1, Z_max[1] + 1)
+    ax.set_xlim(-1, 1)
+    ax.set_ylim(-1, 1)
+    ax.scatter(Z[:, 0], Z[:, 1], c=colormap)
+
+
+def draw_latent_1D(ax, Z, colormap):
+    ax.scatter(Z, np.zeros(Z.shape), c=colormap)
+    #ax.set_title('a')
+    ax.set_ylim(-1, 1)
+
+def draw_error(ax, error_history, epoch):
+    ax.set_ylim(0, np.max(error_history)+0.1)
+    ax.set_title("error_function", fontsize=8)
+    ax.plot(error_history, label='誤差関数')
+    ax.scatter(epoch, error_history[epoch], s=55, marker="*")

Lecture_TUKR/tokunaga/load.py

@@ -0,0 +1,57 @@
+import numpy as np
+import matplotlib.pyplot as plt
+import os
+import cv2
+
+
+def load_animal_data(retlabel_animal=True, retlabel_feature=True):
+    datastore_name = 'datastore/animal'
+    file_name = 'features.txt'
+
+    directory_path = os.path.join(os.path.dirname(__file__), datastore_name)
+    file_path = os.path.join(directory_path, file_name)
+
+    x = np.loadtxt(file_path)
+
+    return_objects = [x]
+
+    if retlabel_animal:
+        label_name = 'labels_animal.txt'
+        label_path = os.path.join(directory_path, label_name)
+        label_animal = np.genfromtxt(label_path, dtype=str)
+        return_objects.append(label_animal)
+
+    if retlabel_feature:
+        label_name = 'labels_feature.txt'
+        label_path = os.path.join(directory_path, label_name)
+        label_feature = np.genfromtxt(label_path, dtype=str)
+        return_objects.append(label_feature)
+
+    return return_objects
+
+# def load_angle_resized_data():
+#     datastore_name = 'datastore/Angle_resized/'
+#     dir_list = os.listdir(datastore_name)
+#     #file_name = '/-5/A_01_-05.jpg'
+#
+#     # directory_path = os.path.join(os.path.dirname(__file__), datastore_name)
+#     # file_path = os.path.join(directory_path, file_name)
+#
+#     dir_name = dir_list[0]
+#     img = []
+#     file_list = os.listdir(datastore_name + dir_name)
+#     for file_name in file_list:
+#         image = cv2.imread(datastore_name + dir_name + '/' + file_name)
+#         img.append(image)
+#     # img = cv2.imread(datastore_name + file_name)
+#     #
+#     # print(img)
+#     # plt.imshow(img)
+#     # plt.show()
+#
+#     return np.array(img)
+
+
+if __name__ == '__main__':
+    img = load_angle_resized_data()
+    print(img)