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 104 105 106 107 108 109 110 111 112 113 114 115
| import numpy as np from typing import Tuple
class CSIPreprocessor: """ CSI信号预处理器 去噪、归一化、特征提取 """ def __init__(self, n_subcarriers: int = 56, n_tx: int = 2, n_rx: int = 2): self.n_subcarriers = n_subcarriers self.n_tx = n_tx self.n_rx = n_rx def preprocess(self, csi_raw: np.ndarray, window_sec: float = 5.0, sample_rate: float = 100.0) -> np.ndarray: """ 预处理CSI数据 Args: csi_raw: 原始CSI数据 (T, N_sub, N_tx, N_rx, 2) window_sec: 分析窗口(秒) sample_rate: 采样率 Returns: csi_features: 预处理后的特征图 """ window_samples = int(window_sec * sample_rate) amplitude = np.abs(csi_raw[-window_samples:, :, :, :, 0] + 1j * csi_raw[-window_samples:, :, :, :, 1]) amplitude = amplitude - np.mean(amplitude, axis=0, keepdims=True) amplitude = amplitude / (np.std(amplitude) + 1e-6) from scipy.signal import butter, filtfilt b, a = butter(4, 0.3, btype='low') amplitude_filtered = filtfilt(b, a, amplitude, axis=0) return amplitude_filtered def extract_features(self, csi_clean: np.ndarray) -> np.ndarray: """ 提取特征 Args: csi_clean: 清洗后的CSI数据 Returns: features: 特征向量 """ features = [] features.append(np.mean(csi_clean, axis=0).flatten()) features.append(np.std(csi_clean, axis=0).flatten()) features.append(np.max(csi_clean, axis=0).flatten()) features.append(np.min(csi_clean, axis=0).flatten()) fft_result = np.fft.fft(csi_clean, axis=0) features.append(np.abs(fft_result).mean(axis=0).flatten()) features = np.concatenate(features) return features
class DeepCPD_Model(nn.Module): """ DeepCPD深度学习模型 基于CSI特征检测儿童存在 """ def __init__(self, input_dim: int = 1024, hidden_dim: int = 256): super().__init__() self.classifier = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.3), nn.Linear(hidden_dim, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 2) ) def forward(self, features: torch.Tensor) -> torch.Tensor: """ 前向传播 Args: features: CSI特征向量 (B, input_dim) Returns: logits: 分类输出 (B, 2) """ return self.classifier(features)
|