ClassificationSummary

预处理：

1.将时域信号只保留Result列

import os


root_input_folder = 'Dataset'
root_intermediate_folder = 'Dataset_preprocess'
if not os.path.exists(root_intermediate_folder):
    os.mkdir(root_intermediate_folder)


for subdir in os.listdir(root_input_folder):

    input_folder = os.path.join(root_input_folder, subdir)
    intermediate_folder = os.path.join(root_intermediate_folder, subdir)

    if not os.path.exists(intermediate_folder):
        os.mkdir(intermediate_folder)


    csv_files = [file for file in os.listdir(input_folder) if file.endswith(".csv")]


    for csv_file in csv_files:
        with open(os.path.join(input_folder, csv_file), 'r') as infile:
            lines = infile.readlines()
            result_data = []
            start = False
            row = 0
            for line in lines:
                if "++++++ Result Data ++++++" in line:
                    start = True
                    continue
                if start and line.strip():
                    row += 1
                    if row != 1:
                        result_value = line.strip().split(",")[2]
                        result_data.append(result_value)


        intermediate_file = os.path.join(intermediate_folder, f"{csv_file[:-4]}.csv")

        with open(intermediate_file, 'w') as outfile:
            for result_value in result_data:
                outfile.write(result_value + '\n')

print("Processing completed!")

2.对数据进行S-G去噪

import os
import pandas as pd
from scipy.signal import savgol_filter


# intermediate_folder = "TimeDomain_pro_c_n"
# denoise_folder = "TimeDomain_pro_c_n_dn"
#
# if not os.path.exists(denoise_folder):
#     os.mkdir(denoise_folder)
#
# csv_files = [file for file in os.listdir(intermediate_folder) if file.endswith(".csv")]

# S-G
def denoise_savgol(data, window_length=11, polyorder=2):
    return savgol_filter(data, window_length, polyorder)

#
# for csv_file in csv_files:
#
#     data = pd.read_csv(os.path.join(intermediate_folder, csv_file), header=None)
#
#     denoised_result = denoise_savgol(data.iloc[:, 0])
#
#     denoise_file = os.path.join(denoise_folder, f"{csv_file[:-4]}.csv")
#
#     pd.DataFrame(denoised_result).to_csv(denoise_file, index=False, header=False)
#
#
#
# import os


root_input_folder = 'Dataset_preprocess'
root_intermediate_folder = 'Dataset_preprocess_denoised'
if not os.path.exists(root_intermediate_folder):
    os.mkdir(root_intermediate_folder)


for subdir in os.listdir(root_input_folder):

    input_folder = os.path.join(root_input_folder, subdir)
    intermediate_folder = os.path.join(root_intermediate_folder, subdir)

    if not os.path.exists(intermediate_folder):
        os.mkdir(intermediate_folder)


    csv_files = [file for file in os.listdir(input_folder) if file.endswith(".csv")]


    for csv_file in csv_files:
        data = pd.read_csv(os.path.join(input_folder, csv_file), header=None)

        denoised_result = denoise_savgol(data.iloc[:, 0])

        denoise_file = os.path.join(intermediate_folder, f"{csv_file[:-4]}.csv")

        pd.DataFrame(denoised_result).to_csv(denoise_file, index=False, header=False)


print("Denoised completed!")

3.对去噪后数据进行裁剪，只保留波形最明显的2500个数据点

import os
import numpy as np


root_input_directory = 'Dataset_preprocess_denoised'
root_output_directory = 'Dataset_]_preprocess_denoised_crop'


os.makedirs(root_output_directory, exist_ok=True)


for subdir in os.listdir(root_input_directory):

    input_directory = os.path.join(root_input_directory, subdir)
    output_directory = os.path.join(root_output_directory, subdir)

    os.makedirs(output_directory, exist_ok=True)


    csv_files = sorted([file for file in os.listdir(input_directory) if file.endswith('.csv')])

    # 时域 8500 - 11000
    # start_index = 8500
    # end_index = 11000
    # 吸收率 75 - 225
    # Transmittance(log) 0 - 300
    start_index = 8500
    end_index = 11000



    for csv_file in csv_files:

        data = np.genfromtxt(os.path.join(input_directory, csv_file), delimiter=',')


        cropped_data = data[start_index:end_index]

        new_csv_file = os.path.splitext(csv_file)[0] + '.csv'


        output_path = os.path.join(output_directory, new_csv_file)
        np.savetxt(output_path, cropped_data, delimiter=',')

        print(f"已保存文件: {new_csv_file}")

