Inception Time 中组合Inception网络
第一段代码包含inception网络；第二段代码包含集成一组 Inception 网络的代码。请问如何将自己的数据应用于NNE,有train和test两个数据集。class Classifier_INCEPTION:
def __init__(self, output_directory, input_shape, nb_classes, verbose=True, build=True, batch_size=64,
nb_filters=32, use_residual=True, use_bottleneck=True, depth=6, kernel_size=41, nb_epochs=15):
self.output_directory = output_directory
self.nb_filters = nb_filters
self.use_residual = use_residual
self.use_bottleneck = use_bottleneck
self.depth = depth
self.kernel_size = kernel_size - 1
self.callbacks = None
self.batch_size = batch_size
self.bottleneck_size = 32
self.nb_epochs = nb_epochs
if build == True:
self.model = self.build_model(input_shape, nb_classes)
if (verbose == True):
self.model.summary()
self.verbose = verbose
self.model.save_weights(self.output_directory + 'model_init.hdf5')
def _inception_module(self, input_tensor, stride=1, activation='linear'):
if self.use_bottleneck and int(input_tensor.shape[-1]) > 1:
input_inception = keras.layers.Conv1D(filters=self.bottleneck_size, kernel_size=1,
padding='same', activation=activation, use_bias=False)(input_tensor)
else:
input_inception = input_tensor
# kernel_size_s =
kernel_size_s =
conv_list = []
for i in range(len(kernel_size_s)):
conv_list.append(keras.layers.Conv1D(filters=self.nb_filters, kernel_size=kernel_size_s,
strides=stride, padding='same', activation=activation, use_bias=False)(
input_inception))
max_pool_1 = keras.layers.MaxPool1D(pool_size=3, strides=stride, padding='same')(input_tensor)
conv_6 = keras.layers.Conv1D(filters=self.nb_filters, kernel_size=1,
padding='same', activation=activation, use_bias=False)(max_pool_1)
conv_list.append(conv_6)
x = keras.layers.Concatenate(axis=2)(conv_list)
x = keras.layers.BatchNormalization()(x)
x = keras.layers.Activation(activation='relu')(x)
return x
def _shortcut_layer(self, input_tensor, out_tensor):
shortcut_y = keras.layers.Conv1D(filters=int(out_tensor.shape[-1]), kernel_size=1,
padding='same', use_bias=False)(input_tensor)
shortcut_y = keras.layers.BatchNormalization()(shortcut_y)
x = keras.layers.Add()()
x = keras.layers.Activation('relu')(x)
return x
def build_model(self, input_shape, nb_classes):
input_layer = keras.layers.Input(input_shape)
x = input_layer
input_res = input_layer
for d in range(self.depth):
x = self._inception_module(x)
if self.use_residual and d % 3 == 2:
x = self._shortcut_layer(input_res, x)
input_res = x
gap_layer = keras.layers.GlobalAveragePooling1D()(x)
output_layer = keras.layers.Dense(nb_classes, activation='softmax')(gap_layer)
model = keras.models.Model(inputs=input_layer, outputs=output_layer)
model.compile(loss='categorical_crossentropy', optimizer=keras.optimizers.Adam(),
metrics=['acc'])
reduce_lr = keras.callbacks.ReduceLROnPlateau(monitor='loss', factor=0.5, patience=50,
min_lr=0.0001)
file_path = self.output_directory + 'best_model.hdf5'
model_checkpoint = keras.callbacks.ModelCheckpoint(filepath=file_path, monitor='loss',
save_best_only=True)
self.callbacks =
return model
def fit(self, x_train, y_train, x_val, y_val, y_true, plot_test_acc=True):
if self.batch_size is None:
mini_batch_size = int(min(x_train.shape / 10, 16))
else:
mini_batch_size = self.batch_size
start_time = time.time()
if plot_test_acc:
hist = self.model.fit(x_train, y_train, batch_size=mini_batch_size, epochs=self.nb_epochs,
verbose=self.verbose, validation_data=(x_val, y_val), callbacks=self.callbacks)
else:
hist = self.model.fit(x_train, y_train, batch_size=mini_batch_size, epochs=self.nb_epochs,
verbose=self.verbose, callbacks=self.callbacks)
duration = time.time() - start_time
self.model.save(self.output_directory + 'last_model.hdf5')
y_pred = self.predict(x_val, y_true, x_train, y_train, y_val,
return_df_metrics=False)
# save predictions
np.save(self.output_directory + 'y_pred.npy', y_pred)
# convert the predicted from binary to integer
y_pred = np.argmax(y_pred, axis=1)
df_metrics = save_logs(self.output_directory, hist, y_pred, y_true, duration,
plot_test_acc=plot_test_acc)
keras.backend.clear_session()
return df_metrics
def predict(self, x_test, y_true, x_train, y_train, y_test, return_df_metrics=True):
start_time = time.time()
model_path = self.output_directory + 'best_model.hdf5'
model = keras.models.load_model(model_path)
y_pred = model.predict(x_test, batch_size=self.batch_size)
if return_df_metrics:
y_pred = np.argmax(y_pred, axis=1)
df_metrics = calculate_metrics(y_true, y_pred, 0.0)
return df_metrics
else:
test_duration = time.time() - start_time
save_test_duration(self.output_directory + 'test_duration.csv', test_duration)
return y_pred
df1 = pd.read_csv("train.csv")
df1 = np.array(df1)
X = np.expand_dims(df1[:, 1:891].astype(float), axis=2)# 对数据进行增维并转化为32为
#X = np.expand_dims(df1[:, 1:891].astype(float), axis=1)
Y = df1[:, 0]
X_train, X_val, y_train, y_val = train_test_split(X, Y, test_size=0.2, random_state=42)
y_train, y_val = transform_labels(y_train, y_val)
# save orignal y because later we will use binary
y_true = y_val.astype(np.int64)
y_true_train = y_train.astype(np.int64)
# transform the labels from integers to one hot vectors
enc = sklearn.preprocessing.OneHotEncoder()
enc.fit(np.concatenate((y_train, y_val), axis=0).reshape(-1, 1))
y_train = enc.transform(y_train.reshape(-1, 1)).toarray()
y_val = enc.transform(y_val.reshape(-1, 1)).toarray()
'''y_train = to_categorical(y_train)# one-hot encoding
y_val= to_categorical(y_val)
y_true = y_val'''
df2 = pd.read_csv("test.csv")
df2 = np.array(df1)
X_test = np.expand_dims(df2[:, 1:891].astype(float), axis=2)
y_test=df2[:, 0]
y_true1 = y_test.astype(np.int64)
y_test = enc.transform(y_test.reshape(-1, 1)).toarray()
# 创建模型
model = Classifier_INCEPTION(output_directory='output_directory', input_shape=X_train.shape, nb_classes=8)
# 训练模型
model.fit(X_train, y_train, X_val, y_val,y_true)
df_metrics=model.predict(X_test, y_true1, X_train, y_train, y_test)
print(df_metrics)
import keras
import numpy as np
from utils.utils import calculate_metrics
from utils.utils import create_directory
from utils.utils import check_if_file_exits
import gc
from utils.constants import UNIVARIATE_ARCHIVE_NAMESas ARCHIVE_NAMES
import time
class Classifier_NNE:
def create_classifier(self, model_name, input_shape, nb_classes, output_directory, verbose=False,
build=True):
if self.check_if_match('inception*', model_name):
from classifiers import inception
return inception.Classifier_INCEPTION(output_directory, input_shape, nb_classes, verbose,
build=build)
def check_if_match(self, rex, name2):
import re
pattern = re.compile(rex)
return pattern.match(name2)
def __init__(self, output_directory, input_shape, nb_classes, verbose=False, nb_iterations=5,
clf_name='inception'):
self.classifiers =
out_add = ''
for cc in self.classifiers:
out_add = out_add + cc + '-'
self.archive_name = ARCHIVE_NAMES
self.iterations_to_take =
for cc in self.iterations_to_take:
out_add = out_add + str(cc) + '-'
self.output_directory = output_directory.replace('nne',
'nne' + '/' + out_add)
create_directory(self.output_directory)
self.dataset_name = output_directory.split('/')[-2]
self.verbose = verbose
self.models_dir = output_directory.replace('nne', 'classifier')
def fit(self, x_train, y_train, x_test, y_test, y_true):
# no training since models are pre-trained
start_time = time.time()
y_pred = np.zeros(shape=y_test.shape)
ll = 0
# loop through all classifiers
for model_name in self.classifiers:
# loop through different initialization of classifiers
for itr in self.iterations_to_take:
if itr == 0:
itr_str = ''
else:
itr_str = '_itr_' + str(itr)
curr_archive_name = self.archive_name + itr_str
curr_dir = self.models_dir.replace('classifier', model_name).replace(
self.archive_name, curr_archive_name)
model = self.create_classifier(model_name, None, None,
curr_dir, build=False)
predictions_file_name = curr_dir + 'y_pred.npy'
# check if predictions already made
if check_if_file_exits(predictions_file_name):
# then load only the predictions from the file
curr_y_pred = np.load(predictions_file_name)
else:
# then compute the predictions
curr_y_pred = model.predict(x_test, y_true, x_train, y_train, y_test,
return_df_metrics=False)
keras.backend.clear_session()
np.save(predictions_file_name, curr_y_pred)
y_pred = y_pred + curr_y_pred
ll += 1
# average predictions
y_pred = y_pred / ll
# save predictions
np.save(self.output_directory + 'y_pred.npy', y_pred)
# convert the predicted from binary to integer
y_pred = np.argmax(y_pred, axis=1)
duration = time.time() - start_time
df_metrics = calculate_metrics(y_true, y_pred, duration)
df_metrics.to_csv(self.output_directory + 'df_metrics.csv', index=False)
gc.collect() 本帖最后由 陶远航 于 2023-10-13 21:37 编辑
这段代码中，首先定义了一个名为Classifier_INCEPTION的类，该类实现了Inception网络的构建和训练。然后，通过读取训练集和测试集的数据，将数据应用到Inception网络中进行训练和预测。
以下是如何将自己的数据应用于这个网络的步骤：
1. 数据准备：首先，你需要准备你的数据。在这个例子中，数据是从CSV文件中读取的。你需要将你的数据格式化为一个可以被网络接受的形式。在这个例子中，输入数据被转换为一个三维的numpy数组，其中第一维是样本数，第二维是特征数，第三维是1（因为这是一个单变量的时间序列分类任务）。标签被转换为one-hot编码的形式。
df1 = pd.read_csv("train.csv")
df1 = np.array(df1)
X = np.expand_dims(df1[:, 1:891].astype(float), axis=2)# 对数据进行增维并转化为32为
Y = df1[:, 0]
X_train, X_val, y_train, y_val = train_test_split(X, Y, test_size=0.2, random_state=42)
y_train, y_val = transform_labels(y_train, y_val)
# save orignal y because later we will use binary
y_true = y_val.astype(np.int64)
y_true_train = y_train.astype(np.int64)
# transform the labels from integers to one hot vectors
enc = sklearn.preprocessing.OneHotEncoder()
enc.fit(np.concatenate((y_train, y_val), axis=0).reshape(-1, 1))
y_train = enc.transform(y_train.reshape(-1, 1)).toarray()
y_val = enc.transform(y_val.reshape(-1, 1)).toarray()
2. 模型创建：然后，你需要创建你的模型。在这个例子中，我们创建了一个Inception网络的实例。
model = Classifier_INCEPTION(output_directory='output_directory', input_shape=X_train.shape, nb_classes=8)
3. 模型训练：接下来，你需要训练你的模型。在这个例子中，我们使用训练数据和验证数据来训练我们的模型。
model.fit(X_train, y_train, X_val, y_val,y_true)
4. 模型预测：最后，你可以使用你的模型进行预测。在这个例子中，我们使用测试数据来进行预测，并打印出预测的结果。
df_metrics=model.predict(X_test, y_true1, X_train, y_train, y_test)
print(df_metrics)
请注意，你需要根据你自己的数据来调整这些步骤。例如，你可能需要调整数据的读取和预处理方式，以适应你自己的数据格式。你也可能需要调整模型的参数，以适应你的任务需求。
页:
[1]