# Copyright 2026 Yokohama National University. All Rights Reserved. # This is a program used in 4.training_and_prediction.py that contains the model configuration. # import public module ## random import random ## np import numpy as np ## os import os ## csv import csv ### layers from keras.layers import Input from keras.layers import Conv2D from keras.layers import MaxPooling2D from keras.layers import BatchNormalization from keras.layers import Flatten from keras.layers import Dense from keras.layers import Dropout ### models from keras.models import Model ## tensorflow module import tensorflow as tf ### optimizers from tensorflow.keras import optimizers ### L2 norm from tensorflow.keras.regularizers import l2 class CNN(): def __init__(self,filter1,kernel1_size,filter2,kernel2_size,filter3,kernel3_size,dense_unit,dropout_rate,x_train,y_train,x_test,y_test,x_val,y_val,x_train_org,y_train_org,x_test_org,y_test_org,x_val_org,y_val_org,test_date,time,label,average,model_ver): """ CNN model built using Keras Parameters ---------- filter1-3: int Number of filters in the 1st–3rd convolutional layers kernel1-3_size: tuple (x,y) Kernel size of the 1st–3rd convolutional layers dense_unit: int Number of units in the 4th fully connected layer dropout_rate: float Dropout rate of the dropout layer x_train, x_train_org: array Feature array of the training data y_train, y_train_org: array Label array of the training data x_val, x_val_org: array Feature array of the validation data y_val, y_val_org: array Label array of the validation data x_test, x_test_org: array Feature array of the test data y_test, y_test_org: array Label array of the test data test_date: str Target dates for prediction (0807, 0809, 0826, 0829, 0905, 0910) time: str Experiment start time average: list Mean value of the data to be predicted model_ver: str Model name """ # Experiment start time self.time = time # Model name self.model_ver = model_ver # Mean value of the data to be predicted self.average = average # Predicted labels self.label = label # Model name self.model_name = "CNN" # Target for estimation self.date = test_date # Configuration of training, validation, and test data self.x_train = x_train self.y_train = y_train self.x_val = x_val self.y_val = y_val self.x_test = x_test self.y_test = y_test # Keep the original data self.x_train_org = x_train_org self.y_train_org = y_train_org self.x_val_org = x_val_org self.y_val_org = y_val_org self.x_test_org = x_test_org self.y_test_org = y_test_org # Size of input features self.input_shape = (32,96,1) # Number and size of convolutional kernels self.filter1 = filter1 self.kernel1_size = kernel1_size self.filter2 = filter2 self.kernel2_size = kernel2_size self.filter3 = filter3 self.kernel3_size = kernel3_size # Kernel size of pooling layers self.pool_size = (4,1) self.pool_strides = (2,1) # Number of units in fully connected layers self.dense_unit = dense_unit # Dropout rate self.dropout_rate = dropout_rate # Optimization method self.learning_rate = 1.0 self.momentum = 0 # Setting the name of the optimization method self.optimizer_name = "Adadelta" # Configure the actual optimizer according to the specified optimizer name if self.optimizer_name == "SGD": self.optimizer = optimizers.SGD(learning_rate=self.learning_rate,momentum=self.momentum,name="SGD") elif self.optimizer_name == "Adam": self.optimizer = optimizers.Adam(learning_rate=self.learning_rate) self.momentum = None elif self.optimizer_name == "Adadelta": self.optimizer = optimizers.Adadelta(learning_rate=self.learning_rate) self.momentum = None elif self.optimizer_name == "Adagrad": self.optimizer = optimizers.Adagrad(learning_rate=self.learning_rate) self.momentum = None else: print("Error: This optimzer is not defined.") self.optimzer = None # Setting the loss function name self.loss_name = "mae" self.loss = "mae" self.metrics = "mae" self.metrics_name = "mae" # Training parameters self.batch_size = 12 self.epochs = 100 self.epoch_patient = 10 # Whether to use early stopping self.early_stopping_switch = True # Parameters self.parameter_dict = {"examine_time":self.time, "filter1":self.filter1, "filter2":self.filter2, "filter3":self.filter3, "kernel1_size":self.kernel1_size, "kernel2_size":self.kernel2_size, "kernel3_size":self.kernel3_size, "pool_size":self.pool_size, "pool_stride":self.pool_strides, "dense_unit":self.dense_unit, "dropout_rate":self.dropout_rate, "optimizer":self.optimizer_name, "learning_rate":self.learning_rate, "momentum":self.momentum, "loss":self.loss_name, "batch_size":self.batch_size, "epoch":self.epochs, "early_stopping":self.early_stopping_switch, "epoch_patient":self.epoch_patient } def random_sampling(self,data_date = None): """ Sample data for each mean value level based on the number of training data samples. # If the number of samples exceeds the reference, perform undersampling. # If the number of samples is below the reference, perform oversampling. # Oversampling is done by repeatedly performing sampling without replacement to ensure that all original data are used at least once. """ if data_date == None: # Store the average value of the target label in ave. ave = self.average[self.label] print(ave) tmp_x = [] tmp_y = [] over_ave_x = [] over_ave_y = [] under_ave_x = [] under_ave_y = [] add_data_x = [] add_data_y = [] over_count = 0 under_count = 0 # Classify samples as above the average (over_ave) or below the average (under_ave), and count the number of samples in each group. for i in range(len(self.y_train_org)): if self.y_train_org[i] >= ave: over_ave_x.append(self.x_train_org[i]) over_ave_y.append(self.y_train_org[i]) over_count += 1 else: under_ave_x.append(self.x_train_org[i]) under_ave_y.append(self.y_train_org[i]) under_count += 1 print(self.date) print("Number of samples above the average:",over_count) print("Number of samples below the average:",under_count) under_ave_x = np.array(under_ave_x).reshape(-1,32,96,1) under_ave_y = np.array(under_ave_y).reshape(-1,) over_ave_x = np.array(over_ave_x).reshape(-1,32,96,1) over_ave_y = np.array(over_ave_y).reshape(-1,) if over_count >= under_count: # When there are more samples above the average: # Select the same number of below-average samples as the number of above-average samples. # Combine over_ave and under_ave. for i in range(over_count): num_o = i % over_count num_u = i % under_count add_data_x.append(under_ave_x[num_u]) add_data_x.append(over_ave_x[num_o]) add_data_y.append(under_ave_y[num_u]) add_data_y.append(over_ave_y[num_o]) else: # When there are more samples below the average: # Select the same number of above-average samples as the number of below-average samples. # Combine over_ave and under_ave. for i in range(under_count): num_o = i % over_count num_u = i % under_count add_data_x.append(under_ave_x[num_u]) add_data_x.append(over_ave_x[num_o]) add_data_y.append(under_ave_y[num_u]) add_data_y.append(over_ave_y[num_o]) add_data_x = np.array(add_data_x).reshape(-1,32,96,1) add_data_y = np.array(add_data_y).reshape(-1,) # Reorder the data created above and create a list of indices. index = list(range(len(add_data_y))) # Shuffle the list of indices. index_shuffle = random.sample(index,len(index)) for i in index_shuffle: tmp_x.append(add_data_x[i]) tmp_y.append(add_data_y[i]) self.x_train = np.array(tmp_x).reshape(-1,32,96,1) self.y_train = np.array(tmp_y).reshape(-1,) #------------------------------------------------------------------------------------------------ # Create a directory to save randomly sampled training data if it does not exist os.makedirs(f"TrainData/{self.model_ver}/{self.time}/random_sampling/",exist_ok=True) # Save randomly sampled training feature data np.save(f"TrainData/{self.model_ver}/{self.time}/random_sampling/{self.date}_x_{self.label}",self.x_train) # Save randomly sampled training label data np.save(f"TrainData/{self.model_ver}/{self.time}/random_sampling/{self.date}_y_{self.label}",self.y_train) # Notify that random sampling and shuffling have been completed print("Shuffled data by random sampling...") else: try: # Attempt to load previously saved randomly sampled training data print("Load start....") print(f"TrainData/{self.model_ver}/{data_date}/random_sampling/{self.date}_x.npy") # Load training feature data self.x_train = np.load(f"TrainData/{self.model_ver}/{data_date}/random_sampling/{self.date}_x_{self.label}.npy",allow_pickle=True) print("x ok") # Load training label data self.y_train = np.load(f"TrainData/{self.model_ver}/{data_date}/random_sampling/{self.date}_y_{self.label}.npy",allow_pickle=True) print("y ok") # Log successful loading of the data print("load {} data ---- save date={}".format(self.date,data_date)) except: # If loading fails, perform random sampling again self.random_sampling() def create_model(self): """ Function to build the CNN model """ # Input layer inputs = Input(shape=self.input_shape,name="inputs") # Convolutional and pooling layers (1st layer) conv1 = Conv2D(filters=self.filter1, kernel_size=self.kernel1_size, activation="relu", kernel_regularizer=l2(0.001), name="conv1")(inputs) pool1 = MaxPooling2D(pool_size=self.pool_size, strides=self.pool_strides, name="pool1")(conv1) norm1 = BatchNormalization(name="norm1")(pool1) drop1 = Dropout(0.1)(norm1) # prevent overfitting # Convolutional and pooling layers (2nd layer) conv2 = Conv2D(filters=self.filter2, kernel_size=self.kernel2_size, activation="relu", kernel_regularizer=l2(0.001), name="conv2")(drop1) pool2 = MaxPooling2D(pool_size=self.pool_size, strides=self.pool_strides, name="pool2")(conv2) norm2 = BatchNormalization(name="norm2")(pool2) drop2 = Dropout(0.2)(norm2) # Convolutional layer (3rd layer) conv3 = Conv2D(filters=self.filter3, kernel_size=self.kernel3_size, activation="relu", kernel_regularizer=l2(0.001), name="conv3")(drop2) norm3 = BatchNormalization(name="norm3")(conv3) drop3 = Dropout(0.25)(norm3) # Fully connected layer flat = Flatten(name="flat")(drop3) dense1 = Dense(units=self.dense_unit, activation="relu", kernel_regularizer=l2(0.001), name="dense1")(flat) norm4 = BatchNormalization(name="relu")(dense1) # Dropout layer drop4 = Dropout(rate = self.dropout_rate)(norm4) # Output layer dense2 = Dense(1,activation="sigmoid", kernel_regularizer=l2(0.001), name="out")(drop4) # Build CNN model self.cnn = Model(inputs = inputs, outputs = dense2) self.cnn.compile(optimizer=self.optimizer,loss = self.loss,metrics=self.metrics) def predict(self): """ Function to train the CNN model and perform prediction using the trained CNN model """ # Create directory for saving training logs (CSVLogger) os.makedirs(f"CSVLogger/{self.model_ver}/{self.model_name}/{self.label}/{self.time}/",exist_ok=True) # CSV logger to record training history (loss, metrics) for each epoch csv_logger = tf.keras.callbacks.CSVLogger(f"CSVLogger/{self.model_ver}/{self.model_name}/{self.label}/{self.time}/{self.date}_({self.filter1}_{self.filter2}_{self.filter3}_{self.dense_unit}.csv",) # Model checkpoint callback to save the model at each epoch model_checkpoint_callback = tf.keras.callbacks.ModelCheckpoint( filepath=f'modeldata/{self.model_ver}/label={self.label}/{self.time}/{self.date}_epoch='+'{epoch:02d}.hdf5', save_weights_only=False, monitor='mean_out_loss', verbose=0, save_best_only=False, save_freq='epoch') # === Training phase === if self.early_stopping_switch: # Early stopping callback to prevent overfitting early_stopping=tf.keras.callbacks.EarlyStopping(monitor="val_loss", min_delta=0.001, patience=self.epoch_patient, verbose=0, mode="min", restore_best_weights=True) # Train the CNN model with early stopping self.fit_result = self.cnn.fit(x = self.x_train,y = self.y_train, batch_size=self.batch_size, epochs=self.epochs, verbose=1, shuffle=False, validation_data=(self.x_val,self.y_val), callbacks=[early_stopping,csv_logger,model_checkpoint_callback]) # Display EarlyStopping configuration print(f"[EarlyStopping] monitor = 'val_loss', min_delta = 0.001, patience = {self.epoch_patient}, mode = 'min'") else: # Train the CNN model without early stopping self.fit_result = self.cnn.fit(x = self.x_train,y = self.y_train, batch_size=self.batch_size, epochs=self.epochs, verbose=1, shuffle=False, validation_data=(self.x_val,self.y_val), callbacks=[csv_logger]) # === Prediction phase === # Predict unified functional spectrum values for the test data self.pred = self.cnn.predict(self.x_test) # Evaluate the trained model on the test data self.eval = self.cnn.evaluate(self.x_test, self.y_test) def output_csv(self): """ Output evaluation results to a CSV file """ # Create directory for evaluation results if it does not exist os.makedirs(f"eval/{self.model_ver}/{self.model_name}/{self.label}/",exist_ok=True) # Create an empty CSV file if it does not already exist if not os.path.isfile(f"eval/{self.model_ver}/{self.model_name}/{self.label}/{self.date}.csv"): with open(f"eval/{self.model_ver}/{self.model_name}/{self.label}/{self.date}.csv","w",newline="") as f: pass # Append evaluation results to the CSV file with open(f"eval/{self.model_ver}/{self.model_name}/{self.label}/{self.date}.csv","a",newline="") as f: writer = csv.writer(f) # Write header row (evaluation metrics) writer.writerow(["time",self.loss_name,self.metrics_name,"additonal_information"]) # Write evaluation results writer.writerow([self.time]+self.eval) def execute(self): """ Execute the full pipeline: model creation, training & prediction, and result output """ # Build the CNN model self.create_model() # Train the model and perform prediction self.predict() # Output evaluation results to CSV self.output_csv()