print("裁剪完成。")

分类

1.1D-CNNs

利用卷积层自动提取时域信号中的特征，然后分类。

模型结构

model = Sequential()


model.add(Conv1D(16, 3, strides=1, padding='same', input_shape=(row_number, 1)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv1D(16, 3, strides=1, padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling1D(pool_size=2, strides=1))


model.add(Conv1D(32, 3, strides=1, padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv1D(32, 3, strides=1, padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling1D(pool_size=2, strides=1))


model.add(Conv1D(64, 3, strides=1, padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Conv1D(64, 3, strides=1, padding='same'))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(MaxPooling1D(pool_size=2, strides=1))

model.add(Flatten())
model.add(Dense(num_classes, activation='softmax'))


optimizer = Adam(learning_rate=0.000005, beta_1=0.9, beta_2=0.999)
model.compile(optimizer=optimizer, loss=categorical_crossentropy, metrics=['accuracy'])


model.summary()

checkpoint = ModelCheckpoint('best_model.h5', monitor='val_accuracy', mode='max', save_best_only=True, verbose=1)

# Train the model
history = model.fit(X_train, y_train, validation_data=(X_test, y_test), batch_size=32, epochs=500, callbacks=[checkpoint])

# Save the final model
model.save('final_model.h5')

模型结构说明

Conv1D层是卷积层，将输入数据视为1D向量处理。数字16、32和64是卷积层将学习的滤波器的数量。数字3是滤波器的宽度，即卷积核的大小。’Strides’设置为1，表示滤波器在沿输入数据扫描时每次移动一步。’Padding’设置为’same’，这意味着它将在输入数据的边缘用零填充，以保持输出大小与输入大小相同。

BatchNormalization层用于在每个批次中对前一层的激活进行归一化，即应用一种变换，使得平均激活接近0，激活标准差接近1。它有助于更快的学习和更高的总体准确性。

Activation层对输出应用激活函数。在这种情况下，使用的是’relu’（修正线性单元），如果输入是正的，它将直接输出输入，否则，它将输出零。

MaxPooling1D层用于对输入进行下采样，沿其空间维度（高度和宽度）。’Pool size’设置为2，意味着它将在2个值的窗口上取最大值。

Flatten层用于将输入展平。例如，如果对具有输入形状为(batch_size, 2,2)的层应用flatten，那么该层的输出形状将为(batch_size, 4)。

Dense层是常规的深度连接神经网络层。它是最常见和频繁使用的层。Dense层对输入执行以下操作并返回输出。

然后使用Adam优化器、分类交叉熵作为损失函数和准确性作为度量来编译模型。Adam优化器的学习率设置为0.000005，beta_1和beta_2参数分别设置为0.9和0.999。

然后使用fit方法训练模型，使用训练数据（X_train, y_train）、验证数据（X_test, y_test）、批量大小为32，并进行500个周期。检查点用于在验证准确性提高时保存模型权重。

最后，训练后将模型保存到’final_model.h5’。

训练结果

训练集和验证集

训练集准确度：0.99
验证集准确度：0.82

训练曲线：

混淆矩阵：

测试集

准确度：0.94 or 17/18

混淆矩阵

2.SVM + 寻优算法

import pandas as pd
import numpy as np
import os
import math
from sklearn import svm
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

STEP_MAX = 1.0
STEP_MIN = 0.1
T = 50


data_folder = 'Dataset_TA_'
labels = []
data = []

for class_folder in os.listdir(data_folder):
    class_folder_path = os.path.join(data_folder, class_folder)
    for file_name in os.listdir(class_folder_path):
        file_path = os.path.join(class_folder_path, file_name)
        sample_data = pd.read_csv(file_path, header=None, delimiter=',').values
        data.append(sample_data)
        labels.append(class_folder)

data = np.array(data)
labels = np.array(labels)
data = data.reshape((data.shape[0], -1))


label_encoder = LabelEncoder()
encoded_labels = label_encoder.fit_transform(labels)

X_train, X_val, y_train, y_val = train_test_split(data, encoded_labels, test_size=0.2, random_state=42)


def spcs_svm_optimization(X_train, y_train, X_val, y_val):
    best_score = 0.0
    best_params = {'C': 1.69, 'gamma': 0.0006}


    C = best_params['C']

    print("updating C")
    for t in range(1, T + 1):
        step = STEP_MIN + (STEP_MAX - STEP_MIN) * (1 - t / T)
        p_a = 1 - math.exp(-t ** 3 / T ** 7)




        svm_model = make_pipeline(StandardScaler(), svm.SVC(C=C, gamma='scale', kernel='rbf'))
        svm_model.fit(X_train, y_train)
        predictions = svm_model.predict(X_val)
        score = accuracy_score(y_val, predictions)


        if score > best_score:
            best_score = score
            best_params['C'] = C

        print(f"updating C: {t+1}/{T}, Best Score: {best_score:.4f}, Best Params: {best_params}")

        C = np.random.uniform(1.5,2) * step

    print("updating gamma")


    gamma = 0.0006


    for t in range(1, T + 1):
        step = STEP_MIN + (STEP_MAX - STEP_MIN) * (1 - t / T)
        p_a = 1 - math.exp(-t ** 3 / T ** 7)


        svm_model = make_pipeline(StandardScaler(), svm.SVC(C=C, gamma=gamma, kernel='rbf'))
        svm_model.fit(X_train, y_train)
        predictions = svm_model.predict(X_val)
        score = accuracy_score(y_val, predictions)


        if score > best_score:
            best_score = score
            best_params['gamma'] = gamma

        print(f"updating gamma: {t + 1}/{T}, Best Score: {best_score:.4f}, Best Params: {best_params}")
        gamma = abs(gamma * step + + np.random.uniform(-0.0001, 0.0001))

    return best_params



best_params = spcs_svm_optimization(X_train, y_train, X_val, y_val)


final_svm = make_pipeline(StandardScaler(), svm.SVC(C=best_params['C'], gamma=best_params['gamma'], kernel='rbf'))
final_svm.fit(X_train, y_train)


y_pred = final_svm.predict(X_val)
print('Accuracy:', accuracy_score(y_val, y_pred))



def predict_directory(directory_path):
    for filename in os.listdir(directory_path):
        if filename.endswith(".csv"):
            file_path = os.path.join(directory_path, filename)
            new_sample_data = pd.read_csv(file_path, header=None, delimiter=',').values.reshape(1, -1)
            new_sample_pred = final_svm.predict(new_sample_data)
            new_sample_label = label_encoder.inverse_transform(new_sample_pred)
            print(f"Predicted class for {filename} is: {new_sample_label[0]}")



directory_path = 'Testset_TA_less/0'
predict_directory(directory_path)

训练结果

训练集准确度：0.8481012658227848

测试集准确度：0.83 or 15/18

3.DRSN + bagging

import torch
import torch.nn as nn
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset, random_split
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import numpy as np
import pandas as pd
import os
from sklearn.metrics import f1_score, roc_auc_score

torch.cuda.empty_cache()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

class BasicBlock(nn.Module):
    expansion = 1

    def __init__(self, in_channels, out_channels, stride=1):
        super().__init__()
        self.shrinkage = Shrinkage(out_channels, gap_size=(1))
        # residual function
        self.residual_function = nn.Sequential(
            nn.Conv1d(in_channels, out_channels, kernel_size=3, stride=stride, padding=1, bias=False),
            nn.BatchNorm1d(out_channels),
            nn.ReLU(inplace=True),
            nn.Conv1d(out_channels, out_channels * BasicBlock.expansion, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm1d(out_channels * BasicBlock.expansion),
            self.shrinkage
        )
        # shortcut
        self.shortcut = nn.Sequential()

        # the shortcut output dimension is not the same with residual function
        # use 1*1 convolution to match the dimension
        if stride != 1 or in_channels != BasicBlock.expansion * out_channels:
            self.shortcut = nn.Sequential(
                nn.Conv1d(in_channels, out_channels * BasicBlock.expansion, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm1d(out_channels * BasicBlock.expansion)
            )

    def forward(self, x):

         return nn.ReLU(inplace=True)(self.residual_function(x) + self.shortcut(x))
        # a = self.residual_function(x),
        # b = self.shortcut(x),
        # c = a+b
        # return c


class Shrinkage(nn.Module):
    def __init__(self, channel, gap_size):
        super(Shrinkage, self).__init__()
        self.gap = nn.AdaptiveAvgPool1d(gap_size)
        self.fc = nn.Sequential(
            nn.Linear(channel, channel),
            nn.BatchNorm1d(channel),
            nn.ReLU(inplace=True),
            nn.Linear(channel, channel),
            nn.Sigmoid(),
        )

    def forward(self, x):
        x_raw = x
        x = torch.abs(x)
        x_abs = x
        x = self.gap(x)
        x = torch.flatten(x, 1)
        average = torch.mean(x, dim=1, keepdim=True)  #CS
        # average = x    #CW
        x = self.fc(x)
        x = torch.mul(average, x)
        x = x.unsqueeze(2)
        # soft thresholding
        sub = x_abs - x
        zeros = sub - sub
        n_sub = torch.max(sub, zeros)
        x = torch.mul(torch.sign(x_raw), n_sub)
        return x


class RSNet(nn.Module):

    def __init__(self, block, num_block, num_classes=17):
        super().__init__()

        self.in_channels = 64

        self.conv1 = nn.Sequential(
            nn.Conv1d(1, 64, kernel_size=3, padding=1, bias=False),
            nn.BatchNorm1d(64),
            nn.ReLU(inplace=True))
        # we use a different inputsize than the original paper
        # so conv2_x's stride is 1
        self.conv2_x = self._make_layer(block, 64, num_block[0], 1)
        self.conv3_x = self._make_layer(block, 128, num_block[1], 2)
        self.conv4_x = self._make_layer(block, 256, num_block[2], 2)
        self.conv5_x = self._make_layer(block, 512, num_block[3], 2)
        #self.conv6_x = self._make_layer(block, 1024, num_block[4], 2)
        self.avg_pool = nn.AdaptiveAvgPool1d((1))
        # self.dropout1 = nn.Dropout(0.5)
        self.fc = nn.Linear(512 * block.expansion, num_classes)

    def _make_layer(self, block, out_channels, num_blocks, stride):
        """make rsnet layers(by layer i didnt mean this 'layer' was the
        same as a neuron netowork layer, ex. conv layer), one layer may
        contain more than one residual shrinkage block

        Args:
            block: block type, basic block or bottle neck block
            out_channels: output depth channel number of this layer
            num_blocks: how many blocks per layer
            stride: the stride of the first block of this layer

        Return:
            return a rsnet layer
        """

        # we have num_block blocks per layer, the first block
        # could be 1 or 2, other blocks would always be 1
        strides = [stride] + [1] * (num_blocks - 1)
        layers = []
        for stride in strides:
            layers.append(block(self.in_channels, out_channels, stride))
            self.in_channels = out_channels * block.expansion

        return nn.Sequential(*layers)

    def forward(self, x):
        output = self.conv1(x)
        output = self.conv2_x(output)
        output = self.conv3_x(output)
        output = self.conv4_x(output)
        output = self.conv5_x(output)
      #  output = self.conv6_x(output)
        output = self.avg_pool(output)
        output = output.view(output.size(0), -1)
       # output = self.dropout1(output)
        output = self.fc(output)

        return output


def rsnet18():
    """ return a RSNet 18 object
    """
    return RSNet(BasicBlock, [2, 2, 2, 2])


def rsnet34():
    """ return a RSNet 34 object
    """
    return RSNet(BasicBlock, [3, 4, 6, 3, 2])

def save_models(models, model_base_name):
    for i, model in enumerate(models):
        torch.save(model.state_dict(), f"{model_base_name}_{i}.pth")

def load_models(model_class, model_base_name, num_models, device):
    models = []
    for i in range(num_models):
        model = model_class().to(device)
        model.load_state_dict(torch.load(f"{model_base_name}_{i}.pth"))
        models.append(model)
    return models


def ensemble_predict(models, data_loader, device):
    all_preds = []
    for model in models:
        model_preds = []
        model.eval()
        with torch.no_grad():
            for data, _ in data_loader:
                data = data.to(device)
                outputs = model(data)
                _, predicted = torch.max(outputs.data, 1)
                model_preds.append(predicted.cpu().numpy())
        all_preds.append(np.concatenate(model_preds))

    # Transpose to get predictions per instance rather than per model
    all_preds = np.array(all_preds).T

    # Use a simple majority vote for the final prediction
    final_preds = []
    for preds in all_preds:
        (values, counts) = np.unique(preds, return_counts=True)
        index = np.argmax(counts)
        final_preds.append(values[index])

    return np.array(final_preds)

if __name__=='__main__':


    row_number = 2650
    column_number = 1
    def load_data(data_dir):
        X, y = [], []
        incorrect_shape_count = 0
        for category in os.listdir(data_dir):
            category_path = os.path.join(data_dir, category)
            for file in os.listdir(category_path):
                file_path = os.path.join(category_path, file)
                data = pd.read_csv(file_path, header=None).values
                if data.shape != (row_number, column_number):  # shape  （1049,1)
                    incorrect_shape_count += 1
                    print(f"Incorrect shape for file: {file_path}. Shape is {data.shape}.")
                    continue
                X.append(data)
                y.append(category)
        X = np.array(X)
        y = np.array(y)
        print(f"Total files with incorrect shape: {incorrect_shape_count}")
        return X, y




    def load_pretrained_model(model, path):
        # Load the state_dict from the file
        pretrained_state_dict = torch.load(path)

        # Remove the 'crnn.' prefix from each key
        corrected_state_dict = {key.replace('', ''): value for key, value in pretrained_state_dict.items()}

        # Load the corrected state_dict into the model
        model.load_state_dict(corrected_state_dict)
        return model


    data_dir = 'Dataset_TA_'
    X, y = load_data(data_dir)

    X = X.reshape(-1, column_number, row_number)
    label_encoder = LabelEncoder()
    y_encoded = label_encoder.fit_transform(y)

    X_train, X_test, y_train, y_test = train_test_split(X, y_encoded, train_size=0.8, test_size=0.2, random_state=42)

    X_train_tensor = torch.tensor(X_train, dtype=torch.float32).to(device)
    y_train_tensor = torch.tensor(y_train, dtype=torch.long).to(device)
    X_test_tensor = torch.tensor(X_test, dtype=torch.float32).to(device)
    y_test_tensor = torch.tensor(y_test, dtype=torch.long).to(device)

    train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

    test_dataset = TensorDataset(X_test_tensor, y_test_tensor)
    test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

    num_models = 5  # 假设我们训练5个模型
    models = []
    best_accs = [0] * num_models

    for i in range(num_models):
        model = rsnet34()
        model = model.to(device)
        criterion = nn.CrossEntropyLoss()
        optimizer = optim.Adam(model.parameters(), lr=0.00001)

        num_epochs = 500
        best_acc = 0

        for epoch in range(num_epochs):
            model.train()
            total_loss = 0
            correct_train = 0
            total_train = 0
            for batch_idx, (data, target) in enumerate(train_loader):
                data, target = data.to(device), target.to(device)
                optimizer.zero_grad()
                outputs = model(data)
                loss = criterion(outputs, target)
                loss.backward()
                optimizer.step()
                total_loss += loss.item()

                _, predicted = torch.max(outputs.data, 1)
                total_train += target.size(0)
                correct_train += (predicted == target).sum().item()

            train_accuracy = 100 * correct_train / total_train

            model.eval()
            correct_val = 0
            total_val = 0
            val_loss = 0
            all_preds = []
            all_true = []
            with torch.no_grad():
                for data, target in test_loader:
                    data, target = data.to(device), target.to(device)
                    outputs = model(data)
                    loss = criterion(outputs, target)
                    val_loss += loss.item()

                    _, predicted = torch.max(outputs.data, 1)
                    total_val += target.size(0)
                    correct_val += (predicted == target).sum().item()

                    all_preds.extend(predicted.cpu().numpy())
                    all_true.extend(target.cpu().numpy())

            val_accuracy = 100 * correct_val / total_val
            f1 = f1_score(all_true, all_preds, average='weighted')
            val_loss = val_loss / len(test_loader)

            print(f"Model {i}, Epoch {epoch + 1}/{num_epochs}, Train Loss: {total_loss / len(train_loader):.4f}, "
                  f"Train Accuracy: {train_accuracy:.2f}%, Val Loss: {val_loss:.4f}, "
                  f"Val Accuracy: {val_accuracy:.2f}%, F1 Score: {f1:.4f}")

            if 100 * correct_val / total_val > best_acc:
                best_acc = 100 * correct_val / total_val
                print(f"Model {i}, save model, best acc: {best_acc} %")
                torch.save(model.state_dict(), f'DRSN_model_inter_{i}.pth')

        models.append(model)
        best_accs[i] = best_acc
    save_models(models, "DRSN_model_final")


    loaded_models = load_models(rsnet34, "DRSN_model_final", num_models, device)

    final_preds = ensemble_predict(loaded_models, test_loader, device)


    all_true = []
    with torch.no_grad():
        for _, target in test_loader:
            all_true.extend(target.cpu().numpy())
    ensemble_accuracy = np.mean(final_preds == np.array(all_true)) * 100
    print(f"Ensemble Model Accuracy: {ensemble_accuracy:.2f}%")

训练结果

测试集准确度：0.7~0.8

4.CRNN + SVM

import os
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from torch.utils.data import DataLoader, TensorDataset
from sklearn.metrics import f1_score, roc_auc_score
import torch.nn.functional as F
import random
from torch.optim.lr_scheduler import ReduceLROnPlateau
from torch.utils.tensorboard import SummaryWriter
from sklearn import svm
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline


torch.cuda.empty_cache()
writer = SummaryWriter('runs/03')

# shape (2500,1)
column_number = 1
row_number = 2650

# set hyperparameter
num_epochs = 1000
learning_rate = 1e-6

# set seeds
seed = 13
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
np.random.seed(seed)
random.seed(seed)
torch.manual_seed(seed)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

# loading data
def load_data(data_dir):
    X, y = [], []
    incorrect_shape_count = 0
    for category in os.listdir(data_dir):
        category_path = os.path.join(data_dir, category)
        for file in os.listdir(category_path):
            file_path = os.path.join(category_path, file)
            data = pd.read_csv(file_path, header=None).values
            if data.shape != (row_number, column_number):
                incorrect_shape_count += 1
                print(f"Incorrect shape for file: {file_path}. Shape is {data.shape}.")
                continue
            X.append(data)
            y.append(category)
    X = np.array(X)
    y = np.array(y)
    print(f"Total files with incorrect shape: {incorrect_shape_count}")
    return X, y

# class CRNN(nn.Module):
#     def __init__(self, num_classes):
#         super(CRNN, self).__init__()
#         self.conv1 = nn.Conv1d(column_number, 16, 7, padding=3)  # First convolutional layer
#         self.conv2 = nn.Conv1d(16, 32, 7, padding=3)  # Second convolutional layer
#         self.maxpool1 = nn.MaxPool1d(kernel_size=2)  # Adjusted kernel_size to 2
#         self.bn1 = nn.BatchNorm1d(32)  # Adjusted input channels to 32
#         self.lstm1 = nn.LSTM(32, 64, batch_first=True)  # Removed one LSTM layer
#         self.fc1 = nn.Linear(64 * (row_number // 2), 128)  # Adjusted input size based on max pooling and LSTM output
#         self.bn2 = nn.BatchNorm1d(128)
#         self.fc2 = nn.Linear(128, num_classes)
#         self.dropout = nn.Dropout(0.5)
#
#     def forward(self, x):
#         x = F.relu(self.bn1(self.maxpool1(self.conv2(F.relu(self.conv1(x))))))
#         # x = self.dropout(x)
#         x = x.permute(0, 2, 1)  # Swap dimensions for RNN layer (batch_size, sequence_length, input_size)
#         x, _ = self.lstm1(x)  # Passing through the remaining LSTM layer
#         x = x.contiguous().view(x.size(0), -1)  # Flatten LSTM output for fully connected layer
#         x = F.relu(self.bn2(self.fc1(x)))
#         # x = self.dropout(x)
#         x = self.fc2(x)
#         return x


class CRNN(nn.Module):
    def __init__(self, num_classes):
        super(CRNN, self).__init__()
        self.conv1 = nn.Conv1d(column_number, 16, 7, padding=3)  # Keeping one convolutional layer
        # self.conv2 = nn.Conv1d(16, 16, 7, padding=3)
        self.maxpool1 = nn.MaxPool1d(kernel_size=1)
        self.bn1 = nn.BatchNorm1d(16)
        self.lstm1 = nn.LSTM(1, 32, batch_first=True)
        self.lstm2 = nn.LSTM(32, 64, batch_first=True)  # Original LSTM layer
        self.fc1 = nn.Linear(64 * row_number, 128)  # Adjust input size based on LSTM output
        self.bn2 = nn.BatchNorm1d(128)
        self.fc2 = nn.Linear(128, num_classes)
        self.dropout = nn.Dropout(0.5)

    def forward(self, x):
        # x = F.relu(self.bn1(self.conv1(x)))
        # # x = F.relu(self.bn1(self.conv2(x)))
        # x = self.maxpool1(x)
        # x = self.dropout(x)
        x = x.permute(0, 2, 1)  # Swap dimensions for RNN layer (batch_size, sequence_length, input_size)
        x, _ = self.lstm1(x)  # Passing through the first LSTM layer
        x, _ = self.lstm2(x)  # Passing through the second LSTM layer
        x = x.contiguous().view(x.size(0), -1)  # Flatten LSTM output for fully connected layer
        x = F.relu(self.bn2(self.fc1(x)))
        x = self.dropout(x)
        x = self.fc2(x)
        return x




class CRNN_SVM(nn.Module):
    def __init__(self, crnn_model):
        super(CRNN_SVM, self).__init__()
        self.crnn = crnn_model
        self.svm = None

    def forward(self, x):
        x = self.crnn(x)
        if self.svm:
            x = torch.tensor(self.svm.decision_function(x.detach().cpu().numpy()), device=x.device)
        return x

    def fit_svm(self, X, y):
        self.svm = Pipeline([
            ('scaler', StandardScaler()),
            ('svc', svm.SVC(kernel='rbf'))
        ])
        self.svm.fit(X.detach().cpu().numpy(), y.detach().cpu().numpy())

# def interactive_learning(model, data_loader, optimizer, criterion):
#     model.train()  # 确保模型处于训练模式
#     for data, target in data_loader:
#         data, target = data.to(device), target.to(device)
#         outputs = model(data)
#         _, predicted = torch.max(outputs.data, 1)
#         for i in range(len(predicted)):
#             print(f"Model Prediction: {predicted[i].item()}, Actual Label: {target[i].item()}")
#             if predicted[i].item() != target[i].item():
#                 print("wrong anser")
#                 correction = target[i].item()
#             else:
#                 correction = ''
#             if correction:
#                 target_corrected = torch.tensor(int(correction)).to(device)
#                 # 在这里，您可以使用您的优化器和损失函数来更新模型
#                 optimizer.zero_grad()
#                 loss = criterion(outputs[i].unsqueeze(0), target_corrected.unsqueeze(0))
#                 loss.backward(retain_graph=True)  # 修改在这里
#                 optimizer.step()

def interactive_learning(model, data_loader, optimizer, criterion, batch_size=5):
    model.train()  # 确保模型处于训练模式
    incorrect_data = []
    incorrect_targets = []
    for data, target in data_loader:
        data, target = data.to(device), target.to(device)
        outputs = model(data)
        _, predicted = torch.max(outputs.data, 1)
        for i in range(len(predicted)):
            if predicted[i].item() != target[i].item():
                incorrect_data.append(data[i])
                incorrect_targets.append(target[i])

    # 如果收集到的错误样本数量达到了指定的批量大小，则进行优化
    if len(incorrect_data) >= batch_size:
        print("optimizing...")
        incorrect_data = torch.stack(incorrect_data)[:batch_size]
        incorrect_targets = torch.tensor(incorrect_targets, dtype=torch.long, device=device)[:batch_size]
        optimizer.zero_grad()
        outputs = model(incorrect_data)
        loss = criterion(outputs, incorrect_targets)
        loss.backward()
        optimizer.step()
        # 重新初始化incorrect_data和incorrect_targets，以便于下次收集
        incorrect_data = []
        incorrect_targets = []

# main
data_dir = 'Dataset_TA_'
X, y = load_data(data_dir)

X = torch.tensor(X).permute(0, 2, 1)
label_encoder = LabelEncoder()
y_encoded = label_encoder.fit_transform(y)

# split dataset
X_train, X_test, y_train, y_test = train_test_split(X, y_encoded, test_size=0.2, random_state=42)

# to tensor
X_train_tensor = torch.tensor(X_train, dtype=torch.float32).to(device)
y_train_tensor = torch.tensor(y_train, dtype=torch.long).to(device)
X_test_tensor = torch.tensor(X_test, dtype=torch.float32).to(device)
y_test_tensor = torch.tensor(y_test, dtype=torch.long).to(device)

# loading
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_dataset = TensorDataset(X_test_tensor, y_test_tensor)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

# init
crnn_model = CRNN(num_classes=len(np.unique(y)))
crnn_model = crnn_model.to(device)
crnn_svm_model = CRNN_SVM(crnn_model)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(crnn_model.parameters(), lr=learning_rate, weight_decay=0.01)
scheduler = ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=1000, verbose=True)

# train CRNN
best_acc = 0
for epoch in range(num_epochs):
    crnn_model.train()
    total_loss = 0
    correct_train = 0
    total_train = 0
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(device), target.to(device)
        optimizer.zero_grad()
        outputs = crnn_model(data)
        loss = criterion(outputs, target)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()

        _, predicted = torch.max(outputs.data, 1)
        total_train += target.size(0)
        correct_train += (predicted == target).sum().item()

    # evaluate in val data
    crnn_model.eval()
    correct_val = 0
    total_val = 0
    val_loss = 0
    all_preds = []
    all_true = []
    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(device), target.to(device)
            outputs = crnn_model(data)
            loss = criterion(outputs, target)
            val_loss += loss.item()

            _, predicted = torch.max(outputs.data, 1)
            total_val += target.size(0)
            correct_val += (predicted == target).sum().item()

            all_preds.extend(predicted.cpu().numpy())
            all_true.extend(target.cpu().numpy())

    train_accuracy = 100 * correct_train / total_train

    val_accuracy = 100 * correct_val / total_val
    f1 = f1_score(all_true, all_preds, average='weighted')

    val_loss = val_loss / len(test_loader)

    if epoch % 10 == 5:
        interactive_learning(crnn_model, test_loader, optimizer, criterion)

    if epoch % 10 == 0:
        crnn_model.eval()
        with torch.no_grad():
            train_features = crnn_model(X_train_tensor)
            test_features = crnn_model(X_test_tensor)

        crnn_svm_model.fit_svm(train_features, y_train_tensor)

        # evaluate combined model
        crnn_svm_model.eval()
        correct = 0
        total = 0
        with torch.no_grad():
            for data, target in test_loader:
                data, target = data.to(device), target.to(device)
                outputs = crnn_svm_model(data)
                _, predicted = torch.max(outputs.data, 1)
                total += target.size(0)
                correct += (predicted == target).sum().item()
        print(f"Accuracy on test set: {100 * correct / total}%")

    if 100 * correct / total > best_acc:
        best_acc = 100 * correct / total
        print(f"save model,best acc: {100 * correct / total} %")
        torch.save(crnn_svm_model.state_dict(), 'CRNN_SVM_model_T.pth')

    print(f"Epoch {epoch + 1}/{num_epochs}, Train Loss: {total_loss / len(train_loader):.4f}, "
          f"Train Accuracy: {train_accuracy:.2f}%, Val Loss: {val_loss:.4f}, "
          f"Val Accuracy: {val_accuracy:.2f}%, F1 Score: {f1:.4f}")

    scheduler.step(total_loss)

# Train SVM
crnn_model.eval()
with torch.no_grad():
    train_features = crnn_model(X_train_tensor)
    test_features = crnn_model(X_test_tensor)

crnn_svm_model.fit_svm(train_features, y_train_tensor)

# evaluate combined model
crnn_svm_model.eval()
correct = 0
total = 0
with torch.no_grad():
    for data, target in test_loader:
        data, target = data.to(device), target.to(device)
        outputs = crnn_svm_model(data)
        _, predicted = torch.max(outputs.data, 1)
        total += target.size(0)
        correct += (predicted == target).sum().item()
print(f"Accuracy on test set: {100 * correct / total}%")

if 100 * correct / total > best_acc:
    best_acc = 100 * correct / total
    print(f"save model,best acc: {100 * correct / total} %")
    torch.save(crnn_svm_model.state_dict(), 'CRNN_SVM_model_T.pth')
np.save('classes.npy', label_encoder.classes_)
writer.close()

训练结果

测试集准确度：0.7~0.8