diff --git a/models/bam.py b/models/bam.py index 9263a86..a10fc74 100644 --- a/models/bam.py +++ b/models/bam.py @@ -7,28 +7,28 @@ # BAM 반복 안정화 함수 (soft sign 기반) # ============================================ -def bam_associative_recall(x, y, w_xy, w_yx, steps=5, temp=1.0): +def bam_associative_recall(input_pattern, output_pattern, weight_x_to_y, weight_y_to_x, steps=5, temperature=1.0): """ BAM 양방향 연상 반복 Args: - x: 입력 패턴 (batch, input_dim) - y: 출력 패턴 초기값 (batch, output_dim) - w_xy: X->Y 가중치 (input_dim, output_dim) - w_yx: Y->X 가중치 (output_dim, input_dim) + input_pattern: 입력 패턴 (batch, input_dim) + output_pattern: 출력 패턴 초기값 (batch, output_dim) + weight_x_to_y: X->Y 가중치 (input_dim, output_dim) + weight_y_to_x: Y->X 가중치 (output_dim, input_dim) steps: 반복 횟수 - temp: temperature parameter + temperature: temperature parameter Returns: 안정화된 출력 패턴 y """ for _ in range(steps): # X -> Y 방향 - y = soft_sign(tf.matmul(x, w_xy), temp=temp) + output_pattern = soft_sign(tf.matmul(input_pattern, weight_x_to_y), temp=temperature) # Y -> X 방향 (안정화) - x = soft_sign(tf.matmul(y, w_yx), temp=temp) + input_pattern = soft_sign(tf.matmul(output_pattern, weight_y_to_x), temp=temperature) - return y + return output_pattern # ============================================ @@ -45,44 +45,44 @@ class BAMAssociativeLayer(layers.Layer): - Temperature 조절 가능 """ - def __init__(self, input_dim, output_dim, steps=5, temp=1.0, name=None): + def __init__(self, input_dim, output_dim, steps=5, temperature=1.0, name=None): super(BAMAssociativeLayer, self).__init__(name=name) self.input_dim = input_dim self.output_dim = output_dim self.steps = steps - self.temp = temp + self.temperature = temperature def build(self, input_shape): # X -> Y 가중치 - self.Wxy = self.add_weight( + self.weight_x_to_y = self.add_weight( shape=(self.input_dim, self.output_dim), initializer='glorot_uniform', trainable=True, - name='Wxy' + name='weight_x_to_y' ) # Y -> X 가중치 (feedback) - self.Wyx = self.add_weight( + self.weight_y_to_x = self.add_weight( shape=(self.output_dim, self.input_dim), initializer='glorot_uniform', trainable=True, - name='Wyx' + name='weight_y_to_x' ) super().build(input_shape) - def call(self, x, training=None): + def call(self, inputs, training=None): # 초기 출력 패턴 (0으로 시작) - y0 = tf.zeros((tf.shape(x)[0], self.output_dim)) + initial_output_pattern = tf.zeros((tf.shape(inputs)[0], self.output_dim)) # 반복 연상으로 안정화 - y = bam_associative_recall( - x, y0, self.Wxy, self.Wyx, + output_pattern = bam_associative_recall( + inputs, initial_output_pattern, self.weight_x_to_y, self.weight_y_to_x, steps=self.steps, - temp=self.temp + temperature=self.temperature ) - return y + return output_pattern def get_config(self): config = super().get_config() @@ -90,7 +90,7 @@ def get_config(self): 'input_dim': self.input_dim, 'output_dim': self.output_dim, 'steps': self.steps, - 'temp': self.temp + 'temperature': self.temperature }) return config @@ -125,52 +125,52 @@ def __init__(self, input_dim, output_dim, delta=0.2): name='W' ) - def call(self, x): + def call(self, inputs): """정방향 연상: 입력 x에 대한 출력 패턴 y 반환""" - y_lin = tf.matmul(x, self.W) + linear_output = tf.matmul(inputs, self.W) # Soft sign 사용 (더 안정적) - y = soft_sign(y_lin, temp=1.0) - return y + output_pattern = soft_sign(linear_output, temp=1.0) + return output_pattern - def train_hebbian(self, X, Y, epochs=1, eta=0.008): + def train_hebbian(self, input_data, target_output, epochs=1, learning_rate=0.008): """ Hebbian 학습 규칙에 따른 가중치 훈련 Args: - X: 학습 입력 배열 - Y: 대응 학습 출력(정답) 배열 + input_data: 학습 입력 배열 + target_output: 대응 학습 출력(정답) 배열 epochs: 전체 패턴 반복 학습 횟수 - eta: 학습률 + learning_rate: 학습률 """ - X = tf.convert_to_tensor(X, dtype=tf.float32) - Y = tf.convert_to_tensor(Y, dtype=tf.float32) - n = tf.shape(X)[0] + input_data = tf.convert_to_tensor(input_data, dtype=tf.float32) + target_output = tf.convert_to_tensor(target_output, dtype=tf.float32) + num_samples = tf.shape(input_data)[0] - for ep in range(epochs): + for epoch in range(epochs): # 패턴 순서를 섞어 학습 - indices = tf.random.shuffle(tf.range(n)) - for i in indices: - x0 = tf.expand_dims(X[i], 0) # (1, input_dim) - y0 = tf.expand_dims(Y[i], 0) # (1, output_dim) + shuffled_indices = tf.random.shuffle(tf.range(num_samples)) + for sample_idx in shuffled_indices: + current_input = tf.expand_dims(input_data[sample_idx], 0) # (1, input_dim) + current_target = tf.expand_dims(target_output[sample_idx], 0) # (1, output_dim) # 정방향으로 출력 패턴 계산 - y_pred = self(x0) + predicted_output = self(current_input) # 역방향으로 입력 패턴 회상 - x_pred = soft_sign( - tf.matmul(y0, self.W, transpose_b=True), + recalled_input = soft_sign( + tf.matmul(current_target, self.W, transpose_b=True), temp=1.0 ) # Hebbian 가중치 변화 계산 - dW = tf.matmul( - (x0 + x_pred), - (y0 - y_pred), + weight_delta = tf.matmul( + (current_input + recalled_input), + (current_target - predicted_output), transpose_a=True ) # 가중치 업데이트 - self.W.assign_add(eta * dW) + self.W.assign_add(learning_rate * weight_delta) # ============================================ diff --git a/models/bam_mf.py b/models/bam_mf.py index 7651055..7d457f1 100644 --- a/models/bam_mf.py +++ b/models/bam_mf.py @@ -6,25 +6,25 @@ # 포화성 큐빅 활성화 함수 # ============================================ -def cubic_activation(x): +def cubic_activation(input_tensor): """ 단순 큐빅 활성화: x - x^3/3 (안정화 버전) 입력을 clipping하여 수치 안정성 보장 """ # ✅ 입력 clipping으로 수치 안정성 확보 (NaN 방지) - x_clipped = tf.clip_by_value(x, -10.0, 10.0) - return x_clipped - tf.pow(x_clipped, 3) / 3.0 + clipped_input = tf.clip_by_value(input_tensor, -10.0, 10.0) + return clipped_input - tf.pow(clipped_input, 3) / 3.0 # Soft Sign 함수 (BAM 안정화용) -def soft_sign(x, temp=1.0): +def soft_sign(input_tensor, temp=1.0): """ 소프트 사인 함수 (tanh 기반) temp: temperature parameter (낮을수록 sharp) """ - return tf.tanh(x / temp) + return tf.tanh(input_tensor / temp) # HebbianDenseLayer (W, V 분리, BatchNorm 사용) @@ -47,27 +47,27 @@ def __init__(self, units, activation=None, use_feedback=True, name=None): self.use_feedback = use_feedback def build(self, input_shape): - d_in = input_shape[-1] + input_dimension = input_shape[-1] # Feedforward 가중치 W - self.W = self.add_weight( - shape=(d_in, self.units), + self.feedforward_weights = self.add_weight( + shape=(input_dimension, self.units), initializer='glorot_uniform', # ✅ glorot이 cubic activation에 더 안정적 trainable=True, - name='W' + name='feedforward_weights' ) # Feedback 가중치 V (선택적, 현재 미사용) # if self.use_feedback: - # self.V = self.add_weight( - # shape=(self.units, d_in), + # self.feedback_weights = self.add_weight( + # shape=(self.units, input_dimension), # initializer='he_uniform', # trainable=True, - # name='V' + # name='feedback_weights' # ) # ✅ BatchNormalization 안정화 설정 - self.norm = layers.BatchNormalization( + self.normalization_layer = layers.BatchNormalization( momentum=0.99, # 더 안정적인 이동 평균 epsilon=1e-3, # 수치 안정성 center=True, @@ -78,14 +78,14 @@ def build(self, input_shape): def call(self, inputs, training=None): # Feedforward: x -> W -> z - z = tf.matmul(inputs, self.W) - z = self.norm(z, training=training) + layer_output = tf.matmul(inputs, self.feedforward_weights) + layer_output = self.normalization_layer(layer_output, training=training) # Activation if self.activation is not None: - z = self.activation(z) + layer_output = self.activation(layer_output) - return z + return layer_output def get_config(self): config = super().get_config() @@ -154,21 +154,21 @@ def __init__(self, input_dim, encoder_dims=[1024, 768, 512, 256]): ) self.decoder_layers.append(layer) - def encode(self, x, training=None): + def encode(self, inputs, training=None): """입력 x를 잠재 표현 z로 인코딩""" - z = x + encoded = inputs for layer in self.encoder_layers: - z = layer(z, training=training) - return z + encoded = layer(encoded, training=training) + return encoded - def decode(self, z, training=None): + def decode(self, latent_representation, training=None): """잠재 표현 z로부터 입력 형태로 복원""" - out = z + decoded = latent_representation for layer in self.decoder_layers: - out = layer(out, training=training) - return out + decoded = layer(decoded, training=training) + return decoded - def call(self, x, training=None): + def call(self, inputs, training=None): """ MF 모듈을 통해 입력을 복원하고 잠재 표현도 반환 @@ -176,19 +176,19 @@ def call(self, x, training=None): (복원 이미지, 잠재 표현) """ # ✅ 입력 검증 및 clipping (NaN 방지) - x = tf.clip_by_value(x, 0.0, 1.0) + inputs = tf.clip_by_value(inputs, 0.0, 1.0) - z = self.encode(x, training=training) + latent_representation = self.encode(inputs, training=training) # ✅ 잠재 표현 안정화 (gradient explosion 방지) - z = tf.clip_by_value(z, -100.0, 100.0) + latent_representation = tf.clip_by_value(latent_representation, -100.0, 100.0) - out = self.decode(z, training=training) + restored_output = self.decode(latent_representation, training=training) # ✅ 출력 안정화 (sigmoid 후에도 보장) - out = tf.clip_by_value(out, 0.0, 1.0) + restored_output = tf.clip_by_value(restored_output, 0.0, 1.0) - return out, z + return restored_output, latent_representation def get_config(self): return { @@ -201,51 +201,52 @@ def get_config(self): # 편의 함수: MF 모듈 빌드 # ============================================ -def build_mf_module(input_vec, hidden_units=[1024, 768, 512, 256]): +def build_mf_module(input_vector, hidden_units=[1024, 768, 512, 256]): """ 함수형 API로 MF 모듈 빌드 Args: - input_vec: 입력 텐서 + input_vector: 입력 텐서 hidden_units: 은닉층 차원 리스트 Returns: 잠재 표현 텐서 """ - x = input_vec - for i, units in enumerate(hidden_units): - x = HebbianDenseLayer( + current_layer = input_vector + for layer_idx, units in enumerate(hidden_units): + current_layer = HebbianDenseLayer( units, activation=cubic_activation, - name=f"mf_layer_{i+1}" - )(x) - return x + name=f"mf_layer_{layer_idx+1}" + )(current_layer) + return current_layer -def build_decoder(x, decoder_units=[512, 768, 1024], original_dim=3072): +def build_decoder(latent_representation, decoder_units=[512, 768, 1024], original_dimension=3072): """ 함수형 API로 디코더 빌드 Args: - x: 잠재 표현 텐서 + latent_representation: 잠재 표현 텐서 decoder_units: 디코더 은닉층 차원 리스트 - original_dim: 원본 입력 차원 + original_dimension: 원본 입력 차원 Returns: 복원된 출력 텐서 """ - for i, u in enumerate(decoder_units): - x = HebbianDenseLayer( - u, + current_layer = latent_representation + for layer_idx, units in enumerate(decoder_units): + current_layer = HebbianDenseLayer( + units, activation=cubic_activation, - name=f"decoder_layer_{i+1}" - )(x) + name=f"decoder_layer_{layer_idx+1}" + )(current_layer) return layers.Dense( - original_dim, + original_dimension, activation='sigmoid', name='restoration_output' - )(x) + )(current_layer) # ============================================ diff --git a/models/bam_sequential.py b/models/bam_sequential.py index 66bf969..5bf2fd2 100644 --- a/models/bam_sequential.py +++ b/models/bam_sequential.py @@ -24,7 +24,7 @@ class SequentialBAM: """ def __init__(self, input_dim=3072, denoise_latent=256, cls_latent=None, - num_classes=10, bam_steps=5, bam_temp=0.5): + num_classes=10, bam_steps=5, bam_temperature=0.5): """ Args: input_dim: 입력 차원 (32*32*3 = 3072) @@ -32,17 +32,17 @@ def __init__(self, input_dim=3072, denoise_latent=256, cls_latent=None, cls_latent: 사용 안함 (호환성 유지) num_classes: 분류 클래스 수 bam_steps: BAM 반복 횟수 - bam_temp: BAM temperature + bam_temperature: BAM temperature """ self.input_dim = input_dim self.denoise_latent = denoise_latent self.num_classes = num_classes self.bam_steps = bam_steps - self.bam_temp = bam_temp + self.bam_temperature = bam_temperature # MF 모듈 생성 (복원용) encoder_dims = [1024, 768, 512, denoise_latent] - self.mf = MF(input_dim=input_dim, encoder_dims=encoder_dims) + self.multifactor_module = MF(input_dim=input_dim, encoder_dims=encoder_dims) # Keras 모델로 래핑 self._build_keras_models() @@ -51,10 +51,10 @@ def _build_keras_models(self): """MF와 BAM을 Keras 모델로 래핑""" # Denoising 모델 (MF) noisy_input = layers.Input(shape=(self.input_dim,), name='noisy_input') - restored, _ = self.mf(noisy_input) + restored_output, _ = self.multifactor_module(noisy_input) self.denoise_model = keras.Model( inputs=noisy_input, - outputs=restored, + outputs=restored_output, name='denoise_bam' ) @@ -62,35 +62,35 @@ def _build_keras_models(self): denoised_input = layers.Input(shape=(self.input_dim,), name='denoised_input') # BAMAssociativeLayer 사용 - cls_logits = BAMAssociativeLayer( + classification_logits = BAMAssociativeLayer( input_dim=self.input_dim, output_dim=self.num_classes, steps=self.bam_steps, - temp=self.bam_temp, + temperature=self.bam_temperature, name='bam_assoc' )(denoised_input) - cls_output = layers.Activation('softmax', name='cls_output')(cls_logits) + classification_output = layers.Activation('softmax', name='classification_output')(classification_logits) - self.cls_model = keras.Model( + self.classification_model = keras.Model( inputs=denoised_input, - outputs=cls_output, + outputs=classification_output, name='classification_bam' ) - def compile_models(self, denoise_lr=1e-3, cls_lr=2e-4): + def compile_models(self, denoise_learning_rate=1e-3, classification_learning_rate=2e-4): """ 두 모델을 독립적으로 컴파일 Args: - denoise_lr: MF 학습률 - cls_lr: BAM 학습률 + denoise_learning_rate: MF 학습률 + classification_learning_rate: BAM 학습률 """ # Stage 1: MF 복원 모델 (MAE 손실) # ✅ Gradient clipping으로 안정성 확보 (NaN 방지) self.denoise_model.compile( optimizer=keras.optimizers.Adam( - learning_rate=denoise_lr, + learning_rate=denoise_learning_rate, clipnorm=1.0, # gradient norm clipping clipvalue=0.5 # gradient value clipping ), @@ -102,9 +102,9 @@ def compile_models(self, denoise_lr=1e-3, cls_lr=2e-4): ) # Stage 2: BAM 분류 모델 - self.cls_model.compile( + self.classification_model.compile( optimizer=keras.optimizers.Adam( - learning_rate=cls_lr, + learning_rate=classification_learning_rate, clipnorm=1.0 ), loss='categorical_crossentropy', @@ -116,14 +116,14 @@ def compile_models(self, denoise_lr=1e-3, cls_lr=2e-4): print("✓ Models compiled successfully (with gradient clipping)") - def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, + def train_stage1(self, noisy_images, clean_images, epochs=50, batch_size=128, validation_split=0.2, callbacks=None): """ Stage 1: MF 복원 학습 Args: - x_noisy: 노이즈가 추가된 입력 (flattened) - x_clean: 깨끗한 타겟 (flattened) + noisy_images: 노이즈가 추가된 입력 (flattened) + clean_images: 깨끗한 타겟 (flattened) epochs: 학습 에폭 수 batch_size: 배치 크기 validation_split: 검증 데이터 비율 @@ -141,8 +141,8 @@ def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, print(f"Loss: MAE (Mean Absolute Error)") print(f"Dropout: None (극저 SNR 최적화)") - history1 = self.denoise_model.fit( - x_noisy, x_clean, + history_stage1 = self.denoise_model.fit( + noisy_images, clean_images, epochs=epochs, batch_size=batch_size, validation_split=validation_split, @@ -151,9 +151,9 @@ def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, ) print("\n✓ Stage 1 training complete!") - return history1 + return history_stage1 - def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, + def train_stage2(self, noisy_images, labels, epochs=50, batch_size=128, validation_split=0.2, callbacks=None): """ Stage 2: BAM 분류 학습 @@ -162,8 +162,8 @@ def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, 복원된 이미지를 입력으로 BAM 분류기를 학습 Args: - x_noisy: 노이즈가 추가된 입력 (복원용) - y_labels: 원-핫 인코딩된 레이블 + noisy_images: 노이즈가 추가된 입력 (복원용) + labels: 원-핫 인코딩된 레이블 epochs: 학습 에폭 수 batch_size: 배치 크기 validation_split: 검증 데이터 비율 @@ -182,37 +182,37 @@ def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, print(f"BAM input dim: {self.input_dim}") print(f"BAM output dim: {self.num_classes}") print(f"BAM steps: {self.bam_steps}") - print(f"BAM temperature: {self.bam_temp}") + print(f"BAM temperature: {self.bam_temperature}") # 1. MF로 훈련 데이터 복원 (배치 단위로 처리 - 메모리 효율) print("Generating denoised images for training...") - n_samples = len(x_noisy) - x_restored_list = [] + num_samples = len(noisy_images) + restored_images_list = [] # 배치 단위로 복원 (메모리 절약) - for i in range(0, n_samples, batch_size * 4): # 4배 큰 청크로 - end_idx = min(i + batch_size * 4, n_samples) - x_batch_restored = self.denoise_model.predict( - x_noisy[i:end_idx], + for start_idx in range(0, num_samples, batch_size * 4): # 4배 큰 청크로 + end_idx = min(start_idx + batch_size * 4, num_samples) + batch_restored = self.denoise_model.predict( + noisy_images[start_idx:end_idx], batch_size=batch_size, verbose=0 ) - x_restored_list.append(x_batch_restored) + restored_images_list.append(batch_restored) - if (i // (batch_size * 4)) % 10 == 0: - print(f" Processed {end_idx}/{n_samples} samples...") + if (start_idx // (batch_size * 4)) % 10 == 0: + print(f" Processed {end_idx}/{num_samples} samples...") import numpy as np - x_restored = np.concatenate(x_restored_list, axis=0) - del x_restored_list # 메모리 해제 + restored_images = np.concatenate(restored_images_list, axis=0) + del restored_images_list # 메모리 해제 - print(f"✓ Restored images shape: {x_restored.shape}") + print(f"✓ Restored images shape: {restored_images.shape}") # 2. BAM 학습 (Keras fit 사용) print("\nTraining BAM with softmax + categorical crossentropy") - history2 = self.cls_model.fit( - x_restored, y_labels, + history_stage2 = self.classification_model.fit( + restored_images, labels, epochs=epochs, batch_size=batch_size, validation_split=validation_split, @@ -221,14 +221,14 @@ def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, ) print("\n✓ Stage 2 training complete!") - return history2 + return history_stage2 - def predict(self, x_noisy, batch_size=128, verbose=0): + def predict(self, noisy_images, batch_size=128, verbose=0): """ 연쇄 추론: x_noisy → MF → x_restored → BAM → y_pred Args: - x_noisy: 노이즈가 추가된 입력 + noisy_images: 노이즈가 추가된 입력 batch_size: 배치 크기 verbose: 진행 표시 레벨 @@ -239,62 +239,62 @@ def predict(self, x_noisy, batch_size=128, verbose=0): } """ # Stage 1: MF로 복원 - x_restored = self.denoise_model.predict( - x_noisy, + restored_images = self.denoise_model.predict( + noisy_images, batch_size=batch_size, verbose=verbose ) # Stage 2: BAM으로 분류 - y_pred = self.cls_model.predict( - x_restored, + predictions = self.classification_model.predict( + restored_images, batch_size=batch_size, verbose=verbose ) return { - "denoised": x_restored, - "predictions": y_pred + "denoised": restored_images, + "predictions": predictions } - def evaluate(self, x_noisy, x_clean, y_labels, batch_size=128): + def evaluate(self, noisy_images, clean_images, labels, batch_size=128): """ 전체 파이프라인 평가 Args: - x_noisy: 노이즈가 추가된 입력 - x_clean: 깨끗한 이미지 (복원 평가용) - y_labels: 원-핫 인코딩된 레이블 + noisy_images: 노이즈가 추가된 입력 + clean_images: 깨끗한 이미지 (복원 평가용) + labels: 원-핫 인코딩된 레이블 batch_size: 배치 크기 Returns: dict: 복원 MSE/MAE/PSNR와 분류 정확도 """ - outputs = self.predict(x_noisy, batch_size=batch_size, verbose=1) + outputs = self.predict(noisy_images, batch_size=batch_size, verbose=1) # 복원 성능 - mse = np.mean((outputs['denoised'] - x_clean) ** 2) - mae = np.mean(np.abs(outputs['denoised'] - x_clean)) - psnr = 10 * np.log10(1.0 / mse) if mse > 0 else float('inf') + mean_squared_error = np.mean((outputs['denoised'] - clean_images) ** 2) + mean_absolute_error = np.mean(np.abs(outputs['denoised'] - clean_images)) + psnr = 10 * np.log10(1.0 / mean_squared_error) if mean_squared_error > 0 else float('inf') # 분류 성능 - y_pred_classes = np.argmax(outputs['predictions'], axis=1) - y_true_classes = np.argmax(y_labels, axis=1) - accuracy = np.mean(y_pred_classes == y_true_classes) + predicted_classes = np.argmax(outputs['predictions'], axis=1) + true_classes = np.argmax(labels, axis=1) + accuracy = np.mean(predicted_classes == true_classes) # Top-3 accuracy - top3_preds = np.argsort(outputs['predictions'], axis=1)[:, -3:] - top3_acc = np.mean([y_true in top3 for y_true, top3 in zip(y_true_classes, top3_preds)]) + top3_predictions = np.argsort(outputs['predictions'], axis=1)[:, -3:] + top3_accuracy = np.mean([true_class in top3 for true_class, top3 in zip(true_classes, top3_predictions)]) return { "restoration": { - "mse": float(mse), - "mae": float(mae), + "mse": float(mean_squared_error), + "mae": float(mean_absolute_error), "psnr": float(psnr) }, "classification": { "accuracy": float(accuracy), - "top3_accuracy": float(top3_acc) + "top3_accuracy": float(top3_accuracy) } } @@ -318,26 +318,26 @@ def load_models(self, denoise_path="bam_mf_denoise.keras", # ============================================ def create_sequential_bam(input_dim=3072, denoise_latent=256, num_classes=10, - denoise_lr=1e-3, cls_lr=2e-4, - bam_steps=5, bam_temp=0.5): + denoise_learning_rate=1e-3, classification_learning_rate=2e-4, + bam_steps=5, bam_temperature=0.5): """ Sequential BAM 생성 및 컴파일 (one-liner) Returns: 컴파일된 SequentialBAM 인스턴스 """ - seq_bam = SequentialBAM( + sequential_bam = SequentialBAM( input_dim=input_dim, denoise_latent=denoise_latent, num_classes=num_classes, bam_steps=bam_steps, - bam_temp=bam_temp + bam_temperature=bam_temperature ) - seq_bam.compile_models( - denoise_lr=denoise_lr, - cls_lr=cls_lr + sequential_bam.compile_models( + denoise_learning_rate=denoise_learning_rate, + classification_learning_rate=classification_learning_rate ) - return seq_bam + return sequential_bam # ============================================ diff --git a/models/cae_sequential.py b/models/cae_sequential.py index 1392060..e8b67de 100644 --- a/models/cae_sequential.py +++ b/models/cae_sequential.py @@ -4,7 +4,7 @@ def build_cae_multitask(input_shape_img=(32, 32, 3), num_classes=10, dropout_rate=0.1, - l2_reg=1e-4) -> Model: + l2_regularization=1e-4) -> Model: """ Multitask CAE (초저 SNR 최적화, Skip connection 없음) @@ -20,70 +20,70 @@ def build_cae_multitask(input_shape_img=(32, 32, 3), num_classes=10, input_shape_img: 입력 이미지 shape (32, 32, 3) num_classes: 분류 클래스 수 dropout_rate: Dropout 비율 (초저 SNR: 0.05~0.1) - l2_reg: L2 regularization 강도 + l2_regularization: L2 regularization 강도 """ - img_in = layers.Input(shape=input_shape_img, name='image_input') + image_input = layers.Input(shape=input_shape_img, name='image_input') # ===================================== # Encoder (4 stages) # ===================================== - x = layers.Conv2D(32, 3, padding='same', activation='relu')(img_in) - x = layers.Conv2D(32, 3, padding='same', activation='relu')(x) - p1 = layers.MaxPooling2D(2)(x) # 16x16 + encoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(image_input) + encoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_1 = layers.MaxPooling2D(2)(encoder_layer) # 16x16 - x = layers.Conv2D(64, 3, padding='same', activation='relu')(p1) - x = layers.Conv2D(64, 3, padding='same', activation='relu')(x) - p2 = layers.MaxPooling2D(2)(x) # 8x8 + encoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(pooled_layer_1) + encoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_2 = layers.MaxPooling2D(2)(encoder_layer) # 8x8 - x = layers.Conv2D(128, 3, padding='same', activation='relu')(p2) - x = layers.Conv2D(128, 3, padding='same', activation='relu')(x) - p3 = layers.MaxPooling2D(2)(x) # 4x4 + encoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(pooled_layer_2) + encoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_3 = layers.MaxPooling2D(2)(encoder_layer) # 4x4 - x = layers.Conv2D(256, 3, padding='same', activation='relu')(p3) - x = layers.Conv2D(256, 3, padding='same', activation='relu')(x) - p4 = layers.MaxPooling2D(2)(x) # 2x2 + encoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(pooled_layer_3) + encoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_4 = layers.MaxPooling2D(2)(encoder_layer) # 2x2 # ===================================== # Bottleneck # ===================================== - x = layers.Conv2D(512, 3, padding='same', activation='relu')(p4) - x = layers.Conv2D(512, 3, padding='same', activation='relu')(x) + bottleneck = layers.Conv2D(512, 3, padding='same', activation='relu')(pooled_layer_4) + bottleneck = layers.Conv2D(512, 3, padding='same', activation='relu')(bottleneck) # ===================================== # Classification Head (from bottleneck) # ===================================== - feat = layers.GlobalAveragePooling2D()(x) - feat = layers.Dense(128, activation='relu', - kernel_regularizer=regularizers.l2(l2_reg))(feat) - feat = layers.Dropout(dropout_rate)(feat) # ✅ 0.5 → 0.1 - cls_out = layers.Dense(num_classes, activation='softmax', - name='classification_output')(feat) + features = layers.GlobalAveragePooling2D()(bottleneck) + features = layers.Dense(128, activation='relu', + kernel_regularizer=regularizers.l2(l2_regularization))(features) + features = layers.Dropout(dropout_rate)(features) # ✅ 0.5 → 0.1 + classification_output = layers.Dense(num_classes, activation='softmax', + name='classification_output')(features) # ===================================== # Decoder (4 stages, symmetric) # NO skip connections - 순수 bottleneck 복원 # ===================================== - d = layers.Conv2DTranspose(256, 2, strides=2, padding='same')(x) # 4x4 - d = layers.Conv2D(256, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(256, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(256, 2, strides=2, padding='same')(bottleneck) # 4x4 + decoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(128, 2, strides=2, padding='same')(d) # 8x8 - d = layers.Conv2D(128, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(128, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(128, 2, strides=2, padding='same')(decoder_layer) # 8x8 + decoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(64, 2, strides=2, padding='same')(d) # 16x16 - d = layers.Conv2D(64, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(64, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(64, 2, strides=2, padding='same')(decoder_layer) # 16x16 + decoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(32, 2, strides=2, padding='same')(d) # 32x32 - d = layers.Conv2D(32, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(32, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(32, 2, strides=2, padding='same')(decoder_layer) # 32x32 + decoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(decoder_layer) # ✅ Sigmoid activation (픽셀 값 [0,1] 범위 보장) - rec = layers.Conv2D(3, 1, activation='sigmoid', - name='restoration_output')(d) + restoration_output = layers.Conv2D(3, 1, activation='sigmoid', + name='restoration_output')(decoder_layer) - return Model(inputs=img_in, outputs=[rec, cls_out], name='CAE_multitask') + return Model(inputs=image_input, outputs=[restoration_output, classification_output], name='CAE_multitask') def build_cae_restoration(input_shape_img=(32, 32, 3)) -> Model: @@ -92,49 +92,49 @@ def build_cae_restoration(input_shape_img=(32, 32, 3)) -> Model: Sequential BAM과 비교하기 위한 복원 전용 모델 """ - img_in = layers.Input(shape=input_shape_img) + image_input = layers.Input(shape=input_shape_img) # Encoder (4 stages) - x = layers.Conv2D(32, 3, padding='same', activation='relu')(img_in) - x = layers.Conv2D(32, 3, padding='same', activation='relu')(x) - p1 = layers.MaxPooling2D(2)(x) # 16x16 + encoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(image_input) + encoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_1 = layers.MaxPooling2D(2)(encoder_layer) # 16x16 - x = layers.Conv2D(64, 3, padding='same', activation='relu')(p1) - x = layers.Conv2D(64, 3, padding='same', activation='relu')(x) - p2 = layers.MaxPooling2D(2)(x) # 8x8 + encoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(pooled_layer_1) + encoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_2 = layers.MaxPooling2D(2)(encoder_layer) # 8x8 - x = layers.Conv2D(128, 3, padding='same', activation='relu')(p2) - x = layers.Conv2D(128, 3, padding='same', activation='relu')(x) - p3 = layers.MaxPooling2D(2)(x) # 4x4 + encoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(pooled_layer_2) + encoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_3 = layers.MaxPooling2D(2)(encoder_layer) # 4x4 - x = layers.Conv2D(256, 3, padding='same', activation='relu')(p3) - x = layers.Conv2D(256, 3, padding='same', activation='relu')(x) - p4 = layers.MaxPooling2D(2)(x) # 2x2 + encoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(pooled_layer_3) + encoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(encoder_layer) + pooled_layer_4 = layers.MaxPooling2D(2)(encoder_layer) # 2x2 # Bottleneck - x = layers.Conv2D(512, 3, padding='same', activation='relu')(p4) - x = layers.Conv2D(512, 3, padding='same', activation='relu')(x) + bottleneck = layers.Conv2D(512, 3, padding='same', activation='relu')(pooled_layer_4) + bottleneck = layers.Conv2D(512, 3, padding='same', activation='relu')(bottleneck) # Decoder (4 stages) - d = layers.Conv2DTranspose(256, 2, strides=2, padding='same')(x) # 4x4 - d = layers.Conv2D(256, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(256, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(256, 2, strides=2, padding='same')(bottleneck) # 4x4 + decoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(128, 2, strides=2, padding='same')(d) # 8x8 - d = layers.Conv2D(128, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(128, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(128, 2, strides=2, padding='same')(decoder_layer) # 8x8 + decoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(64, 2, strides=2, padding='same')(d) # 16x16 - d = layers.Conv2D(64, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(64, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(64, 2, strides=2, padding='same')(decoder_layer) # 16x16 + decoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(decoder_layer) - d = layers.Conv2DTranspose(32, 2, strides=2, padding='same')(d) # 32x32 - d = layers.Conv2D(32, 3, padding='same', activation='relu')(d) - d = layers.Conv2D(32, 3, padding='same', activation='relu')(d) + decoder_layer = layers.Conv2DTranspose(32, 2, strides=2, padding='same')(decoder_layer) # 32x32 + decoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(decoder_layer) + decoder_layer = layers.Conv2D(32, 3, padding='same', activation='relu')(decoder_layer) # ✅ Sigmoid activation - rec = layers.Conv2D(3, 1, activation='sigmoid')(d) - return Model(inputs=img_in, outputs=rec, name='CAE_restoration') + restoration_output = layers.Conv2D(3, 1, activation='sigmoid')(decoder_layer) + return Model(inputs=image_input, outputs=restoration_output, name='CAE_restoration') def build_cae_classification(input_shape_img=(32, 32, 3), num_classes=10, @@ -145,28 +145,28 @@ def build_cae_classification(input_shape_img=(32, 32, 3), num_classes=10, Sequential BAM과 비교하기 위한 분류 전용 모델 복원된 이미지를 입력으로 받아 분류 """ - img_in = layers.Input(shape=input_shape_img, name='restored_input') + restored_image_input = layers.Input(shape=input_shape_img, name='restored_input') # Encoder (간단한 버전 - 이미 복원된 이미지 입력) - x = layers.Conv2D(64, 3, padding='same', activation='relu')(img_in) - x = layers.MaxPooling2D(2)(x) # 16x16 + encoder_layer = layers.Conv2D(64, 3, padding='same', activation='relu')(restored_image_input) + encoder_layer = layers.MaxPooling2D(2)(encoder_layer) # 16x16 - x = layers.Conv2D(128, 3, padding='same', activation='relu')(x) - x = layers.MaxPooling2D(2)(x) # 8x8 + encoder_layer = layers.Conv2D(128, 3, padding='same', activation='relu')(encoder_layer) + encoder_layer = layers.MaxPooling2D(2)(encoder_layer) # 8x8 - x = layers.Conv2D(256, 3, padding='same', activation='relu')(x) - x = layers.MaxPooling2D(2)(x) # 4x4 + encoder_layer = layers.Conv2D(256, 3, padding='same', activation='relu')(encoder_layer) + encoder_layer = layers.MaxPooling2D(2)(encoder_layer) # 4x4 - x = layers.Conv2D(512, 3, padding='same', activation='relu')(x) - x = layers.GlobalAveragePooling2D()(x) + encoder_layer = layers.Conv2D(512, 3, padding='same', activation='relu')(encoder_layer) + encoder_layer = layers.GlobalAveragePooling2D()(encoder_layer) # Classifier - x = layers.Dense(128, activation='relu', - kernel_regularizer=regularizers.l2(1e-4))(x) - x = layers.Dropout(dropout_rate)(x) - cls_out = layers.Dense(num_classes, activation='softmax')(x) + features = layers.Dense(128, activation='relu', + kernel_regularizer=regularizers.l2(1e-4))(encoder_layer) + features = layers.Dropout(dropout_rate)(features) + classification_output = layers.Dense(num_classes, activation='softmax')(features) - return Model(inputs=img_in, outputs=cls_out, name='CAE_classification') + return Model(inputs=restored_image_input, outputs=classification_output, name='CAE_classification') class SequentialCAE: @@ -179,21 +179,21 @@ def __init__(self, input_shape=(32, 32, 3), num_classes=10): self.restore_model = build_cae_restoration(input_shape) self.cls_model = build_cae_classification(input_shape, num_classes) - def compile_models(self, restore_lr=1e-3, cls_lr=1e-3, restore_loss='mse'): + def compile_models(self, restoration_learning_rate=1e-3, classification_learning_rate=1e-3, restoration_loss='mse'): """두 모델을 독립적으로 컴파일""" self.restore_model.compile( - optimizer=tf.keras.optimizers.Adam(learning_rate=restore_lr), - loss=restore_loss, # 'mse' or 'mae' + optimizer=tf.keras.optimizers.Adam(learning_rate=restoration_learning_rate), + loss=restoration_loss, # 'mse' or 'mae' metrics=['mse', 'mae'] ) self.cls_model.compile( - optimizer=tf.keras.optimizers.Adam(learning_rate=cls_lr), + optimizer=tf.keras.optimizers.Adam(learning_rate=classification_learning_rate), loss='categorical_crossentropy', metrics=['accuracy'] ) - def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, + def train_stage1(self, noisy_images, clean_images, epochs=50, batch_size=128, validation_split=0.1, callbacks=None): """Stage 1: 복원 학습""" print("\n" + "="*60) @@ -201,7 +201,7 @@ def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, print("="*60) history1 = self.restore_model.fit( - x_noisy, x_clean, + noisy_images, clean_images, epochs=epochs, batch_size=batch_size, validation_split=validation_split, @@ -210,7 +210,7 @@ def train_stage1(self, x_noisy, x_clean, epochs=50, batch_size=128, ) return history1 - def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, + def train_stage2(self, noisy_images, labels, epochs=50, batch_size=128, validation_split=0.1, callbacks=None): """Stage 2: 분류 학습""" # 메모리 정리 @@ -225,34 +225,34 @@ def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, print("Generating restored images for training...") import numpy as np - n_samples = len(x_noisy) - x_restored_list = [] + num_samples = len(noisy_images) + restored_images_list = [] # 큰 청크로 나눠서 처리 (4배 배치 크기) chunk_size = batch_size * 4 - for i in range(0, n_samples, chunk_size): - end_idx = min(i + chunk_size, n_samples) - x_batch_restored = self.restore_model.predict( - x_noisy[i:end_idx], + for start_idx in range(0, num_samples, chunk_size): + end_idx = min(start_idx + chunk_size, num_samples) + batch_restored = self.restore_model.predict( + noisy_images[start_idx:end_idx], batch_size=batch_size, verbose=0 ) - x_restored_list.append(x_batch_restored) + restored_images_list.append(batch_restored) # 진행 상황 출력 - if (i // chunk_size) % 10 == 0: - print(f" Processed {end_idx}/{n_samples} samples...") + if (start_idx // chunk_size) % 10 == 0: + print(f" Processed {end_idx}/{num_samples} samples...") - x_restored = np.concatenate(x_restored_list, axis=0) - del x_restored_list # 메모리 해제 + restored_images = np.concatenate(restored_images_list, axis=0) + del restored_images_list # 메모리 해제 gc.collect() - print(f"✓ Restored images shape: {x_restored.shape}") + print(f"✓ Restored images shape: {restored_images.shape}") # 분류 학습 print("\nTraining classification model...") history2 = self.cls_model.fit( - x_restored, y_labels, + restored_images, labels, epochs=epochs, batch_size=batch_size, validation_split=validation_split, @@ -261,54 +261,54 @@ def train_stage2(self, x_noisy, y_labels, epochs=50, batch_size=128, ) return history2 - def predict(self, x_noisy, batch_size=128): + def predict(self, noisy_images, batch_size=128): """연쇄 예측""" - x_restored = self.restore_model.predict(x_noisy, batch_size=batch_size, verbose=0) - y_pred = self.cls_model.predict(x_restored, batch_size=batch_size, verbose=0) + restored_images = self.restore_model.predict(noisy_images, batch_size=batch_size, verbose=0) + predictions = self.cls_model.predict(restored_images, batch_size=batch_size, verbose=0) return { - "restored": x_restored, - "predictions": y_pred + "restored": restored_images, + "predictions": predictions } - def evaluate(self, x_noisy, x_clean, y_labels, batch_size=128): + def evaluate(self, noisy_images, clean_images, labels, batch_size=128): """ 전체 파이프라인 평가 Args: - x_noisy: 노이즈가 추가된 입력 - x_clean: 깨끗한 이미지 (복원 평가용) - y_labels: 원-핫 인코딩된 레이블 + noisy_images: 노이즈가 추가된 입력 + clean_images: 깨끗한 이미지 (복원 평가용) + labels: 원-핫 인코딩된 레이블 batch_size: 배치 크기 Returns: dict: 복원 MSE/MAE/PSNR와 분류 정확도 """ - outputs = self.predict(x_noisy, batch_size=batch_size) + outputs = self.predict(noisy_images, batch_size=batch_size) # 복원 성능 import numpy as np - mse = np.mean((outputs['restored'] - x_clean) ** 2) - mae = np.mean(np.abs(outputs['restored'] - x_clean)) - psnr = 10 * np.log10(1.0 / mse) if mse > 0 else float('inf') + mean_squared_error = np.mean((outputs['restored'] - clean_images) ** 2) + mean_absolute_error = np.mean(np.abs(outputs['restored'] - clean_images)) + psnr = 10 * np.log10(1.0 / mean_squared_error) if mean_squared_error > 0 else float('inf') # 분류 성능 - y_pred_classes = np.argmax(outputs['predictions'], axis=1) - y_true_classes = np.argmax(y_labels, axis=1) - accuracy = np.mean(y_pred_classes == y_true_classes) + predicted_classes = np.argmax(outputs['predictions'], axis=1) + true_classes = np.argmax(labels, axis=1) + accuracy = np.mean(predicted_classes == true_classes) # Top-3 accuracy - top3_preds = np.argsort(outputs['predictions'], axis=1)[:, -3:] - top3_acc = np.mean([y_true in top3 for y_true, top3 in zip(y_true_classes, top3_preds)]) + top3_predictions = np.argsort(outputs['predictions'], axis=1)[:, -3:] + top3_accuracy = np.mean([true_class in top3 for true_class, top3 in zip(true_classes, top3_predictions)]) return { "restoration": { - "mse": float(mse), - "mae": float(mae), + "mse": float(mean_squared_error), + "mae": float(mean_absolute_error), "psnr": float(psnr) }, "classification": { "accuracy": float(accuracy), - "top3_accuracy": float(top3_acc) + "top3_accuracy": float(top3_accuracy) } } \ No newline at end of file diff --git a/models/dncnn.py b/models/dncnn.py index 337869f..0e70e42 100644 --- a/models/dncnn.py +++ b/models/dncnn.py @@ -2,96 +2,96 @@ import tensorflow as tf -def conv_block_dncnn(x, filters, use_bn=True, activation='relu'): +def conv_block_dncnn(input_tensor, num_filters, use_batch_norm=True, activation='relu'): """DnCNN convolution block with optional batch normalization.""" - x = layers.Conv2D(filters, 3, padding='same', use_bias=not use_bn)(x) - if use_bn: - x = layers.BatchNormalization()(x) + output_tensor = layers.Conv2D(num_filters, 3, padding='same', use_bias=not use_batch_norm)(input_tensor) + if use_batch_norm: + output_tensor = layers.BatchNormalization()(output_tensor) if activation: - x = layers.Activation(activation)(x) - return x + output_tensor = layers.Activation(activation)(output_tensor) + return output_tensor def build_dncnn_multitask(input_shape_img=(32, 32, 3), num_classes=10, - depth=17, filters=64, use_bn=True) -> Model: + depth=17, num_filters=64, use_batch_norm=True) -> Model: """Modern DnCNN-style multitask model with improved architecture. Args: input_shape_img: Input image shape num_classes: Number of classification classes depth: Number of convolution layers (typically 17 or 20) - filters: Number of filters in each layer - use_bn: Whether to use batch normalization + num_filters: Number of filters in each layer + use_batch_norm: Whether to use batch normalization Returns: Compiled DnCNN multitask model """ - img_in = layers.Input(shape=input_shape_img, name='image_input') + image_input = layers.Input(shape=input_shape_img, name='image_input') # First layer: Conv + ReLU (with bias) - x = layers.Conv2D(filters, 3, padding='same', activation='relu')(img_in) + network_layer = layers.Conv2D(num_filters, 3, padding='same', activation='relu')(image_input) # Hidden layers: Conv + BN + ReLU (no bias) - for i in range(depth - 2): - x = conv_block_dncnn(x, filters, use_bn=use_bn, activation='relu') + for layer_idx in range(depth - 2): + network_layer = conv_block_dncnn(network_layer, num_filters, use_batch_norm=use_batch_norm, activation='relu') # Last layer: Conv (no activation, no bias) - res = layers.Conv2D(3, 3, padding='same', activation='linear', - use_bias=False, name='residual_pred')(x) + residual_prediction = layers.Conv2D(3, 3, padding='same', activation='linear', + use_bias=False, name='residual_pred')(network_layer) # Restoration output: input - predicted noise - rec = layers.Subtract(name='restoration_output')([img_in, res]) + restoration_output = layers.Subtract(name='restoration_output')([image_input, residual_prediction]) # Classification head from bottleneck features - feat = layers.GlobalAveragePooling2D()(x) - feat = layers.Dense(256, activation='relu', - kernel_regularizer=regularizers.l2(1e-4))(feat) - feat = layers.Dropout(0.5)(feat) - feat = layers.Dense(128, activation='relu', - kernel_regularizer=regularizers.l2(1e-4))(feat) - feat = layers.Dropout(0.3)(feat) - cls_out = layers.Dense(num_classes, activation='softmax', - name='classification_output')(feat) + features = layers.GlobalAveragePooling2D()(network_layer) + features = layers.Dense(256, activation='relu', + kernel_regularizer=regularizers.l2(1e-4))(features) + features = layers.Dropout(0.5)(features) + features = layers.Dense(128, activation='relu', + kernel_regularizer=regularizers.l2(1e-4))(features) + features = layers.Dropout(0.3)(features) + classification_output = layers.Dense(num_classes, activation='softmax', + name='classification_output')(features) - return Model(inputs=img_in, outputs=[rec, cls_out], - name=f'DnCNN_multitask_{depth}L_{filters}F') + return Model(inputs=image_input, outputs=[restoration_output, classification_output], + name=f'DnCNN_multitask_{depth}L_{num_filters}F') def build_dncnn_restoration(input_shape_img=(32, 32, 3), depth=17, - filters=64, use_bn=True) -> Model: + num_filters=64, use_batch_norm=True) -> Model: """Modern DnCNN restoration-only model with improved architecture. Args: input_shape_img: Input image shape depth: Number of convolution layers - filters: Number of filters in each layer - use_bn: Whether to use batch normalization + num_filters: Number of filters in each layer + use_batch_norm: Whether to use batch normalization Returns: DnCNN restoration model """ - img_in = layers.Input(shape=input_shape_img, name='image_input') + image_input = layers.Input(shape=input_shape_img, name='image_input') # First layer: Conv + ReLU (with bias) - x = layers.Conv2D(filters, 3, padding='same', activation='relu')(img_in) + network_layer = layers.Conv2D(num_filters, 3, padding='same', activation='relu')(image_input) # Hidden layers: Conv + BN + ReLU (no bias) - for i in range(depth - 2): - x = conv_block_dncnn(x, filters, use_bn=use_bn, activation='relu') + for layer_idx in range(depth - 2): + network_layer = conv_block_dncnn(network_layer, num_filters, use_batch_norm=use_batch_norm, activation='relu') # Last layer: Conv (no activation, no bias) - res = layers.Conv2D(3, 3, padding='same', activation='linear', - use_bias=False)(x) + residual_prediction = layers.Conv2D(3, 3, padding='same', activation='linear', + use_bias=False)(network_layer) # Restoration output: input - predicted noise - rec = layers.Subtract(name='restoration_output')([img_in, res]) + restoration_output = layers.Subtract(name='restoration_output')([image_input, residual_prediction]) - return Model(inputs=img_in, outputs=rec, - name=f'DnCNN_restoration_{depth}L_{filters}F') + return Model(inputs=image_input, outputs=restoration_output, + name=f'DnCNN_restoration_{depth}L_{num_filters}F') def build_dncnn_modern(input_shape_img=(32, 32, 3), num_classes=10, - depth=20, filters=64, use_bn=True, + depth=20, num_filters=64, use_batch_norm=True, residual_learning=True) -> Model: """Modern DnCNN with additional improvements. @@ -99,51 +99,51 @@ def build_dncnn_modern(input_shape_img=(32, 32, 3), num_classes=10, input_shape_img: Input image shape num_classes: Number of classification classes depth: Number of convolution layers - filters: Number of filters in each layer - use_bn: Whether to use batch normalization + num_filters: Number of filters in each layer + use_batch_norm: Whether to use batch normalization residual_learning: Whether to use residual learning (predict noise) Returns: Modern DnCNN model """ - img_in = layers.Input(shape=input_shape_img, name='image_input') + image_input = layers.Input(shape=input_shape_img, name='image_input') # First layer: Conv + ReLU - x = layers.Conv2D(filters, 3, padding='same', activation='relu')(img_in) + network_layer = layers.Conv2D(num_filters, 3, padding='same', activation='relu')(image_input) # Hidden layers with skip connections every 5 layers skip_connections = [] - for i in range(depth - 2): - x = conv_block_dncnn(x, filters, use_bn=use_bn, activation='relu') + for layer_idx in range(depth - 2): + network_layer = conv_block_dncnn(network_layer, num_filters, use_batch_norm=use_batch_norm, activation='relu') # Add skip connection every 5 layers - if (i + 1) % 5 == 0 and i < depth - 3: - skip_connections.append(x) + if (layer_idx + 1) % 5 == 0 and layer_idx < depth - 3: + skip_connections.append(network_layer) # Last layer: Conv (no activation, no bias) if residual_learning: # Predict noise residual - res = layers.Conv2D(3, 3, padding='same', activation='linear', - use_bias=False, name='residual_pred')(x) - rec = layers.Subtract(name='restoration_output')([img_in, res]) + residual_prediction = layers.Conv2D(3, 3, padding='same', activation='linear', + use_bias=False, name='residual_pred')(network_layer) + restoration_output = layers.Subtract(name='restoration_output')([image_input, residual_prediction]) else: # Direct image prediction - rec = layers.Conv2D(3, 3, padding='same', activation='linear', - name='restoration_output')(x) + restoration_output = layers.Conv2D(3, 3, padding='same', activation='linear', + name='restoration_output')(network_layer) # Classification head - feat = layers.GlobalAveragePooling2D()(x) - feat = layers.Dense(512, activation='relu', - kernel_regularizer=regularizers.l2(1e-4))(feat) - feat = layers.Dropout(0.5)(feat) - feat = layers.Dense(256, activation='relu', - kernel_regularizer=regularizers.l2(1e-4))(feat) - feat = layers.Dropout(0.3)(feat) - cls_out = layers.Dense(num_classes, activation='softmax', - name='classification_output')(feat) - - return Model(inputs=img_in, outputs=[rec, cls_out], - name=f'DnCNN_modern_{depth}L_{filters}F') + features = layers.GlobalAveragePooling2D()(network_layer) + features = layers.Dense(512, activation='relu', + kernel_regularizer=regularizers.l2(1e-4))(features) + features = layers.Dropout(0.5)(features) + features = layers.Dense(256, activation='relu', + kernel_regularizer=regularizers.l2(1e-4))(features) + features = layers.Dropout(0.3)(features) + classification_output = layers.Dense(num_classes, activation='softmax', + name='classification_output')(features) + + return Model(inputs=image_input, outputs=[restoration_output, classification_output], + name=f'DnCNN_modern_{depth}L_{num_filters}F') # Backward compatibility aliases diff --git a/utils.py b/utils.py index eabce48..831975a 100644 --- a/utils.py +++ b/utils.py @@ -38,31 +38,31 @@ def setup_gpu_memory(memory_limit_mb=8192): return False -def add_noise_by_snr(clean_data, snr_db, noise_type='gaussian'): +def add_noise_by_snr(clean_data, signal_to_noise_ratio_db, noise_type='gaussian'): """ 특정 SNR로 노이즈 추가 Args: clean_data: 깨끗한 데이터 (N, H, W, C) 또는 (N, D) - snr_db: 목표 SNR (dB) + signal_to_noise_ratio_db: 목표 SNR (dB) noise_type: 'gaussian', 'sp', 'burst' Returns: 노이즈가 추가된 데이터 """ signal_power = np.mean(clean_data ** 2) - snr_linear = 10 ** (snr_db / 10) + snr_linear = 10 ** (signal_to_noise_ratio_db / 10) noise_power = signal_power / snr_linear - noise_std = np.sqrt(noise_power) + noise_standard_deviation = np.sqrt(noise_power) if noise_type == 'gaussian': - noise = np.random.normal(0, noise_std, clean_data.shape).astype('float32') + noise = np.random.normal(0, noise_standard_deviation, clean_data.shape).astype('float32') noisy_data = clean_data + noise elif noise_type == 'sp': # Salt & Pepper noise noisy_data = clean_data.copy() - noise_ratio = min(0.5, noise_std * 2) + noise_ratio = min(0.5, noise_standard_deviation * 2) # Salt (1.0) salt_mask = np.random.random(clean_data.shape) < noise_ratio / 2 @@ -75,7 +75,7 @@ def add_noise_by_snr(clean_data, snr_db, noise_type='gaussian'): elif noise_type == 'burst': burst_types = ['dead_pixels', 'column_row', 'block'] burst_type = np.random.choice(burst_types) - noisy_data = add_burst_noise(clean_data, noise_std, burst_type) + noisy_data = add_burst_noise(clean_data, noise_standard_deviation, burst_type) else: raise ValueError(f"Unknown noise type: {noise_type}") @@ -83,60 +83,60 @@ def add_noise_by_snr(clean_data, snr_db, noise_type='gaussian'): return np.clip(noisy_data, 0.0, 1.0).astype('float32') -def add_burst_noise(clean_data, noise_std, burst_type='dead_pixels'): +def add_burst_noise(clean_data, noise_standard_deviation, burst_type='dead_pixels'): """ Burst 노이즈 추가 Args: clean_data: 깨끗한 데이터 (N, H, W, C) - noise_std: 노이즈 강도 + noise_standard_deviation: 노이즈 강도 burst_type: 'dead_pixels', 'column_row', 'block' """ noisy_data = clean_data.copy() - N, H, W, C = clean_data.shape + num_samples, height, width, channels = clean_data.shape - affected_ratio = min(0.3, noise_std * 3) + affected_ratio = min(0.3, noise_standard_deviation * 3) if burst_type == 'dead_pixels': - for i in range(N): - num_dead = int(H * W * affected_ratio) - dead_positions = np.random.choice(H * W, num_dead, replace=False) + num_dead_pixels = int(height * width * affected_ratio) + for sample_idx in range(num_samples): + dead_positions = np.random.choice(height * width, num_dead_pixels, replace=False) - for pos in dead_positions: - row = pos // W - col = pos % W - noisy_data[i, row, col, :] = np.random.choice([0.0, 1.0]) + for position in dead_positions: + row = position // width + col = position % width + noisy_data[sample_idx, row, col, :] = np.random.choice([0.0, 1.0]) elif burst_type == 'column_row': - for i in range(N): - num_lines = max(1, int(max(H, W) * affected_ratio / 10)) + num_lines = max(1, int(max(height, width) * affected_ratio / 10)) + for sample_idx in range(num_samples): for _ in range(num_lines): if np.random.random() < 0.5: - col = np.random.randint(0, W) - noisy_data[i, :, col, :] = np.random.choice([0.0, 1.0]) + col = np.random.randint(0, width) + noisy_data[sample_idx, :, col, :] = np.random.choice([0.0, 1.0]) else: - row = np.random.randint(0, H) - noisy_data[i, row, :, :] = np.random.choice([0.0, 1.0]) + row = np.random.randint(0, height) + noisy_data[sample_idx, row, :, :] = np.random.choice([0.0, 1.0]) elif burst_type == 'block': - for i in range(N): + for sample_idx in range(num_samples): num_blocks = max(1, int(10 * affected_ratio)) for _ in range(num_blocks): - block_h = np.random.randint(2, max(3, H // 4)) - block_w = np.random.randint(2, max(3, W // 4)) + block_height = np.random.randint(2, max(3, height // 4)) + block_width = np.random.randint(2, max(3, width // 4)) - start_h = np.random.randint(0, H - block_h + 1) - start_w = np.random.randint(0, W - block_w + 1) + start_height = np.random.randint(0, height - block_height + 1) + start_width = np.random.randint(0, width - block_width + 1) block_value = np.random.choice([0.0, 1.0]) - noisy_data[i, start_h:start_h+block_h, start_w:start_w+block_w, :] = block_value + noisy_data[sample_idx, start_height:start_height+block_height, start_width:start_width+block_width, :] = block_value return noisy_data -def get_callbacks(model_name, monitor='val_loss', patience=30, initial_lr=1e-3, epochs=200): +def get_callbacks(model_name, monitor='val_loss', patience=30, initial_learning_rate=1e-3, epochs=200): """ 학습 콜백 생성 @@ -144,11 +144,11 @@ def get_callbacks(model_name, monitor='val_loss', patience=30, initial_lr=1e-3, model_name: 모델 이름 monitor: 모니터링할 메트릭 patience: Early stopping patience - initial_lr: 초기 학습률 + initial_learning_rate: 초기 학습률 epochs: 전체 에폭 수 """ - final_lr = initial_lr * 0.1 - decay_rate = (final_lr / initial_lr) ** (1 / epochs) + final_learning_rate = initial_learning_rate * 0.1 + decay_rate = (final_learning_rate / initial_learning_rate) ** (1 / epochs) callbacks = [] @@ -171,11 +171,11 @@ def get_callbacks(model_name, monitor='val_loss', patience=30, initial_lr=1e-3, callbacks.append(checkpoint) # 3. Learning Rate Scheduler (Exponential Decay) - def lr_schedule(epoch, lr): - return initial_lr * (decay_rate ** epoch) + def learning_rate_schedule(epoch, current_learning_rate): + return initial_learning_rate * (decay_rate ** epoch) - lr_scheduler = keras.callbacks.LearningRateScheduler(lr_schedule, verbose=0) - callbacks.append(lr_scheduler) + learning_rate_scheduler = keras.callbacks.LearningRateScheduler(learning_rate_schedule, verbose=0) + callbacks.append(learning_rate_scheduler) # 4. CSV Logger csv_logger = keras.callbacks.CSVLogger( @@ -268,7 +268,7 @@ def create_directories(): print("✓ Directories created:", ", ".join(directories)) -def print_training_config(epochs, batch_size, validation_split, initial_lr): +def print_training_config(epochs, batch_size, validation_split, initial_learning_rate): """ 학습 설정 출력 """ @@ -276,24 +276,24 @@ def print_training_config(epochs, batch_size, validation_split, initial_lr): print(f" Epochs: {epochs}") print(f" Batch size: {batch_size}") print(f" Validation split: {validation_split}") - print(f" Initial LR: {initial_lr}") - print(f" Final LR: {initial_lr * 0.1} (10%)") + print(f" Initial LR: {initial_learning_rate}") + print(f" Final LR: {initial_learning_rate * 0.1} (10%)") -def calculate_psnr(img1, img2): +def calculate_psnr(image1, image2): """ PSNR 계산 Args: - img1: 첫 번째 이미지 - img2: 두 번째 이미지 + image1: 첫 번째 이미지 + image2: 두 번째 이미지 Returns: PSNR 값 (dB) """ - mse = np.mean((img1 - img2) ** 2) - if mse == 0: + mean_squared_error = np.mean((image1 - image2) ** 2) + if mean_squared_error == 0: return float('inf') - max_pixel = 1.0 - psnr = 20 * np.log10(max_pixel / np.sqrt(mse)) + max_pixel_value = 1.0 + psnr = 20 * np.log10(max_pixel_value / np.sqrt(mean_squared_error)) return psnr