import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader from torchvision.transforms import ToTensor import numpy as np

定义CNN模型

class CNN(nn.Module): def init(self): super(CNN, self).init() self.conv1 = nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1) self.relu = nn.ReLU(inplace=True) self.pool = nn.MaxPool2d(kernel_size=2, stride=2) self.conv2 = nn.Conv2d(16, 32, kernel_size=3, stride=1, padding=1) self.fc = nn.Linear(3288, 128)

def forward(self, x):
    x = x.view(-1, 1, 8, 8)  # 转换为图片格式，1表示通道数
    x = self.conv1(x)
    x = self.relu(x)
    x = self.pool(x)
    x = self.conv2(x)
    x = self.relu(x)
    x = self.pool(x)
    x = x.view(x.size(0), -1)
    x = self.fc(x)
    return x

定义BiGRU模型

class BiGRU(nn.Module): def init(self): super(BiGRU, self).init() self.gru = nn.GRU(input_size=23, hidden_size=64, num_layers=2, batch_first=True, bidirectional=True)

def forward(self, x):
    x, _ = self.gru(x)
    return x[:, -1, :]

定义特征融合模型

class FusionModel(nn.Module): def init(self): super(FusionModel, self).init() self.cnn_model = CNN() self.gru_model = BiGRU() self.fc1 = nn.Linear(128+128, 64) self.fc2 = nn.Linear(64, 8)

def forward(self, x_img, x_gru):
    x_img = self.cnn_model(x_img)
    x_gru = self.gru_model(x_gru)
    x = torch.cat((x_img, x_gru), dim=1)
    x = self.fc1(x)
    x = self.fc2(x)
    return x

定义数据集类

class MyDataset(Dataset): def init(self, txt_path, transform=None): self.data = np.loadtxt(txt_path, delimiter=',') # 读取数据 self.transform = transform

def __getitem__(self, index):
    features = self.data[index, :-1]  # 获取特征值
    label = self.data[index, -1]  # 获取标签
    if self.transform:
        features = self.transform(features)
    return features, label

def __len__(self):
    return len(self.data)

定义转换函数

def transform(features): return torch.tensor(features, dtype=torch.float32)

定义训练函数

def train(model, train_loader, optimizer, criterion): model.train() running_loss = 0.0 for features, labels in train_loader: optimizer.zero_grad() output = model(features[0], features[1]) loss = criterion(output, labels) loss.backward() optimizer.step() running_loss += loss.item() return running_loss / len(train_loader)

定义验证函数

def validate(model, val_loader, criterion): model.eval() running_loss = 0.0 correct = 0 total = 0 with torch.no_grad(): for features, labels in val_loader: output = model(features[0], features[1]) loss = criterion(output, labels) running_loss += loss.item() _, predicted = torch.max(output.data, 1) total += labels.size(0) correct += (predicted == labels).sum().item() val_loss = running_loss / len(val_loader) val_acc = correct / total return val_loss, val_acc

定义测试函数

def test(model, test_loader): model.eval() correct = 0 total = 0 with torch.no_grad(): for features, labels in test_loader: output = model(features[0], features[1]) _, predicted = torch.max(output.data, 1) total += labels.size(0) correct += (predicted == labels).sum().item() test_acc = correct / total return test_acc

设置随机种子

seed = 1234 torch.manual_seed(seed) torch.cuda.manual_seed(seed) np.random.seed(seed)

设置参数

batch_size = 32 learning_rate = 0.001 num_epochs = 10

加载数据集

train_dataset = MyDataset('训练集.txt', transform=transform) val_dataset = MyDataset('验证集.txt', transform=transform) test_dataset = MyDataset('测试集.txt', transform=transform)

创建数据加载器

train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False) test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

创建模型和优化器

model = FusionModel() optimizer = optim.Adam(model.parameters(), lr=learning_rate) criterion = nn.CrossEntropyLoss()

训练模型

best_val_acc = 0.0 for epoch in range(num_epochs): train_loss = train(model, train_loader, optimizer, criterion) val_loss, val_acc = validate(model, val_loader, criterion) if val_acc > best_val_acc: best_val_acc = val_acc torch.save(model.state_dict(), 'best_model.pt') print(f'Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}')

加载最佳模型并进行测试

model.load_state_dict(torch.load('best_model.pt')) test_acc = test(model, test_loader) print(f'Test Accuracy: {test_acc:.4f}')