1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
| import torch import torch.nn as nn
class SelectiveSSM(nn.Module): """选择性状态空间模型(Mamba风格简化版)""" def __init__(self, d_model: int, d_state: int = 16): super().__init__() self.d_model = d_model self.d_state = d_state self.proj_in = nn.Linear(d_model, d_model * 2) self.A = nn.Parameter(torch.randn(d_state, d_model) * 0.01) self.B = nn.Parameter(torch.randn(d_state, d_model) * 0.01) self.C = nn.Parameter(torch.randn(d_model, d_state) * 0.01) self.dt = nn.Parameter(torch.ones(d_model) * 0.1) self.proj_out = nn.Linear(d_model, d_model) def forward(self, x): """x: [B, T, D]""" x = self.proj_in(x) a, b = x.chunk(2, dim=-1) x = a * torch.sigmoid(b) dt = torch.softplus(self.dt) A_bar = torch.exp(-dt.unsqueeze(0) * self.A) h = torch.zeros(x.shape[0], self.d_state, self.d_model, device=x.device) outputs = [] for t in range(x.shape[1]): B_t = self.B * x[:, t:t+1, :].mean(dim=1, keepdim=True).transpose(0, 1) h = A_bar * h + B_t y = torch.einsum('bd,ds->bs', h.mean(dim=1), self.C) outputs.append(y.unsqueeze(1)) out = torch.cat(outputs, dim=1) return self.proj_out(out + x)
class TriStreamNTSM(nn.Module): """NTSM: 三流选择性状态空间模型""" def __init__(self, n_classes=3): super().__init__() self.spectral_ssm = SelectiveSSM(64) self.spectral_proj = nn.Linear(5, 64) self.spatial_ssm = SelectiveSSM(64) self.spatial_proj = nn.Linear(17, 64) self.eog_ssm = SelectiveSSM(32) self.eog_proj = nn.Linear(6, 32) self.fusion = nn.Sequential( nn.Linear(64 + 64 + 32, 128), nn.ReLU(), nn.Dropout(0.2), nn.Linear(128, n_classes) ) def forward(self, eeg_spectral, eeg_spatial, eog_features): s = self.spectral_proj(eeg_spectral) s = self.spectral_ssm(s) p = self.spatial_proj(eeg_spatial) p = self.spatial_ssm(p) e = self.eog_proj(eog_features) e = self.eog_ssm(e) s = s.mean(dim=1) p = p.mean(dim=1) e = e.mean(dim=1) fused = torch.cat([s, p, e], dim=-1) return self.fusion(fused)
if __name__ == "__main__": model = TriStreamNTSM(n_classes=3) spec = torch.randn(4, 100, 5) spat = torch.randn(4, 100, 17) eog = torch.randn(4, 100, 6) out = model(spec, spat, eog) print(f"NTSM输出: {out.shape}") print(f"参数: {sum(p.numel() for p in model.parameters()):,}")
|