23-06-22

np random seed 고정
각 함수의 설명 추가

auto_tunning.py

@@ -14,7 +14,6 @@ from keras import backend as K
 
 # from pso_tf import PSO
 from pso import Optimizer
-# from optimizer import Optimizer
 
 import numpy as np
 
@@ -27,7 +26,6 @@ import gc
 # print(tf.config.list_physical_devices())
 # print(f"Num GPUs Available: {len(tf.config.list_physical_devices('GPU'))}")
 
-
 def get_data():
     (x_train, y_train), (x_test, y_test) = mnist.load_data()
 

iris.py

@@ -4,6 +4,9 @@ os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
 import tensorflow as tf
 tf.random.set_seed(777)  # for reproducibility
 
+import numpy as np
+np.random.seed(777)
+
 from sklearn.datasets import load_iris
 from sklearn.model_selection import train_test_split
 
@@ -37,11 +40,11 @@ def load_data():
 model = make_model()
 x_train, x_test, y_train, y_test = load_data()
 
-loss = 'categorical_crossentropy'
+loss = ['categorical_crossentropy', 'accuracy','mse']
 
 pso_iris = Optimizer(
     model,
-    loss=loss, 
+    loss=loss[0], 
     n_particles=75, 
     c0=0.4, 
     c1=0.8, 

mnist.py

@@ -5,6 +5,9 @@ os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
 import tensorflow as tf
 tf.random.set_seed(777)  # for reproducibility
 
+import numpy as np
+np.random.seed(777)
+
 from tensorflow import keras
 from keras.datasets import mnist
 from keras.models import Sequential
@@ -16,7 +19,6 @@ from keras import backend as K
 from pso import Optimizer
 # from optimizer import Optimizer
 
-import numpy as np
 
 from datetime import date
 from tqdm import tqdm
@@ -59,6 +61,8 @@ def make_model():
 
     return model
 
+# %%
+
 # %%
 model = make_model()
 x_test, y_test = get_data_test()

plt.ipynb

@@ -102,7 +102,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.9.16"
+   "version": "3.8.16"
   },
   "widgets": {
    "application/vnd.jupyter.widget-state+json": {

pso/optimizer.py

@@ -25,17 +25,6 @@ if gpus:
         print(e)
 
 class Optimizer:
-    """
-    Args:
-        model (keras.models): 모델 구조
-        loss (str): 손실함수
-        n_particles (int): 파티클 개수
-        c0 (float): local rate - 지역 최적값 관성 수치
-        c1 (float): global rate - 전역 최적값 관성 수치
-        w_min (float): 최소 관성 수치
-        w_max (float): 최대 관성 수치
-        nefative_swarm (float): 최적해와 반대로 이동할 파티클 비율 - 0 ~ 1 사이의 값
-    """
 
     def __init__(
         self,
@@ -48,6 +37,17 @@ class Optimizer:
         w_max=1.5,
         negative_swarm: float = 0,
     ):
+        """
+        Args:
+            model (keras.models): 모델 구조
+            loss (str): 손실함수
+            n_particles (int): 파티클 개수
+            c0 (float): local rate - 지역 최적값 관성 수치
+            c1 (float): global rate - 전역 최적값 관성 수치
+            w_min (float): 최소 관성 수치
+            w_max (float): 최대 관성 수치
+            nefative_swarm (float): 최적해와 반대로 이동할 파티클 비율 - 0 ~ 1 사이의 값
+        """
         self.model = model  # 모델 구조
         self.loss = loss  # 손실함수
         self.n_particles = n_particles  # 파티클 개수
@@ -91,7 +91,11 @@ class Optimizer:
         del self.avg_score
         gc.collect()
 
+
+    def _encode(self, weights):
         """
+        가중치를 1차원으로 풀어서 반환
+        
         Args:
             weights (list) : keras model의 가중치
         Returns:
@@ -99,8 +103,6 @@ class Optimizer:
             (list) : 가중치의 원본 shape
             (list) : 가중치의 원본 shape의 길이
         """
-
-    def _encode(self, weights):
         # w_gpu = cp.array([])
         w_gpu = np.array([])
         lenght = []
@@ -115,16 +117,19 @@ class Optimizer:
         del weights
         return w_gpu, shape, lenght
 
+
+    def _decode(self, weight, shape, lenght):
         """
+        _encode 로 인코딩된 가중치를 원본 shape으로 복원
+        파라미터는 encode의 리턴값을 그대로 사용을 권장
+        
         Args:
-        weight (numpy array) : 가중치 - 1차원으로 풀어진 상태
+            weight (numpy|cupy array): 가중치 - 1차원으로 풀어서 반환
             shape (list): 가중치의 원본 shape
             lenght (list): 가중치의 원본 shape의 길이
         Returns:
             (list) : 가중치 원본 shape으로 복원
         """
-
-    def _decode(self, weight, shape, lenght):
         weights = []
         start = 0
         for i in range(len(shape)):
@@ -142,6 +147,17 @@ class Optimizer:
         return weights
 
     def f(self, x, y, weights):
+        """
+        EBPSO의 목적함수 (예상)
+
+        Args:
+            x (list): 입력 데이터
+            y (list): 출력 데이터
+            weights (list): 가중치
+
+        Returns:
+            (float): 목적 함수 값
+        """
         self.model.set_weights(weights)
         self.model.compile(loss=self.loss, optimizer="sgd", metrics=["accuracy"])
         score = self.model.evaluate(x, y, verbose=0)[1]
@@ -150,18 +166,6 @@ class Optimizer:
         else:
             return 1 + np.abs(score)
 
-    """
-    Args:
-        x_test : numpy.ndarray,
-        y_test : numpy.ndarray,
-        epochs : int,
-        save : bool - True : save, False : not save
-        save_path : str ex) "./result",
-        renewal : str ex) "acc" or "loss",
-        empirical_balance : bool - True : 
-        Dispersion : bool - True : g_best 의 값을 분산시켜 전역해를 찾음, False : g_best 의 값만 사용
-        check_point : int - 저장할 위치 - None : 저장 안함
-    """
 
     def fit(
         self,
@@ -175,6 +179,18 @@ class Optimizer:
         Dispersion: bool = False,
         check_point: int = None,
     ):
+        """
+        Args:
+            x_test : numpy.ndarray,
+            y_test : numpy.ndarray,
+            epochs : int,
+            save : bool - True : save, False : not save
+            save_path : str ex) "./result",
+            renewal : str ex) "acc" or "loss",
+            empirical_balance : bool - True : 
+            Dispersion : bool - True : g_best 의 값을 분산시켜 전역해를 찾음, False : g_best 의 값만 사용
+            check_point : int - 저장할 위치 - None : 저장 안함
+        """
         self.save_path = save_path
         self.empirical_balance = empirical_balance
         self.Dispersion = Dispersion
@@ -235,8 +251,10 @@ class Optimizer:
         print(f"initial g_best_score : {self.g_best_score}")
 
         try:
-            for _ in range(epochs):
+            epochs_pbar = tqdm(range(epochs), desc=f"best {self.renewal} : {self.g_best_score:.4f}", ascii=True, leave=True)
-                print(f"epoch {_ + 1}/{epochs}")
+            for _ in epochs_pbar:
+                epochs_pbar.set_description(f"best {self.renewal} : {self.g_best_score:.4f}")
+
                 acc = 0
                 loss = 0
                 min_score = np.inf
@@ -250,8 +268,9 @@ class Optimizer:
                 g_ = (1 - decrement) * g_ + decrement * ts
                 self.g_best_ = self._decode(g_, g_sh, g_len)
 
-                # for i in tqdm(range(len(self.particles)), desc=f"epoch {_ + 1}/{epochs}", ascii=True):
+                part_pbar = tqdm(range(len(self.particles)), desc=f"acc : {max_score:.4f} loss : {min_loss:.4f}", ascii=True, leave=False)
-                for i in range(len(self.particles)):
+                for i in part_pbar:
+                    part_pbar.set_description(f"acc : {max_score:.4f} loss : {min_loss:.4f}")
                     w = self.w_max - (self.w_max - self.w_min) * _ / epochs
 
                     if Dispersion:
@@ -333,10 +352,9 @@ class Optimizer:
 
                 # print(f"loss min : {min_loss} | loss max : {max_loss} | acc min : {min_score} | acc max : {max_score}")
                 # print(f"loss avg : {loss/self.n_particles} | acc avg : {acc/self.n_particles} | Best {renewal} : {self.g_best_score}")
-                print(
+                # print(
-                    f"loss min : {round(min_loss, 4)} | acc max : {round(max_score, 4)} | Best {renewal} : {self.g_best_score}"
+                    # f"loss min : {round(min_loss, 4)} | acc max : {round(max_score, 4)} | Best {renewal} : {self.g_best_score}"
-                )
+                # )
-
 
                 if check_point is not None:
                     if _ % check_point == 0:
@@ -361,18 +379,42 @@ class Optimizer:
             return self.g_best_score
 
     def get_best_model(self):
+        """
+        최고 점수를 받은 모델을 반환
+
+        Returns:
+            (keras.models): 모델
+        """ 
         model = keras.models.model_from_json(self.model.to_json())
         model.set_weights(self.g_best)
         model.compile(loss=self.loss, optimizer="sgd", metrics=["accuracy"])
         return model
 
     def get_best_score(self):
+        """
+        최고 점수를 반환
+
+        Returns:
+            (float): 점수
+        """
         return self.g_best_score
 
     def get_best_weights(self):
+        """
+        최고 점수를 받은 가중치를 반환
+
+        Returns:
+            (float): 가중치
+        """
         return self.g_best
 
     def save_info(self, path: str = "./result"):
+        """
+        학습 정보를 저장
+
+        Args:
+            path (str, optional): 저장 위치. Defaults to "./result".
+        """
         json_save = {
             "name": f"{self.day}_{self.n_particles}_{self.c0}_{self.c1}_{self.w_min}.h5",
             "n_particles": self.n_particles,
@@ -395,10 +437,25 @@ class Optimizer:
             json.dump(json_save, f, indent=4)
 
     def _check_point_save(self, save_path: str = f"./result/check_point"):
+        """
+        중간 저장
+
+        Args:
+            save_path (str, optional): checkpoint 저장 위치 및 이름. Defaults to f"./result/check_point".
+        """
         model = self.get_best_model()
         model.save_weights(save_path)
 
     def model_save(self, save_path: str = "./result"):
+        """
+        최고 점수를 받은 모델 저장
+
+        Args:
+            save_path (str, optional): 모델의 저장 위치. Defaults to "./result".
+
+        Returns:
+            (keras.models): 모델
+        """
         model = self.get_best_model()
         model.save(
             f"./{save_path}/{self.day}/{self.n_particles}_{self.c0}_{self.c1}_{self.w_min}.h5"

pso/particle.py

@@ -7,7 +7,16 @@ import gc
 
 
 class Particle:
-    def __init__(self, model: keras.models, loss, negative: bool = False):
+    """
+    Particle Swarm Optimization의 Particle을 구현한 클래스
+    """
+    def __init__(self, model: keras.models, loss, negative: bool = False|True):
+        """
+        Args:
+            model (keras.models): 학습 및 검증을 위한 모델
+            loss (str|): 손실 함수
+            negative (bool, optional): 음의 가중치 사용 여부 - 전역 탐색 용도(조기 수렴 방지). Defaults to False.
+        """
         self.model = model
         self.loss = loss
         init_weights = self.model.get_weights()
@@ -31,14 +40,17 @@ class Particle:
         del self.best_weights
         gc.collect()
 
+    def _encode(self, weights: list):
         """
+        가중치를 1차원으로 풀어서 반환
+        
+        Args:
+            weights (list) : keras model의 가중치
         Returns:
-        (cupy array) : 가중치 - 1차원으로 풀어서 반환
+            (numpy array) : 가중치 - 1차원으로 풀어서 반환
             (list) : 가중치의 원본 shape
             (list) : 가중치의 원본 shape의 길이
         """
-
-    def _encode(self, weights: list):
         # w_gpu = cp.array([])
         w_gpu = np.array([])
         lenght = []
@@ -52,12 +64,19 @@ class Particle:
 
         return w_gpu, shape, lenght
 
+
+    def _decode(self, weight: list, shape, lenght):
         """
+        _encode 로 인코딩된 가중치를 원본 shape으로 복원
+        파라미터는 encode의 리턴값을 그대로 사용을 권장
+        
+        Args:
+            weight (numpy|cupy array): 가중치 - 1차원으로 풀어서 반환
+            shape (list): 가중치의 원본 shape
+            lenght (list): 가중치의 원본 shape의 길이
         Returns:
             (list) : 가중치 원본 shape으로 복원
         """
-
-    def _decode(self, weight: list, shape, lenght):
         weights = []
         start = 0
         for i in range(len(shape)):
@@ -75,6 +94,17 @@ class Particle:
         return weights
 
     def get_score(self, x, y, renewal: str = "acc"):
+        """
+        모델의 성능을 평가하여 점수를 반환
+
+        Args:
+            x (list): 입력 데이터
+            y (list): 출력 데이터
+            renewal (str, optional): 점수 갱신 방식. Defaults to "acc" | "acc" or "loss".
+
+        Returns:
+            _type_: _description_
+        """
         self.model.compile(loss=self.loss, optimizer="sgd", metrics=["accuracy"])
         score = self.model.evaluate(x, y, verbose=0)
         # print(score)
@@ -90,6 +120,15 @@ class Particle:
         return score
 
     def _update_velocity(self, local_rate, global_rate, w, g_best):
+        """
+        현재 속도 업데이트
+
+        Args:
+            local_rate (flost): 지역 최적해의 영향력
+            global_rate (float): 전역 최적해의 영향력
+            w (float): 현재 속도의 영향력 - 관성 | 0.9 ~ 0.4 이 적당
+            g_best (list): 전역 최적해
+        """
         encode_w, w_sh, w_len = self._encode(weights=self.model.get_weights())
         encode_v, v_sh, v_len = self._encode(weights=self.velocities)
         encode_p, p_sh, p_len = self._encode(weights=self.best_weights)
@@ -116,6 +155,18 @@ class Particle:
         del r0, r1
 
     def _update_velocity_w(self, local_rate, global_rate, w, w_p, w_g, g_best):
+        """
+        현재 속도 업데이트
+        기본 업데이트의 변형으로 지역 최적해와 전역 최적해를 분산시켜 조기 수렴을 방지
+
+        Args:
+            local_rate (float): 지역 최적해의 영향력
+            global_rate (float): 전역 최적해의 영향력
+            w (float): 현재 속도의 영향력 - 관성 | 0.9 ~ 0.4 이 적당
+            w_p (float): 지역 최적해의 분산 정도
+            w_g (float): 전역 최적해의 분산 정도
+            g_best (list):  전역 최적해
+        """
         encode_w, w_sh, w_len = self._encode(weights=self.model.get_weights())
         encode_v, v_sh, v_len = self._encode(weights=self.velocities)
         encode_p, p_sh, p_len = self._encode(weights=self.best_weights)
@@ -142,6 +193,9 @@ class Particle:
         del r0, r1
 
     def _update_weights(self):
+        """
+        가중치 업데이트
+        """
         encode_w, w_sh, w_len = self._encode(weights=self.model.get_weights())
         encode_v, v_sh, v_len = self._encode(weights=self.velocities)
         new_w = encode_w + encode_v
@@ -150,6 +204,17 @@ class Particle:
         del encode_v, v_sh, v_len
 
     def f(self, x, y, weights):
+        """
+        EBPSO의 목적함수(예상)
+
+        Args:
+            x (list): 입력 데이터
+            y (list): 출력 데이터
+            weights (list): 가중치
+
+        Returns:
+            flost: 목적함수 값
+        """
         self.model.set_weights(weights)
         score = self.model.evaluate(x, y, verbose=0)[1]
         if score > 0:
@@ -158,6 +223,21 @@ class Particle:
             return 1 + np.abs(score)
 
     def step(self, x, y, local_rate, global_rate, w, g_best, renewal: str = "acc"):
+        """
+        파티클의 한 스텝을 진행합니다.
+
+        Args:
+            x (list): 입력 데이터
+            y (list): 출력 데이터
+            local_rate (float): 지역최적해의 영향력
+            global_rate (float): 전역최적해의 영향력
+            w (float): 관성
+            g_best (list): 전역최적해
+            renewal (str, optional): 최고점수 갱신 방식. Defaults to "acc" | "acc" or "loss"
+
+        Returns:
+            list: 현재 파티클의 점수
+        """
         self._update_velocity(local_rate, global_rate, w, g_best)
         self._update_weights()
         return self.get_score(x, y, renewal)
@@ -165,12 +245,42 @@ class Particle:
     def step_w(
         self, x, y, local_rate, global_rate, w, g_best, w_p, w_g, renewal: str = "acc"
     ):
+        """
+        파티클의 한 스텝을 진행합니다.
+        기본 스텝의 변형으로, 지역최적해와 전역최적해의 분산 정도를 조정할 수 있습니다
+
+        Args:
+            x (list): 입력 데이터
+            y (list): 출력 데이터
+            local_rate (float): 지역 최적해의 영향력
+            global_rate (float): 전역 최적해의 영향력
+            w (float): 관성
+            g_best (list): 전역 최적해
+            w_p (float): 지역 최적해의 분산 정도
+            w_g (float): 전역 최적해의 분산 정도
+            renewal (str, optional): 최고점수 갱신 방식. Defaults to "acc" | "acc" or "loss"
+
+        Returns:
+            float: 현재 파티클의 점수
+        """
         self._update_velocity_w(local_rate, global_rate, w, w_p, w_g, g_best)
         self._update_weights()
         return self.get_score(x, y, renewal)
 
     def get_best_score(self):
+        """
+        파티클의 최고점수를 반환합니다.
+
+        Returns:
+            float: 최고점수
+        """
         return self.best_score
 
     def get_best_weights(self):
+        """
+        파티클의 최고점수를 받은 가중치를 반환합니다
+
+        Returns:
+            list: 가중치 리스트
+        """
         return self.best_weights

test.ipynb

@@ -9,7 +9,7 @@
     "import numpy as np\n",
     "import pandas as pd\n",
     "\n",
-    "import matplotlib.pyplot as plt\n"
+    "import matplotlib.pyplot as plt"
    ]
   },
   {

xor.py

@@ -5,11 +5,13 @@ os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
 import tensorflow as tf
 tf.random.set_seed(777)  # for reproducibility
 
+import numpy as np
+np.random.seed(777)
+
 # from pso_tf import PSO
 from pso import Optimizer
 from tensorflow import keras
 
-import numpy as np
 
 from tensorflow import keras
 from tensorflow.keras.models import Sequential
@@ -38,20 +40,11 @@ def make_model():
 # %%
 model = make_model()
 x_test, y_test = get_data()
-# loss = 'binary_crossentropy'
+
-# loss = 'categorical_crossentropy'
+loss = ['mean_squared_error', 'mean_squared_logarithmic_error', 'binary_crossentropy', 'categorical_crossentropy', 'sparse_categorical_crossentropy', 'kullback_leibler_divergence', 'poisson', 'cosine_similarity', 'log_cosh', 'huber_loss', 'mean_absolute_error', 'mean_absolute_percentage_error']
-# loss = 'sparse_categorical_crossentropy'
-# loss = 'kullback_leibler_divergence'
-# loss = 'poisson'
-# loss = 'cosine_similarity'
-# loss = 'log_cosh'
-# loss = 'huber_loss' 
-# loss = 'mean_absolute_error'
-# loss = 'mean_absolute_percentage_error'
-loss = 'mean_squared_error'
 
 pso_xor = Optimizer(model,
-    loss=loss, n_particles=75, c0=0.35, c1=0.8, w_min=0.6, w_max=1.2, negative_swarm=0.25)
+    loss=loss[0], n_particles=75, c0=0.35, c1=0.8, w_min=0.6, w_max=1.2, negative_swarm=0.25)
 best_score = pso_xor.fit(
     x_test, y_test, epochs=200, save=True, save_path="./result/xor", renewal="acc", empirical_balance=False, Dispersion=False, check_point=25)