diff --git a/model/get_tool.py b/model/get_tool.py deleted file mode 100644 index 00a1303..0000000 --- a/model/get_tool.py +++ /dev/null @@ -1,378 +0,0 @@ -# -*- coding: utf-8 -*- -import os -import sys -import serial -import threading -import queue -import re -from collections import deque -import numpy as np -import torch -import torch.nn as nn - -import pyqtgraph as pg -from pyqtgraph.Qt import QtWidgets, QtCore, QtGui - -# Set random seed -torch.manual_seed(42) -np.random.seed(42) - -LABEL_MAP = {'draw': 0, 'stand-up': 1, 'wave': 2} -INV_LABEL_MAP = { - 0: '画圈', - 1: '起立', - 2: '挥手' -} - -# ----------------- Model Architecture ----------------- -class Advanced1DCNN(nn.Module): - def __init__(self, n_subcarriers=114, num_classes=3): - super(Advanced1DCNN, self).__init__() - self.conv1 = nn.Sequential( - nn.Conv1d(n_subcarriers, 64, kernel_size=5, padding=2), - nn.BatchNorm1d(64), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.MaxPool1d(2) - ) - self.conv2 = nn.Sequential( - nn.Conv1d(64, 128, kernel_size=5, padding=2), - nn.BatchNorm1d(128), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.MaxPool1d(2) - ) - self.conv3 = nn.Sequential( - nn.Conv1d(128, 256, kernel_size=5, padding=2), - nn.BatchNorm1d(256), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.AdaptiveAvgPool1d(4) - ) - self.classifier = nn.Sequential( - nn.Flatten(), - nn.Linear(256 * 4, 128), - nn.ReLU(), - nn.Dropout(0.5), - nn.Linear(128, num_classes) - ) - def forward(self, x): - x = self.conv1(x) - x = self.conv2(x) - x = self.conv3(x) - x = self.classifier(x) - return x - -# ----------------- Serial Receiver Thread ----------------- -class SerialReceiver(QtCore.QThread): - frame_received = QtCore.pyqtSignal(list) - log_message = QtCore.pyqtSignal(str) - - def __init__(self, port, baud): - super().__init__() - self.port = port - self.baud = baud - self.running = False - self.ser = None - - def run(self): - try: - self.ser = serial.Serial(self.port, self.baud, timeout=0.5) - self.running = True - self.log_message.emit(f"Successfully connected to {self.port} at {self.baud} bps.") - except Exception as e: - self.log_message.emit(f"Error opening serial port: {e}") - return - - # Pattern to capture data:[...] array - pattern = re.compile(r'data:\s*\[([^\]]+)\]') - fallback_pattern = re.compile(r'\[([^\]]+)\]') - - while self.running: - try: - if self.ser.in_waiting > 0: - line = self.ser.readline().decode('utf-8', errors='ignore').strip() - if not line: - continue - - match = pattern.search(line) or fallback_pattern.search(line) - if not match: - continue - - data_str = match.group(1) - data_list = [] - for x in data_str.split(','): - x = x.strip() - try: - data_list.append(int(x)) - except ValueError: - continue - - if len(data_list) > 2: - self.frame_received.emit(data_list) - except Exception as e: - self.log_message.emit(f"Serial read error: {e}") - self.msleep(100) - - if self.ser and self.ser.is_open: - self.ser.close() - self.log_message.emit("Serial port closed.") - - def stop(self): - self.running = False - self.wait() - -# ----------------- Main GUI Window ----------------- -class MainWindow(QtWidgets.QMainWindow): - def __init__(self): - super().__init__() - self.setWindowTitle("WiFi CSI Real-time Gesture Inference (CNN1D)") - self.resize(1100, 700) - - self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - self.model = None - self.receiver = None - - # CSI Buffer: sliding window of size 50 - self.csi_window = deque(maxlen=50) - self.motion_threshold = 2.5 # Threshold for motion detection (std dev) - - # Load model weights - self.load_model() - - # Build UI layout - self.init_ui() - - # Update Timer (approx. 60 FPS) - self.timer = QtCore.QTimer() - self.timer.timeout.connect(self.process_queue) - self.data_queue = queue.Queue(maxsize=2000) - - def load_model(self): - # Look for weights in both muti-model folder and root folder - model_path = os.path.join("muti-model", "best_cnn1d.pth") - if not os.path.exists(model_path): - model_path = "best_cnn1d.pth" - - self.model = Advanced1DCNN(n_subcarriers=114, num_classes=3).to(self.device) - if os.path.exists(model_path): - try: - self.model.load_state_dict(torch.load(model_path, map_location=self.device)) - self.model.eval() - print(f"Model loaded successfully from '{model_path}' on {self.device}") - except Exception as e: - print(f"Error loading weight parameters: {e}") - else: - print(f"Warning: CNN1D weight file not found at '{model_path}'. Running with random weights.") - - def init_ui(self): - # Central widget and layout - central_widget = QtWidgets.QWidget() - self.setCentralWidget(central_widget) - main_layout = QtWidgets.QHBoxLayout(central_widget) - - # 1. Left side: Plots - left_layout = QtWidgets.QVBoxLayout() - main_layout.addLayout(left_layout, stretch=7) - - # Plot 1: CSI Amplitude per carrier - self.amp_plot = pg.PlotWidget(title="Real-time CSI Amplitude (114 Subcarriers)") - self.amp_plot.setLabel('left', 'Amplitude') - self.amp_plot.setLabel('bottom', 'Subcarrier Index') - self.amp_curve = self.amp_plot.plot(pen=pg.mkPen('y', width=2)) - left_layout.addWidget(self.amp_plot) - - # Plot 2: CSI Waterfall (Time series) - self.waterfall_plot = pg.PlotWidget(title="Sliding Window CSI Waterfall (50 frames)") - self.waterfall_image = pg.ImageItem() - self.waterfall_plot.addItem(self.waterfall_image) - # Apply colormap to look like a heat map/waterfall - colormap = pg.colormap.get('viridis') - self.waterfall_image.setLookupTable(colormap.getLookupTable()) - left_layout.addWidget(self.waterfall_plot) - - # 2. Right side: Controls & Predictions - right_layout = QtWidgets.QVBoxLayout() - main_layout.addLayout(right_layout, stretch=3) - - # Group 1: Serial connection settings - conn_group = QtWidgets.QGroupBox("Serial Settings") - conn_layout = QtWidgets.QFormLayout(conn_group) - - self.port_input = QtWidgets.QLineEdit("COM3") - self.baud_input = QtWidgets.QComboBox() - self.baud_input.addItems(["921600", "115200", "57600", "9600"]) - self.connect_btn = QtWidgets.QPushButton("Connect") - self.connect_btn.clicked.connect(self.toggle_connection) - - conn_layout.addRow("Port:", self.port_input) - conn_layout.addRow("Baud Rate:", self.baud_input) - conn_layout.addRow(self.connect_btn) - right_layout.addWidget(conn_group) - - # Group 2: Inference Results panel - infer_group = QtWidgets.QGroupBox("Real-time Inference") - infer_layout = QtWidgets.QVBoxLayout(infer_group) - - # Large prediction display - self.pred_label = QtWidgets.QLabel("IDLE") - self.pred_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) - self.pred_label.setFont(QtGui.QFont("Microsoft YaHei", 36, QtGui.QFont.Weight.Bold)) - self.pred_label.setStyleSheet("color: #7f8c8d; background-color: #2c3e50; padding: 20px; border-radius: 10px;") - infer_layout.addWidget(self.pred_label) - - # Motion Level progress bar and text - self.motion_label = QtWidgets.QLabel("Motion Level: 0.00 / Threshold: 2.50") - self.motion_label.setFont(QtGui.QFont("Microsoft YaHei", 11)) - infer_layout.addWidget(self.motion_label) - - self.motion_bar = QtWidgets.QProgressBar() - self.motion_bar.setRange(0, 100) - self.motion_bar.setValue(0) - infer_layout.addWidget(self.motion_bar) - - self.motion_status = QtWidgets.QLabel("⚪ Motion Status: Standby") - self.motion_status.setFont(QtGui.QFont("Microsoft YaHei", 12)) - infer_layout.addWidget(self.motion_status) - - right_layout.addWidget(infer_group) - - # Group 3: Connection logs - log_group = QtWidgets.QGroupBox("System Logs") - log_layout = QtWidgets.QVBoxLayout(log_group) - self.log_text = QtWidgets.QPlainTextEdit() - self.log_text.setReadOnly(True) - log_layout.addWidget(self.log_text) - right_layout.addWidget(log_group) - - def append_log(self, text): - self.log_text.appendPlainText(text) - - def toggle_connection(self): - if self.receiver is None or not self.receiver.isRunning(): - # Start connection - port = self.port_input.text().strip() - baud = int(self.baud_input.currentText()) - self.receiver = SerialReceiver(port, baud) - self.receiver.frame_received.connect(self.enqueue_frame) - self.receiver.log_message.connect(self.append_log) - self.receiver.start() - self.connect_btn.setText("Disconnect") - self.timer.start(15) - else: - # Stop connection - self.receiver.stop() - self.receiver = None - self.connect_btn.setText("Connect") - self.timer.stop() - self.pred_label.setText("IDLE") - self.pred_label.setStyleSheet("color: #7f8c8d; background-color: #2c3e50; padding: 20px; border-radius: 10px;") - self.motion_status.setText("⚪ Motion Status: Disconnected") - - def enqueue_frame(self, csi_frame): - # Prevent queue overflow - if self.data_queue.full(): - try: - self.data_queue.get_nowait() - except queue.Empty: - pass - self.data_queue.put(csi_frame) - - def process_queue(self): - # Read all available items in the queue - new_frame_added = False - latest_amp = None - - while not self.data_queue.empty(): - raw_list = self.data_queue.get() - - # 1. Parse complex number amplitude - if len(raw_list) % 2 != 0: - raw_list = raw_list[:-1] - - imag = np.array(raw_list[0::2]) - real = np.array(raw_list[1::2]) - csi = real + 1j * imag - amplitude = np.abs(csi) - - # Filter non-zero subcarriers (ESP32 CSI active subcarriers) - amplitude = amplitude[amplitude > 0] - - # Ensure shape is exactly 114 - if len(amplitude) > 114: - amplitude = amplitude[:114] - elif len(amplitude) < 114: - amplitude = np.pad(amplitude, (0, 114 - len(amplitude)), 'edge') - - self.csi_window.append(amplitude) - latest_amp = amplitude - new_frame_added = True - - if not new_frame_added: - return - - # 2. Update CSI Line Plot - self.amp_curve.setData(np.arange(114), latest_amp) - - # 3. Handle inference and sliding window logic - if len(self.csi_window) == 50: - window_matrix = np.array(self.csi_window) # Shape: (50, 114) - - # Update waterfall plot (transpose for vertical time scroll) - self.waterfall_image.setImage(window_matrix.T, autoLevels=True) - - # Calculate Motion Level using variance/standard deviation - # We measure the variance over the temporal dimension (axis 0) across all subcarriers - subcarrier_stds = np.std(window_matrix, axis=0) - motion_val = float(subcarrier_stds.mean()) - - # Update motion labels and progress bar - self.motion_label.setText(f"Motion Level: {motion_val:.2f} / Threshold: {self.motion_threshold:.2f}") - bar_val = min(100, int(motion_val * 20)) # scale for UI - self.motion_bar.setValue(bar_val) - - # Run inference if motion level exceeds the trigger threshold - if motion_val >= self.motion_threshold: - self.motion_status.setText("🔴 Motion Status: ACTIVE MOTION") - - # Apply SR-Std Standardization ($\epsilon=2.0$) - mean = window_matrix.mean(axis=0, keepdims=True) - std = window_matrix.std(axis=0, keepdims=True) - window_norm = (window_matrix - mean) / (std + 2.0) - - # Reshape to (batch, subcarriers, time) = (1, 114, 50) - tensor_in = torch.tensor(window_norm, dtype=torch.float32).unsqueeze(0).permute(0, 2, 1) - tensor_in = tensor_in.to(self.device) - - # Predict - with torch.no_grad(): - logits = self.model(tensor_in) - _, predicted = torch.max(logits, 1) - pred_idx = predicted.item() - label_name = INV_LABEL_MAP[pred_idx] - - # Update UI Label color based on gesture prediction - color_map = { - '画圈': "#3498db", # Blue - '起立': "#2ecc71", # Green - '挥手': "#e67e22" # Orange/Red - } - color = color_map.get(label_name, "#ffffff") - self.pred_label.setText(label_name) - self.pred_label.setStyleSheet(f"color: white; background-color: {color}; padding: 20px; border-radius: 10px;") - else: - self.motion_status.setText("⚪ Motion Status: Standby") - self.pred_label.setText("NO MOTION") - self.pred_label.setStyleSheet("color: #7f8c8d; background-color: #2c3e50; padding: 20px; border-radius: 10px;") - - def closeEvent(self, event): - if self.receiver and self.receiver.isRunning(): - self.receiver.stop() - self.timer.stop() - event.accept() - -if __name__ == "__main__": - app = QtWidgets.QApplication(sys.argv) - win = MainWindow() - win.show() - sys.exit(app.exec()) diff --git a/model/main.py b/model/main.py deleted file mode 100644 index 7310f22..0000000 --- a/model/main.py +++ /dev/null @@ -1,200 +0,0 @@ -import torch -import torch.nn as nn -import torch.optim as optim -import numpy as np -import matplotlib.pyplot as plt -from torch.utils.data import DataLoader, TensorDataset -from sklearn.model_selection import train_test_split -from tqdm import tqdm - -# 1. 数据加载与预处理 -def load_and_preprocess(): - try: - amp1 = np.load("filter_dataset/amp/fist_data.npz", allow_pickle=True)['data'] - pha1 = np.load("filter_dataset/pha/fist_data.npz", allow_pickle=True)['data'] - amp2 = np.load("filter_dataset/amp/paper_data.npz", allow_pickle=True)['data'] - pha2 = np.load("filter_dataset/pha/paper_data.npz", allow_pickle=True)['data'] - amp3 = np.load("filter_dataset/amp/scissor_data.npz", allow_pickle=True)['data'] - pha3 = np.load("filter_dataset/pha/scissor_data.npz", allow_pickle=True)['data'] - print("成功加载本地 .npz 数据集") - except FileNotFoundError: - print("未找到文件") - - def format_to_torch(data): - data = data.reshape(-1, 100, 2, 166) - data = np.transpose(data, (0, 2, 1, 3)) - return torch.tensor(data) - - X = torch.cat([format_to_torch(np.concatenate((amp1, pha1), axis=2).astype(np.float32)), - format_to_torch(np.concatenate((amp2, pha2), axis=2).astype(np.float32)), - format_to_torch(np.concatenate((amp3, pha3), axis=2).astype(np.float32))], dim=0) - Y = torch.cat([torch.zeros(amp1.shape[0], dtype=torch.long), - torch.ones(amp2.shape[0], dtype=torch.long), - torch.full((amp3.shape[0],), 2, dtype=torch.long)]) - - return train_test_split(X, Y, test_size=0.2, random_state=42) - -#多层感知机 -class SimpleMLP(nn.Module): - def __init__(self, num_classes=3): - super(SimpleMLP, self).__init__() - self.input_dim = 2 * 100 * 166 - self.network = nn.Sequential( - nn.Flatten(), - nn.Linear(self.input_dim, 256), - nn.ReLU(), - nn.Linear(256, 128), - nn.ReLU(), - nn.Linear(128, num_classes) - ) - - def forward(self, x): - return self.network(x) -class CSIGestureCNN(nn.Module): - - def __init__(self, num_classes=3): - - super(CSIGestureCNN, self).__init__() - - self.conv_block = nn.Sequential( - - nn.Conv2d(2, 32, kernel_size=3, padding=1), - - nn.BatchNorm2d(32), - - nn.ReLU(), - - nn.MaxPool2d(2), - - nn.Conv2d(32, 64, kernel_size=3, padding=1), - - nn.BatchNorm2d(64), - - nn.ReLU(), - - nn.MaxPool2d(2) - - ) - - self.classifier = nn.Sequential( - - nn.Flatten(), - - nn.Linear(64 * 25 * 41, 128), - - nn.ReLU(), - - nn.Dropout(0.5), - - nn.Linear(128, num_classes) - - ) - - - - def forward(self, x): - - x = self.conv_block(x) - - x = self.classifier(x) - - return x - - - -if __name__ == "__main__": - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - x_train, x_test, y_train, y_test = load_and_preprocess() - - print(f"训练集大小: {x_train.shape[0]}") # 80% - print(f"测试集大小: {x_test.shape[0]}") # 20% - train_loader = DataLoader(TensorDataset(x_train, y_train), batch_size=64, shuffle=True) - test_loader = DataLoader(TensorDataset(x_test, y_test), batch_size=300, shuffle=False) - model = SimpleMLP(num_classes=3).to(device) - #model = CSIGestureCNN(num_classes=3).to(device) - criterion = nn.CrossEntropyLoss() - optimizer = optim.Adam(model.parameters(), lr=1e-5) - - - history = { - 'train_loss': [], - 'train_acc': [], - 'test_acc': [] - } - - epochs = 100 - print(f"\n开始在 {device} 上训练...") - - for epoch in range(epochs): - model.train() - running_loss = 0.0 - train_correct = 0 - train_total = 0 - - pbar = tqdm(train_loader, desc=f"Epoch [{epoch+1}/{epochs}]") - - for inputs, labels in pbar: - inputs, labels = inputs.to(device), labels.to(device) - - optimizer.zero_grad() - outputs = model(inputs) - loss = criterion(outputs, labels) - loss.backward() - optimizer.step() - - running_loss += loss.item() - - # 修改 2：计算训练集准确率 - _, predicted = torch.max(outputs.data, 1) - train_total += labels.size(0) - train_correct += (predicted == labels).sum().item() - - pbar.set_postfix({'Loss': f"{loss.item():.4f}"}) - - train_accuracy = 100 * train_correct / train_total - - # 计算测试准确率 - model.eval() - test_correct, test_total = 0, 0 - with torch.no_grad(): - for inputs, labels in test_loader: - inputs, labels = inputs.to(device), labels.to(device) - outputs = model(inputs) - _, predicted = torch.max(outputs.data, 1) - test_total += labels.size(0) - test_correct += (predicted == labels).sum().item() - - test_accuracy = 100 * test_correct / test_total - avg_loss = running_loss / len(train_loader) - - # 保存历史记录 - history['train_loss'].append(avg_loss) - history['train_acc'].append(train_accuracy) - history['test_acc'].append(test_accuracy) - - print(f"-> Epoch Summary: Loss: {avg_loss:.4f}, Train Acc: {train_accuracy:.2f}%, Test Acc: {test_accuracy:.2f}%") - - - plt.figure(figsize=(12, 5)) - - # 绘制 Loss 曲线 - plt.subplot(1, 2, 1) - plt.plot(range(1, epochs + 1), history['train_loss'], 'r-', label='Train Loss') - plt.title('Training Loss') - plt.xlabel('Epochs') - plt.ylabel('Loss') - plt.grid(True) - plt.legend() - - # 绘制 Accuracy 曲线 (同时显示训练和测试) - plt.subplot(1, 2, 2) - plt.plot(range(1, epochs + 1), history['train_acc'], 'g-', label='Train Accuracy') - plt.plot(range(1, epochs + 1), history['test_acc'], 'b-', label='Test Accuracy') - plt.title('Accuracy Comparison') - plt.xlabel('Epochs') - plt.ylabel('Accuracy (%)') - plt.grid(True) - plt.legend() - - plt.tight_layout() - plt.show() \ No newline at end of file diff --git a/model/muti-model/cnn1d_history.png b/model/muti-model/cnn1d_history.png deleted file mode 100644 index 2b43d38..0000000 Binary files a/model/muti-model/cnn1d_history.png and /dev/null differ diff --git a/model/muti-model/fcn_history.png b/model/muti-model/fcn_history.png deleted file mode 100644 index 91c9ad9..0000000 Binary files a/model/muti-model/fcn_history.png and /dev/null differ diff --git a/model/muti-model/lstm_history.png b/model/muti-model/lstm_history.png deleted file mode 100644 index 05a4b15..0000000 Binary files a/model/muti-model/lstm_history.png and /dev/null differ diff --git a/model/muti-model/mlp_history.png b/model/muti-model/mlp_history.png deleted file mode 100644 index 29f4739..0000000 Binary files a/model/muti-model/mlp_history.png and /dev/null differ diff --git a/model/muti-model/model_comparison.png b/model/muti-model/model_comparison.png deleted file mode 100644 index 2e6432c..0000000 Binary files a/model/muti-model/model_comparison.png and /dev/null differ diff --git a/model/muti-model/model_comparison_report.md b/model/muti-model/model_comparison_report.md deleted file mode 100644 index 0a78bd1..0000000 --- a/model/muti-model/model_comparison_report.md +++ /dev/null @@ -1,121 +0,0 @@ -# WiFi CSI 手势识别模型多架构评估报告 - -本报告旨在评估在**子载波正则化标准化 (SR-Std, $\epsilon=2.0$)** 预处理和**严格 50/50 顺序划分**（无 Shuffle）的条件下，五种不同的深度学习架构在 WiFi CSI 手势识别任务上的性能表现。 - ---- - -## 1. 实验设计与评估标准 - -为了解决会话内数据泄漏和时序分布偏移问题，我们统一采用以下标准： -1. **数据集划分**：只加载根目录 `dataset/dataset_2026_6_23` 数据。对每个类别，前 50% 样本作为训练集，后 50% 样本作为测试集，完全不打乱时序顺序。 -2. **特征归一化**：采用 **SR-Std (Subcarrier-wise Regularized Standardization)**，对每个样本窗口单独标准化，参数为 $\epsilon = 2.0$。 -3. **优化器与超参数**：统一使用 AdamW 优化器，学习率 1e-3，余弦退火调度器（CosineAnnealingLR），交叉熵损失（带 0.1 标签平滑），训练 30 个 Epoch，Batch Size 为 64。 - ---- - -## 2. 模型性能汇总 - -各模型在测试集上的最佳准确率（Best Test Accuracy）汇总如下表： - -| 模型架构 (Model) | 最佳测试集准确率 (Best Test Acc) | 训练收敛速度 (Convergence) | 时序泛化能力 (Temporal Generalization) | -| :--- | :---: | :---: | :--- | -| **CNN1D (Advanced 1D CNN)** | **87.28%** | 极快 (约 10 个 Epoch 稳定) | **优秀**。局部卷积对于时序平移不敏感。 | -| **FCN (Fully Convolutional Network)** | **86.90%** | 极快 (约 12 个 Epoch 稳定) | **优秀**。全局平均池化 (GAP) 提供了优异的翻译不变性。 | -| **LSTM (Bidirectional LSTM)** | **82.46%** | 较快 (约 15 个 Epoch 稳定) | **良好**。双向时序循环有利于学习手势动态，但对全局偏移敏感。 | -| **Transformer (Encoder)** | **82.24%** | 较快 (约 15 个 Epoch 稳定) | **良好**。自注意力机制学习长距离依赖，但数据量少时易发生微度过拟合。 | -| **MLP (Multi-Layer Perceptron)** | **79.95%** | 较慢 (约 25 个 Epoch 稳定) | **一般**。没有平移不变性，完全依赖全连接层拟合全局时序映射。 | - ---- - -## 3. 各模型详细分析与曲线 - -### 3.1 总体模型对比 - -各模型在训练过程中测试集准确率的收敛曲线对比图如下： - -![各模型测试准确率对比图](./model_comparison.png) - -* **观察**：**CNN1D** 与 **FCN** 属于第一梯队，测试准确率稳步爬升并率先突破 86%。**LSTM** 和 **Transformer** 属于第二梯队，在 82% 左右振荡。**MLP** 虽然起步极慢，但最终也达到了近 80% 的准确率，这进一步验证了 **SR-Std** 预处理能够极大降低模型对环境的过拟合程度。 - ---- - -### 3.2 1D CNN 模型 (CNN1D) - -CNN1D 模型包含多层一维卷积、批归一化、Dropout 以及池化层，并在特征提取后使用 AdaptiveAvgPool1d 映射到分类器。 - -* **最佳测试准确率**：**87.28%** -* **各类别表现**： - * `draw` (画圈)：**97.42%** - * `stand-up` (起立)：**68.91%** - * `wave` (挥手)：**89.03%** -* **训练曲线**： - -![CNN1D 训练曲线图](./cnn1d_history.png) - ---- - -### 3.3 全卷积网络 (FCN) - -FCN 使用了三层较宽的一维卷积，无池化，并在最顶层使用全局平均池化 (GAP) 提取时序全局均值特征。 - -* **最佳测试准确率**：**86.90%** -* **各类别表现**： - * `draw` (画圈)：**94.34%** - * `stand-up` (起立)：**76.02%** - * `wave` (挥手)：**89.57%** -* **训练曲线**： - -![FCN 训练曲线图](./fcn_history.png) - ---- - -### 3.4 双向 LSTM 模型 (LSTM) - -双向 2 层 LSTM 通过在时序上双向循环传播获取前后文信息，使用最后一个时间步的双向隐状态拼接后做分类。 - -* **最佳测试准确率**：**82.46%** -* **各类别表现**： - * `draw` (画圈)：**82.93%** - * `stand-up` (起立)：**80.98%** - * `wave` (挥手)：**83.60%** -* **训练曲线**： - -![LSTM 训练曲线图](./lstm_history.png) - ---- - -### 3.5 Transformer 编码器模型 (Transformer) - -在输入端使用一维 Linear 投影，叠加正弦位置编码，接着送入 2 层 4 头 TransformerEncoder，最后使用 Temporal Mean Pooling 和 MLP 分类器。 - -* **最佳测试准确率**：**82.24%** -* **各类别表现**： - * `draw` (画圈)：**88.31%** - * `stand-up` (起立)：**75.15%** - * `wave` (挥手)：**84.30%** -* **训练曲线**： - -![Transformer 训练曲线图](./transformer_history.png) - ---- - -### 3.6 MLP 模型 (MLP) - -MLP 将 `(50, 114)` 的时序图展平为长度 5700 的一维向量，通过包含 Dropout 的三层 Dense 进行分类。 - -* **最佳测试准确率**：**79.95%** -* **各类别表现**： - * `draw` (画圈)：**84.13%** - * `stand-up` (起立)：**82.21%** - * `wave` (挥手)：**69.93%** -* **训练曲线**： - -![MLP 训练曲线图](./mlp_history.png) - ---- - -## 4. 结论与总结 - -1. **时序平移不变性是关键**：CNN1D 和 FCN 由于其本身的感受野和全局平均池化机制，在面对 50/50 顺序划分（这意味着测试手势可能在时间窗口的不同位置发生）时表现最为稳健，准确率均超过了 **86%**。 -2. **序列级循环与注意力的泛化潜力**：LSTM 和 Transformer 在分类性能上非常均衡（各类别准确率的标准差极小），但由于容易过度关注长时序的特定绝对排列，泛化极限略微逊于 CNN 架构，获得约 **82%** 的测试准确率。 -3. **阻断泄露后标准化的核心价值**：即便在没有任何平移不变性的 MLP 架构上，配合 SR-Std 标准化，依然能取得 **79.95%** 的高准确率。这表明，**子载波独立去除环境静态多径反射，对于 WiFi CSI 信号处理具有无可替代的重要性。** diff --git a/model/muti-model/train_comparisons.py b/model/muti-model/train_comparisons.py deleted file mode 100644 index d11e881..0000000 --- a/model/muti-model/train_comparisons.py +++ /dev/null @@ -1,407 +0,0 @@ -import os -import sys -import numpy as np -import torch -import torch.nn as nn -import torch.optim as optim -from torch.utils.data import DataLoader, TensorDataset -import matplotlib.pyplot as plt - -# Set random seed for reproducibility -torch.manual_seed(42) -np.random.seed(42) - -LABEL_MAP = {'draw': 0, 'stand-up': 1, 'wave': 2} -INV_LABEL_MAP = {v: k for k, v in LABEL_MAP.items()} - -# ----------------- Data Loading & Preprocessing ----------------- -def load_dataset(dataset_dir="dataset/dataset_2026_6_23"): - if not os.path.exists(dataset_dir): - raise FileNotFoundError(f"Dataset directory '{dataset_dir}' not found.") - npz_files = sorted([f for f in os.listdir(dataset_dir) if f.endswith('.npz')]) - X_dict = {0: [], 1: [], 2: []} - - print(f"Loading dataset from {dataset_dir}...") - for filename in sorted(npz_files): - file_path = os.path.join(dataset_dir, filename) - label_name = filename.split('_')[0] - if label_name not in LABEL_MAP: - continue - label_idx = LABEL_MAP[label_name] - data = np.load(file_path, allow_pickle=True)['dataset'] - X_dict[label_idx].append(data) - - X_by_label = {} - for idx in X_dict: - X_by_label[idx] = np.concatenate(X_dict[idx], axis=0) - print(f" Class '{INV_LABEL_MAP[idx]}': {X_by_label[idx].shape[0]} samples") - - return X_by_label - -def split_50_50(X_by_label): - x_train_list, x_test_list = [], [] - y_train_list, y_test_list = [], [] - - print("\nSplitting dataset (50% train, 50% test sequentially)...") - for idx in sorted(X_by_label.keys()): - X = X_by_label[idx] - n_samples = len(X) - split_point = int(n_samples * 0.8) - - x_train_list.append(X[:split_point]) - x_test_list.append(X[split_point:]) - y_train_list.append(np.full(split_point, idx, dtype=np.int64)) - y_test_list.append(np.full(n_samples - split_point, idx, dtype=np.int64)) - print(f" Class '{INV_LABEL_MAP[idx]}': Train={split_point}, Test={n_samples - split_point}") - - x_train = np.concatenate(x_train_list, axis=0) - x_test = np.concatenate(x_test_list, axis=0) - y_train = np.concatenate(y_train_list, axis=0) - y_test = np.concatenate(y_test_list, axis=0) - - return x_train, x_test, y_train, y_test - -def preprocess_sr_std(X, eps=2.0): - # Subcarrier-wise Regularized Standardization (SR-Std) - X_norm = np.zeros_like(X) - for i in range(len(X)): - mean = X[i].mean(axis=0, keepdims=True) - std = X[i].std(axis=0, keepdims=True) - X_norm[i] = (X[i] - mean) / (std + eps) - return X_norm - -# ----------------- Model Architectures ----------------- - -# 1. Fully Convolutional Network (FCN) -class FCN(nn.Module): - def __init__(self, input_dim=114, num_classes=3): - super(FCN, self).__init__() - self.conv1 = nn.Sequential( - nn.Conv1d(input_dim, 128, kernel_size=8, padding=4), - nn.BatchNorm1d(128), - nn.ReLU() - ) - self.conv2 = nn.Sequential( - nn.Conv1d(128, 256, kernel_size=5, padding=2), - nn.BatchNorm1d(256), - nn.ReLU() - ) - self.conv3 = nn.Sequential( - nn.Conv1d(256, 128, kernel_size=3, padding=1), - nn.BatchNorm1d(128), - nn.ReLU() - ) - self.gap = nn.AdaptiveAvgPool1d(1) - self.fc = nn.Sequential( - nn.Dropout(0.5), - nn.Linear(128, num_classes) - ) - - def forward(self, x): - x = self.conv1(x) - x = self.conv2(x) - x = self.conv3(x) - x = self.gap(x) - x = x.squeeze(-1) - return self.fc(x) - -# 2. 1D Convolutional Neural Network (CNN) -class Advanced1DCNN(nn.Module): - def __init__(self, n_subcarriers=114, num_classes=3): - super(Advanced1DCNN, self).__init__() - self.conv1 = nn.Sequential( - nn.Conv1d(n_subcarriers, 64, kernel_size=5, padding=2), - nn.BatchNorm1d(64), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.MaxPool1d(2) - ) - self.conv2 = nn.Sequential( - nn.Conv1d(64, 128, kernel_size=5, padding=2), - nn.BatchNorm1d(128), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.MaxPool1d(2) - ) - self.conv3 = nn.Sequential( - nn.Conv1d(128, 256, kernel_size=5, padding=2), - nn.BatchNorm1d(256), - nn.ReLU(), - nn.Dropout1d(0.2), - nn.AdaptiveAvgPool1d(4) - ) - self.classifier = nn.Sequential( - nn.Flatten(), - nn.Linear(256 * 4, 128), - nn.ReLU(), - nn.Dropout(0.5), - nn.Linear(128, num_classes) - ) - def forward(self, x): - x = self.conv1(x) - x = self.conv2(x) - x = self.conv3(x) - x = self.classifier(x) - return x - -# 3. Multi-Layer Perceptron (MLP) -class SimpleMLP(nn.Module): - def __init__(self, input_dim=5700, num_classes=3): # 114 * 50 = 5700 - super(SimpleMLP, self).__init__() - self.net = nn.Sequential( - nn.Flatten(), - nn.Linear(input_dim, 256), - nn.ReLU(), - nn.Dropout(0.3), - nn.Linear(256, 128), - nn.ReLU(), - nn.Dropout(0.3), - nn.Linear(128, num_classes) - ) - def forward(self, x): - return self.net(x) - -# 4. Long Short-Term Memory (LSTM) -class LSTMGestureClassifier(nn.Module): - def __init__(self, input_dim=114, hidden_dim=128, num_layers=2, num_classes=3): - super(LSTMGestureClassifier, self).__init__() - self.lstm = nn.LSTM( - input_size=input_dim, - hidden_size=hidden_dim, - num_layers=num_layers, - batch_first=True, - dropout=0.3, - bidirectional=True - ) - self.classifier = nn.Sequential( - nn.Linear(hidden_dim * 2, 64), - nn.ReLU(), - nn.Dropout(0.3), - nn.Linear(64, num_classes) - ) - def forward(self, x): - x = x.permute(0, 2, 1) # -> (batch, 50, 114) - lstm_out, (hidden, cell) = self.lstm(x) - last_hidden = torch.cat((hidden[-2], hidden[-1]), dim=1) # Bidirectional last hidden states - return self.classifier(last_hidden) - -# 5. Transformer Encoder -class PositionalEncoding(nn.Module): - def __init__(self, d_model, max_len=500): - super().__init__() - pe = torch.zeros(max_len, d_model) - position = torch.arange(0, max_len).unsqueeze(1).float() - div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(np.log(10000.0) / d_model)) - pe[:, 0::2] = torch.sin(position * div_term) - pe[:, 1::2] = torch.cos(position * div_term) - pe = pe.unsqueeze(0) - self.register_buffer('pe', pe) - - def forward(self, x): - return x + self.pe[:, :x.size(1)] - -class CSITransformer(nn.Module): - def __init__(self, input_dim=114, d_model=128, nhead=4, num_layers=2, num_classes=3): - super(CSITransformer, self).__init__() - self.input_proj = nn.Linear(input_dim, d_model) - self.pos_encoder = PositionalEncoding(d_model) - encoder_layer = nn.TransformerEncoderLayer( - d_model=d_model, - nhead=nhead, - dim_feedforward=256, - dropout=0.1, - batch_first=True - ) - self.transformer = nn.TransformerEncoder(encoder_layer, num_layers) - self.classifier = nn.Sequential( - nn.Linear(d_model, 64), - nn.ReLU(), - nn.Dropout(0.3), - nn.Linear(64, num_classes) - ) - - def forward(self, x): - x = x.permute(0, 2, 1) # -> (batch, 50, 114) - x = self.input_proj(x) - x = self.pos_encoder(x) - x = self.transformer(x) - x = x.mean(dim=1) # Global Average Pooling over temporal axis - return self.classifier(x) - -# ----------------- Training Pipeline ----------------- -def train_model(model_name, model, train_loader, test_loader, epochs, device): - print(f"\n--- Training {model_name} ---") - criterion = nn.CrossEntropyLoss(label_smoothing=0.1) - optimizer = optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-2) - scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs) - - best_test_acc = 0.0 - history = {'train_loss': [], 'train_acc': [], 'test_acc': []} - best_weights = None - - for epoch in range(epochs): - model.train() - correct, total, loss_val = 0, 0, 0.0 - for inputs, labels in train_loader: - inputs, labels = inputs.to(device), labels.to(device) - optimizer.zero_grad() - outputs = model(inputs) - loss = criterion(outputs, labels) - loss.backward() - optimizer.step() - loss_val += loss.item() - _, predicted = torch.max(outputs.data, 1) - total += labels.size(0) - correct += (predicted == labels).sum().item() - - train_acc = 100 * correct / total - scheduler.step() - - # Test evaluation - model.eval() - test_correct, test_total = 0, 0 - with torch.no_grad(): - for inputs, labels in test_loader: - inputs, labels = inputs.to(device), labels.to(device) - outputs = model(inputs) - _, predicted = torch.max(outputs.data, 1) - test_total += labels.size(0) - test_correct += (predicted == labels).sum().item() - - test_acc = 100 * test_correct / test_total - avg_loss = loss_val / len(train_loader) - - history['train_loss'].append(avg_loss) - history['train_acc'].append(train_acc) - history['test_acc'].append(test_acc) - - if test_acc > best_test_acc: - best_test_acc = test_acc - best_weights = model.state_dict().copy() - - print(f"Epoch {epoch+1:02d}/{epochs:02d}: Loss={avg_loss:.4f}, Train Acc={train_acc:.2f}%, Test Acc={test_acc:.2f}%") - - print(f"Finished {model_name}! Best Test Accuracy: {best_test_acc:.2f}%") - return best_test_acc, best_weights, history - -# ----------------- Main Execution ----------------- -def main(): - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - print(f"Using device: {device}") - script_dir = os.path.dirname(os.path.abspath(__file__)) - - # 1. Load and Split Dataset (strictly sequentially 50/50, root directory only) - X_by_label = load_dataset() - x_train_raw, x_test_raw, y_train, y_test = split_50_50(X_by_label) - - # 2. Preprocess (SR-Std) - print("\nApplying Subcarrier-wise Regularized Standardization (SR-Std)...") - x_train = preprocess_sr_std(x_train_raw, eps=2.0) - x_test = preprocess_sr_std(x_test_raw, eps=2.0) - - # 3. Convert to Tensors and reshape to (batch, subcarriers, time) - x_train_tensor = torch.tensor(x_train, dtype=torch.float32).permute(0, 2, 1) - x_test_tensor = torch.tensor(x_test, dtype=torch.float32).permute(0, 2, 1) - - train_dataset = TensorDataset(x_train_tensor, torch.tensor(y_train, dtype=torch.long)) - test_dataset = TensorDataset(x_test_tensor, torch.tensor(y_test, dtype=torch.long)) - - train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) - test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False) - - # Define models to train - models_to_train = { - 'FCN': (FCN(input_dim=114, num_classes=3), 'best_fcn.pth'), - 'CNN1D': (Advanced1DCNN(n_subcarriers=114, num_classes=3), 'best_cnn1d.pth'), - 'MLP': (SimpleMLP(input_dim=5700, num_classes=3), 'best_mlp.pth'), - 'LSTM': (LSTMGestureClassifier(input_dim=114, hidden_dim=128, num_layers=2, num_classes=3), 'best_lstm.pth'), - 'Transformer': (CSITransformer(input_dim=114, d_model=128, nhead=4, num_layers=2, num_classes=3), 'best_transformer.pth') - } - - epochs = 30 - all_histories = {} - best_accuracies = {} - - # Train each model - for name, (model, weight_file) in models_to_train.items(): - model = model.to(device) - best_acc, weights, history = train_model(name, model, train_loader, test_loader, epochs, device) - best_accuracies[name] = best_acc - all_histories[name] = history - - # Save weights - torch.save(weights, os.path.join(script_dir, weight_file)) - print(f"Saved {name} weights to {weight_file}") - - # Save individual model plot - plt.figure(figsize=(10, 4)) - plt.subplot(1, 2, 1) - plt.plot(history['train_loss'], label='Train Loss') - plt.title(f'{name} Training Loss') - plt.xlabel('Epoch') - plt.ylabel('Loss') - plt.grid(True) - plt.legend() - - plt.subplot(1, 2, 2) - plt.plot(history['train_acc'], label='Train Acc') - plt.plot(history['test_acc'], label='Test Acc') - plt.title(f'{name} Accuracy History') - plt.xlabel('Epoch') - plt.ylabel('Accuracy (%)') - plt.grid(True) - plt.legend() - - plt.tight_layout() - plot_filename = f"{name.lower()}_history.png" - plt.savefig(os.path.join(script_dir, plot_filename)) - plt.close() - print(f"Saved {name} curves to {plot_filename}") - - # Final per-class evaluation for the best model state - model.load_state_dict(weights) - model.eval() - all_preds = [] - all_labels = [] - with torch.no_grad(): - for inputs, labels in test_loader: - inputs = inputs.to(device) - outputs = model(inputs) - _, predicted = torch.max(outputs.data, 1) - all_preds.extend(predicted.cpu().numpy()) - all_labels.extend(labels.numpy()) - - all_preds = np.array(all_preds) - all_labels = np.array(all_labels) - - print(f"\n=== Per-Class Accuracy Evaluation for {name} ===") - for class_name, class_idx in LABEL_MAP.items(): - indices = np.where(all_labels == class_idx)[0] - class_correct = np.sum(all_preds[indices] == class_idx) - class_acc = 100 * class_correct / len(indices) - print(f" Class '{class_name}': Acc={class_acc:.2f}% ({class_correct}/{len(indices)})") - - # 4. Generate Joint Comparison Plot - plt.figure(figsize=(10, 6)) - for name, history in all_histories.items(): - plt.plot(range(1, epochs + 1), history['test_acc'], label=f"{name} (Best: {best_accuracies[name]:.2f}%)") - plt.title('Test Accuracy Comparison across Architectures (SR-Std + 50/50 Split)') - plt.xlabel('Epoch') - plt.ylabel('Test Accuracy (%)') - plt.grid(True) - plt.legend(loc='lower right') - plt.tight_layout() - comparison_filename = "model_comparison.png" - plt.savefig(os.path.join(script_dir, comparison_filename)) - plt.close() - print(f"\nSaved overall comparison plot to {comparison_filename}") - - # 5. Print final summary table - print("\n================ Final Performance Summary ================") - print(f"{'Model':15s} | {'Best Test Accuracy (%)':22s}") - print("-" * 43) - for name, acc in best_accuracies.items(): - print(f"{name:15s} | {acc:22.2f}%") - -if __name__ == '__main__': - main() diff --git a/model/muti-model/training_history.png b/model/muti-model/training_history.png deleted file mode 100644 index 46205ce..0000000 Binary files a/model/muti-model/training_history.png and /dev/null differ diff --git a/model/muti-model/transformer_history.png b/model/muti-model/transformer_history.png deleted file mode 100644 index 04e4cb3..0000000 Binary files a/model/muti-model/transformer_history.png and /dev/null differ diff --git a/model/new_model.py b/model/new_model.py deleted file mode 100644 index d36e861..0000000 --- a/model/new_model.py +++ /dev/null @@ -1,326 +0,0 @@ -from numpy import dtype -from torch import nn -from torch.utils.data import Dataset, DataLoader -from tqdm import tqdm -import matplotlib.pyplot as plt - -import numpy as np -import torch - -device = 'cuda' if torch.cuda.is_available() else 'cpu' - - - -def complex_to_real(frame): - return np.concatenate(( - np.abs(frame).astype(np.float32), # 振幅 - np.angle(frame).astype(np.float32)), axis=0) # 相位 - - -class PositionalEncoding(nn.Module): - def __init__(self, d_model, max_len=5000): - super().__init__() - pe = torch.zeros(max_len, d_model) - position = torch.arange(0, max_len).unsqueeze(1).float() - div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(np.log(10000.0) / d_model)) - pe[:, 0::2] = torch.sin(position * div_term) - pe[:, 1::2] = torch.cos(position * div_term) - pe = pe.unsqueeze(0) - self.register_buffer('pe', pe) - - def forward(self, x): - # x shape: (batch, seq_len, d_model) - return x + self.pe[:, :x.size(1)] - -''' -class CSIModel(nn.Module): - def __init__(self, input_dim=60, d_model=128, n_head=8, num_layers=4): - super().__init__() - self.avg_pool = nn.AdaptiveAvgPool1d(4) - self.classifier = nn.Sequential( - nn.Linear(input_dim * 16, input_dim * 4), - nn.Dropout(0.3), - nn.ReLU(), - nn.Linear(input_dim * 4, input_dim * 2), - nn.Dropout(0.2), - nn.ReLU(), - nn.Linear(input_dim * 2, 3), - ) - self.classifier = nn.Linear(input_dim*4, 3) - - def forward(self, x, mask=None): - x = x.permute(0, 2, 1) - x = self.avg_pool(x) - x = x.reshape(x.size(0), -1) - if mask is not None: pass - - return self.classifier(x) -''' -class CSIModel(nn.Module): - def __init__(self, input_dim=60, d_model=128, n_head=8, num_layers=4): - super().__init__() - self.input_proj = nn.Linear(input_dim, d_model) - self.pos_encoder = PositionalEncoding(d_model) - - encoder_layer = nn.TransformerEncoderLayer( - d_model=d_model, nhead=n_head, dim_feedforward=512, - dropout=0.1, batch_first=True - ) - self.transformer = nn.TransformerEncoder(encoder_layer, num_layers) - self.classifier = nn.Linear(d_model, 3) - - def forward(self, x, mask=None): - # x: (batch, seq_len, input_dim) - x = self.input_proj(x) # (batch, seq_len, d_model) - - # 添加位置编码 - x = self.pos_encoder(x) # (batch, seq_len, d_model) - - # Transformer 编码 (使用 batch_first=True) - x = self.transformer(x, src_key_padding_mask=mask) # (batch, seq_len, d_model) - - # 池化（考虑 mask） - if mask is not None: - # mask: (batch, seq_len), True 表示需要 mask 的位置 - valid_mask = (~mask).float().unsqueeze(-1) # (batch, seq_len, 1) - x = (x * valid_mask).sum(dim=1) / (valid_mask.sum(dim=1) + 1e-8) # 避免除零 - else: - x = x.mean(dim=1) - - return self.classifier(x) - - -class CSIDataset(Dataset): - def __init__(self, data1, data2, data3, - clean, # 滤波数据去掉这行 - is_train=True, train_ratio=0.65, seed=42): - self.keep_indices = np.where(np.array(clean) == False)[0] # 滤波数据去掉这行 - - data_list = [] - labels = [] - - for label_idx, data in enumerate([data1, data2, data3]): - # Shuffle the data before splitting - if seed is not None: - np.random.seed(seed + label_idx) # Different seed for each class - indices = np.random.permutation(len(data)) - shuffled_data = data[indices] - - # Split after shuffling - n = len(shuffled_data) - split_idx = int(n * train_ratio) - split_data = shuffled_data[:split_idx] if is_train else shuffled_data[split_idx:] - data_list.append(split_data) - labels.append(np.full(len(split_data), label_idx, dtype=int)) - - self.is_train = is_train - self.data = np.concatenate(data_list, axis=0) - self.labels = np.concatenate(labels) - - def __len__(self): - return len(self.data) - - def __getitem__(self, idx): - data = self.data[idx][:, self.keep_indices] - - data = np.concatenate([ - np.abs(data).astype(np.float32), - np.angle(data).astype(np.float32) - ], axis=1) - data = (data - data.mean()) / (data.std() + 1e-8) - - return torch.from_numpy(data), self.labels[idx] - -def collate_fn(batch): - sequences, labels = zip(*batch) - lengths = torch.tensor([len(seq) for seq in sequences]) - max_len = lengths.max() - - padded = [] - for seq in sequences: - # 转换为 tensor - seq_tensor = torch.from_numpy(seq) if isinstance(seq, np.ndarray) else seq - # 填充 - if len(seq_tensor) < max_len: - padding = torch.zeros(max_len - len(seq_tensor), seq_tensor.shape[1]) - seq_tensor = torch.cat([seq_tensor, padding]) - padded.append(seq_tensor) - - padded = torch.stack(padded) - mask = torch.arange(max_len).unsqueeze(0) >= lengths.unsqueeze(1) - - return padded, torch.tensor(labels), mask - - -def train_epoch(model, loader, criterion, optimizer): - model.train() - total_loss, correct, total = 0, 0, 0 - - pbar = tqdm(loader, desc='Training') - for data, labels, mask in pbar: - data, labels, mask = data.to(device), labels.to(device), mask.to(device) - - optimizer.zero_grad() - output = model(data, mask) - loss = criterion(output, labels) - loss.backward() - - # 计算梯度统计信息 - grad_norm = 0.0 - max_grad = 0.0 - for p in model.parameters(): - if p.grad is not None: - grad_norm += p.grad.norm().item() ** 2 - max_grad = max(max_grad, p.grad.abs().max().item()) - grad_norm = grad_norm ** 0.5 - - optimizer.step() - - total_loss += loss.item() - correct += (output.argmax(1) == labels).sum().item() - total += len(labels) - - # 实时更新进度条显示 - current_loss = total_loss / (pbar.n + 1) - current_acc = correct / total - pbar.set_postfix({ - 'loss': f'{current_loss:.4f}', - 'acc': f'{current_acc:.4f}', - 'grad_norm': f'{grad_norm:.2e}', - 'max_grad': f'{max_grad:.2e}' - }) - - return total_loss / len(loader), correct / total - - -def validate(model, loader, criterion): - model.eval() - total_loss, correct, total = 0, 0, 0 - - with torch.no_grad(): - pbar = tqdm(loader, desc='Validating') - for data, labels, mask in pbar: - data, labels, mask = data.to(device), labels.to(device), mask.to(device) - output = model(data, mask) - loss = criterion(output, labels) - - total_loss += loss.item() - correct += (output.argmax(1) == labels).sum().item() - total += len(labels) - - # 实时更新进度条显示 - current_loss = total_loss / (pbar.n + 1) - current_acc = correct / total - pbar.set_postfix({ - 'loss': f'{current_loss:.4f}', - 'acc': f'{current_acc:.4f}' - }) - - return total_loss / len(loader), correct / total - - -def plot_history(train_loss, train_acc, val_loss, val_acc): - plt.figure(figsize=(12, 4)) - - plt.subplot(1, 2, 1) - plt.plot(train_loss, label='Train Loss') - plt.plot(val_loss, label='Val Loss') - plt.xlabel('Epoch') - plt.ylabel('Loss') - plt.legend() - plt.grid(True) - - plt.subplot(1, 2, 2) - plt.plot(train_acc, label='Train Acc') - plt.plot(val_acc, label='Val Acc') - plt.xlabel('Epoch') - plt.ylabel('Accuracy') - plt.legend() - plt.grid(True) - - plt.tight_layout() - plt.show() - -def train(): - # 数据加载 - # 滤波数据 - train_set = CSIDataset(data1, data2, data3, clean, is_train=True) - val_set = CSIDataset(data1, data2, data3, clean, is_train=False) - # 不滤波数据 - # train_set = CSIDataset(data1, data2, data3, is_train=True) - # val_set = CSIDataset(data1, data2, data3, is_train=False) - - train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True, collate_fn=collate_fn, num_workers=4) - val_loader = DataLoader(val_set, batch_size=batch_size, shuffle=False, collate_fn=collate_fn, num_workers=4) - - criterion = nn.CrossEntropyLoss() - optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01) - scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50) - - # 训练 - train_losses, train_accs = [], [] - val_losses, val_accs = [], [] - best_acc = 0 - - for epoch in range(1, epochs + 1): - # print(f'\nEpoch {epoch}/{epochs}') - - train_loss, train_acc = train_epoch(model, train_loader, criterion, optimizer) - val_loss, val_acc = validate(model, val_loader, criterion) - - train_losses.append(train_loss) - train_accs.append(train_acc) - val_losses.append(val_loss) - val_accs.append(val_acc) - - scheduler.step() - - if val_acc > best_acc: - best_acc = val_acc - torch.save(model.state_dict(), 'models/best_model.pth') - print(f'✓ Saved best model (acc: {val_acc:.4f})') - else: - torch.save(model.state_dict(), f'models/epoch{epoch}.pth') - print(f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}') - print(f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}') - - plot_history(train_losses, train_accs, val_losses, val_accs) - print(f'\nBest validation accuracy: {best_acc:.4f}') - - -if __name__ == '__main__': - ''' - # 滤波数据 - amp1 = np.load("dataset4/amp/fist_data.npz", allow_pickle=True)['data'] - pha1 = np.load("dataset4/pha/fist_data.npz", allow_pickle=True)['data'] - amp2 = np.load("dataset4/amp/paper_data.npz", allow_pickle=True)['data'] - pha2 = np.load("dataset4/pha/paper_data.npz", allow_pickle=True)['data'] - amp3 = np.load("dataset4/amp/scissor_data.npz", allow_pickle=True)['data'] - pha3 = np.load("dataset4/pha/scissor_data.npz", allow_pickle=True)['data'] - data1 = np.concatenate((amp1, pha1), axis=2, dtype=np.float32) - data2 = np.concatenate((amp2, pha2), axis=2, dtype=np.float32) - data3 = np.concatenate((amp3, pha3), axis=2, dtype=np.float32) - ''' - # 不滤波数据 - #data1 = np.load("dataset3/fist_complex_data.npz", allow_pickle=True)['data'] - #data2 = np.load("dataset3/paper_complex_data.npz", allow_pickle=True)['data'] - #data3 = np.load("dataset3/scissor_complex_data.npz", allow_pickle=True)['data'] - # 滤波数据去掉这行 - #clean = [True, True, True, True, True, True, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, True, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, True, True, True, True, True, True, True, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, True, True, True, True, True, True, True, True, True, True, True, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, True, ] - - - lr = 1e-5 - batch_size = 32 - epochs = 20 - # 模型 - model = CSIModel(input_dim=332, d_model=128, n_head=4, num_layers=2).to(device) - # 读训练好的模型，想微调或者预测把这行注释去了 - model.load_state_dict(torch.load('models/no_filter_best_model.pth', map_location=device)) - print(device) - - # 如果想训练 - # train() - - # 假如采集到新的数据 - model.eval() # 开启推理模式，模型不会学习 - result = model( ) # result就是预测向量，想转成概率向量就过一遍softmax，取argmax就是预测结果 diff --git a/model/pyproject.toml b/model/pyproject.toml index dc0089b..bd1807e 100644 --- a/model/pyproject.toml +++ b/model/pyproject.toml @@ -6,6 +6,10 @@ readme = "README.md" requires-python = ">=3.12" dependencies = [ "matplotlib>=3.10.9", + "numpy>=2.4.4", "scikit-learn>=1.8.0", + "seaborn>=0.13.2", + "torch>=2.12.0", + "torchvision>=0.27.0", "tqdm>=4.67.3", ] diff --git a/model/train_50_50.py b/model/train_50_50.py deleted file mode 100644 index 06d5efe..0000000 --- a/model/train_50_50.py +++ /dev/null @@ -1,232 +0,0 @@ -import os -import sys -import numpy as np -import torch -import torch.nn as nn -import torch.optim as optim -from torch.utils.data import DataLoader, TensorDataset -import matplotlib.pyplot as plt - -# Set random seed for reproducibility -torch.manual_seed(42) -np.random.seed(42) - -LABEL_MAP = {'draw': 0, 'stand-up': 1, 'wave': 2} -INV_LABEL_MAP = {v: k for k, v in LABEL_MAP.items()} - -def load_dataset(dataset_dir="dataset/dataset_2026_6_23"): - if not os.path.exists(dataset_dir): - raise FileNotFoundError(f"Dataset directory '{dataset_dir}' not found.") - npz_files = sorted([f for f in os.listdir(dataset_dir) if f.endswith('.npz')]) - X_dict = {0: [], 1: [], 2: []} - - print(f"Loading dataset from {dataset_dir}...") - for filename in sorted(npz_files): - file_path = os.path.join(dataset_dir, filename) - label_name = filename.split('_')[0] - if label_name not in LABEL_MAP: - continue - label_idx = LABEL_MAP[label_name] - data = np.load(file_path, allow_pickle=True)['dataset'] - X_dict[label_idx].append(data) - - X_by_label = {} - for idx in X_dict: - X_by_label[idx] = np.concatenate(X_dict[idx], axis=0) - print(f" Class '{INV_LABEL_MAP[idx]}': {X_by_label[idx].shape[0]} samples") - - return X_by_label - -def split_50_50(X_by_label): - x_train_list, x_test_list = [], [] - y_train_list, y_test_list = [], [] - - print("\nSplitting dataset (50% train, 50% test sequentially)...") - for idx in sorted(X_by_label.keys()): - X = X_by_label[idx] - n_samples = len(X) - split_point = int(n_samples * 0.5) - - x_train_list.append(X[:split_point]) - x_test_list.append(X[split_point:]) - y_train_list.append(np.full(split_point, idx, dtype=np.int64)) - y_test_list.append(np.full(n_samples - split_point, idx, dtype=np.int64)) - print(f" Class '{INV_LABEL_MAP[idx]}': Train={split_point}, Test={n_samples - split_point}") - - x_train = np.concatenate(x_train_list, axis=0) - x_test = np.concatenate(x_test_list, axis=0) - y_train = np.concatenate(y_train_list, axis=0) - y_test = np.concatenate(y_test_list, axis=0) - - return x_train, x_test, y_train, y_test - -def preprocess_sr_std(X, eps=2.0): - # Subcarrier-wise Regularized Standardization (SR-Std) - X_norm = np.zeros_like(X) - for i in range(len(X)): - mean = X[i].mean(axis=0, keepdims=True) - std = X[i].std(axis=0, keepdims=True) - X_norm[i] = (X[i] - mean) / (std + eps) - return X_norm - -class FCN(nn.Module): - def __init__(self, input_dim=114, num_classes=3): - super(FCN, self).__init__() - self.conv1 = nn.Sequential( - nn.Conv1d(input_dim, 128, kernel_size=8, padding=4), - nn.BatchNorm1d(128), - nn.ReLU() - ) - self.conv2 = nn.Sequential( - nn.Conv1d(128, 256, kernel_size=5, padding=2), - nn.BatchNorm1d(256), - nn.ReLU() - ) - self.conv3 = nn.Sequential( - nn.Conv1d(256, 128, kernel_size=3, padding=1), - nn.BatchNorm1d(128), - nn.ReLU() - ) - self.gap = nn.AdaptiveAvgPool1d(1) - self.fc = nn.Sequential( - nn.Dropout(0.5), - nn.Linear(128, num_classes) - ) - - def forward(self, x): - # input x shape: (batch, input_dim, time) - x = self.conv1(x) - x = self.conv2(x) - x = self.conv3(x) - x = self.gap(x) - x = x.squeeze(-1) - return self.fc(x) - -def main(): - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - print(f"Using device: {device}") - script_dir = os.path.dirname(os.path.abspath(__file__)) - - # Load raw dataset - X_by_label = load_dataset() - x_train_raw, x_test_raw, y_train, y_test = split_50_50(X_by_label) - - # Apply Preprocessing (SR-Std) - print("\nApplying Subcarrier-wise Regularized Standardization (SR-Std)...") - x_train = preprocess_sr_std(x_train_raw, eps=2.0) - x_test = preprocess_sr_std(x_test_raw, eps=2.0) - - # Convert to Tensor and permute to (batch, subcarriers, time) for 1D convolution - x_train_tensor = torch.tensor(x_train, dtype=torch.float32).permute(0, 2, 1) - x_test_tensor = torch.tensor(x_test, dtype=torch.float32).permute(0, 2, 1) - - train_dataset = TensorDataset(x_train_tensor, torch.tensor(y_train, dtype=torch.long)) - test_dataset = TensorDataset(x_test_tensor, torch.tensor(y_test, dtype=torch.long)) - - train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) - test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False) - - model = FCN(input_dim=114, num_classes=3).to(device) - criterion = nn.CrossEntropyLoss(label_smoothing=0.1) - optimizer = optim.AdamW(model.parameters(), lr=1e-5, weight_decay=1e-2) - scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30) - - epochs = 30 - history = {'train_loss': [], 'train_acc': [], 'test_acc': []} - best_test_acc = 0.0 - - print("\n=== Starting Training (FCN + SR-Std + 50/50 Sequential Split) ===") - for epoch in range(epochs): - model.train() - correct, total, loss_val = 0, 0, 0.0 - for inputs, labels in train_loader: - inputs, labels = inputs.to(device), labels.to(device) - optimizer.zero_grad() - outputs = model(inputs) - loss = criterion(outputs, labels) - loss.backward() - optimizer.step() - loss_val += loss.item() - _, predicted = torch.max(outputs.data, 1) - total += labels.size(0) - correct += (predicted == labels).sum().item() - - train_acc = 100 * correct / total - scheduler.step() - - # Test evaluation - model.eval() - test_correct, test_total = 0, 0 - with torch.no_grad(): - for inputs, labels in test_loader: - inputs, labels = inputs.to(device), labels.to(device) - outputs = model(inputs) - _, predicted = torch.max(outputs.data, 1) - test_total += labels.size(0) - test_correct += (predicted == labels).sum().item() - - test_acc = 100 * test_correct / test_total - avg_loss = loss_val / len(train_loader) - - history['train_loss'].append(avg_loss) - history['train_acc'].append(train_acc) - history['test_acc'].append(test_acc) - - if test_acc > best_test_acc: - best_test_acc = test_acc - torch.save(model.state_dict(), os.path.join(script_dir, "best_fcn.pth")) - - print(f"Epoch {epoch+1:02d}/{epochs:02d}: Loss={avg_loss:.4f}, Train Acc={train_acc:.2f}%, Test Acc={test_acc:.2f}%") - - print(f"\nTraining Complete! Best Test Accuracy: {best_test_acc:.2f}%") - - # Save training history plot - plt.figure(figsize=(12, 5)) - plt.subplot(1, 2, 1) - plt.plot(history['train_loss'], label='Train Loss') - plt.title('Training Loss') - plt.xlabel('Epoch') - plt.ylabel('Loss') - plt.grid(True) - plt.legend() - - plt.subplot(1, 2, 2) - plt.plot(history['train_acc'], label='Train Accuracy') - plt.plot(history['test_acc'], label='Test Accuracy') - plt.title('Accuracy History') - plt.xlabel('Epoch') - plt.ylabel('Accuracy (%)') - plt.grid(True) - plt.legend() - - plt.tight_layout() - plt.savefig(os.path.join(script_dir, "fcn_history.png")) - print(f"Saved training curves to '{os.path.join(script_dir, 'fcn_history.png')}'") - - # Load best model for final evaluation - model.load_state_dict(torch.load(os.path.join(script_dir, "best_fcn.pth"), map_location=device)) - model.eval() - - all_preds = [] - all_labels = [] - with torch.no_grad(): - for inputs, labels in test_loader: - inputs = inputs.to(device) - outputs = model(inputs) - _, predicted = torch.max(outputs.data, 1) - all_preds.extend(predicted.cpu().numpy()) - all_labels.extend(labels.numpy()) - - all_preds = np.array(all_preds) - all_labels = np.array(all_labels) - - # Per-class accuracy - print("\n=== Per-Class Accuracy Evaluation ===") - for class_name, class_idx in LABEL_MAP.items(): - indices = np.where(all_labels == class_idx)[0] - class_correct = np.sum(all_preds[indices] == class_idx) - class_acc = 100 * class_correct / len(indices) - print(f" Class '{class_name}': Acc={class_acc:.2f}% ({class_correct}/{len(indices)})") - -if __name__ == '__main__': - main() diff --git a/src/firmware-code/p4_remote_wifi/transform_tool/quantize_cnn.py b/src/firmware-code/p4_remote_wifi/transform_tool/quantize_cnn.py new file mode 100644 index 0000000..484741e --- /dev/null +++ b/src/firmware-code/p4_remote_wifi/transform_tool/quantize_cnn.py @@ -0,0 +1,107 @@ +import os +import sys +import numpy as np +import torch +import torch.nn as nn + +# Add train_src to system path for imports +sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "train_src")) +from models import Advanced1DCNN +from dataset import load_dataset, preprocess_csi_fusion + +from esp_ppq.api import espdl_quantize_onnx + +def main(): + # Configure console to output UTF-8 to prevent encoding crashes on Windows with emojis + try: + sys.stdout.reconfigure(encoding='utf-8') + sys.stderr.reconfigure(encoding='utf-8') + except AttributeError: + pass # Fallback for older python where reconfigure is not available + + transform_tool_dir = os.path.dirname(os.path.abspath(__file__)) + model_path = os.path.join(transform_tool_dir, "models", "best_optimized_cnn.pth") + onnx_path = os.path.join(transform_tool_dir, "cnn_model.onnx") + espdl_path = os.path.join(transform_tool_dir, "cnn_model.espdl") + dataset_dir = os.path.join(transform_tool_dir, "dataset", "dataset_2026_6_10") + + print(f"Loading weights from {model_path}...") + state_dict = torch.load(model_path, map_location="cpu") + + # Instantiate Advanced1DCNN + model = Advanced1DCNN(n_subcarriers=228, num_classes=4) + model.load_state_dict(state_dict) + model.eval() + + # 1. Export to ONNX + dummy_input = torch.randn(1, 228, 50) + print(f"Exporting model to ONNX at {onnx_path}...") + torch.onnx.export( + model, + dummy_input, + onnx_path, + export_params=True, + opset_version=13, + do_constant_folding=True, + input_names=['input'], + output_names=['output'], + dynamic_axes={'input': {0: 'batch_size'}, 'output': {0: 'batch_size'}} + ) + print("ONNX model exported successfully.") + + # 2. Load calibration dataset + print(f"Loading calibration dataset from {dataset_dir}...") + X_by_label = load_dataset(dataset_dir) + + # Accumulate samples for calibration + calib_samples = [] + # Collect up to 32 samples spread across classes + samples_per_class = 12 + for label_idx, data_complex in X_by_label.items(): + # data_complex shape: (N, 50, 114) + N = len(data_complex) + select_count = min(N, samples_per_class) + indices = np.linspace(0, N - 1, select_count, dtype=int) + class_samples = data_complex[indices] + + # Preprocess using csi fusion + preprocessed = preprocess_csi_fusion(class_samples) # (select_count, 50, 228) + + for sample in preprocessed: + # sample shape: (50, 228) + # We need shape (1, 228, 50) as PyTorch tensor + t_sample = torch.tensor(sample, dtype=torch.float32).unsqueeze(0).permute(0, 2, 1) + calib_samples.append(t_sample) + + print(f"Total calibration samples collected: {len(calib_samples)}") + + # Truncate or pad to exactly 32 samples if needed + if len(calib_samples) < 32: + # Pad with duplicate samples + while len(calib_samples) < 32: + calib_samples.append(calib_samples[0]) + calib_samples = calib_samples[:32] + + def collate_fn(batch): + return batch.to("cpu") + + # 3. Quantize using esp-ppq + print(f"Running esp-ppq quantization to generate {espdl_path}...") + espdl_quantize_onnx( + onnx_import_file=onnx_path, + espdl_export_file=espdl_path, + calib_dataloader=calib_samples, + calib_steps=32, + input_shape=[1, 228, 50], + target="esp32p4", + num_of_bits=8, + collate_fn=collate_fn, + device="cpu", + error_report=True, + export_test_values=True, + verbose=1 + ) + print("esp-ppq Quantization completed successfully.") + +if __name__ == '__main__': + main() diff --git a/src/firmware-code/p4_remote_wifi/transform_tool/send_dataset_sample.py b/src/firmware-code/p4_remote_wifi/transform_tool/send_dataset_sample.py new file mode 100644 index 0000000..b7bccc3 --- /dev/null +++ b/src/firmware-code/p4_remote_wifi/transform_tool/send_dataset_sample.py @@ -0,0 +1,313 @@ +import os +import sys +import struct +import time +import re +import random +import numpy as np +import serial +import serial.tools.list_ports + +# Add train_src to system path for imports +sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "train_src")) +from dataset import load_dataset + +# Mappings for display +LABEL_MAP = {0: 'cut (挥手)', 1: 'grip (抓握)', 2: 'draw_o (画圈)'} + +def get_serial_ports(): + return serial.tools.list_ports.comports() + +def main(): + print("====================================================") + print(" ESP32-P4 1D-CNN Dataset Sample Sender ") + print("====================================================") + + # List available COM ports + ports = get_serial_ports() + if not ports: + print("No serial ports found! Please connect your ESP32-P4 board.") + sys.exit(1) + + print("Available Serial Ports:") + for idx, port in enumerate(ports): + print(f"[{idx}] {port.device} - {port.description}") + + while True: + try: + choice = input(f"\nSelect port [0-{len(ports)-1}]: ").strip() + idx = int(choice) + if 0 <= idx < len(ports): + port_name = ports[idx].device + break + else: + print("Invalid selection.") + except ValueError: + print("Please enter a valid number.") + + baud_rate = 921600 + print(f"\nOpening {port_name} at {baud_rate} baud...") + try: + ser = serial.Serial(port_name, baud_rate, timeout=1) + print("Port successfully opened!") + except Exception as e: + print(f"Error opening port: {e}") + sys.exit(1) + + # Load dataset + transform_tool_dir = os.path.dirname(os.path.abspath(__file__)) + dataset_dir = os.path.join(transform_tool_dir, "dataset", "dataset_2026_6_10") + print(f"\nLoading dataset from {dataset_dir}...") + try: + X_by_label = load_dataset(dataset_dir) + except Exception as e: + print(f"Error loading dataset: {e}") + ser.close() + sys.exit(1) + + while True: + print("\n====================================================") + print(" MAIN OPTIONS ") + print("====================================================") + print("[1] Send Single Sample (Manual Index)") + print("[2] Run Batch Test Sweep (Continuous dataset evaluation)") + print("[3] Exit") + + opt = input("Select an option [1-3]: ").strip() + if opt == '1': + print("\n-----------------------------") + print("Classes available in dataset:") + for k, v in LABEL_MAP.items(): + count = len(X_by_label[k]) if k in X_by_label else 0 + print(f" [{k}] {v} - Count: {count} samples") + + try: + class_choice = input("Select a class [0-2]: ").strip() + class_idx = int(class_choice) + if class_idx not in X_by_label: + print("Invalid class index.") + continue + + samples = X_by_label[class_idx] + max_idx = len(samples) - 1 + + sample_choice = input(f"Select sample index [0-{max_idx}] (default 0): ").strip() + if not sample_choice: + sample_idx = 0 + else: + sample_idx = int(sample_choice) + if not (0 <= sample_idx <= max_idx): + print(f"Index out of range! Must be between 0 and {max_idx}.") + continue + + print(f"\nProcessing sample {sample_idx} from class '{LABEL_MAP[class_idx]}'") + + # Select 1 complex sample: shape (50, 114) + complex_data = samples[sample_idx] # Shape: (50, 114) complex + + # Serialize raw complex data (interleave imaginary and real parts) + raw_floats = np.empty((50, 228), dtype=np.float32) + raw_floats[:, 0::2] = complex_data.imag + raw_floats[:, 1::2] = complex_data.real + flat_features = raw_floats.flatten() + + # Construct framing payload + header = bytes([0xAA, 0xBB, 0xCC, 0xDD]) + data_bytes = struct.pack('<11400f', *flat_features) + length_bytes = struct.pack(' 0: + try: + line = ser.readline().decode('utf-8', errors='ignore').strip() + if line: + print(f"[Board Log] {line}") + if "Inference Result:" in line or "Inference failed" in line: + break + except Exception as read_err: + print(f"Error reading logs: {read_err}") + break + time.sleep(0.01) + + except ValueError: + print("Please enter valid numbers.") + except Exception as e: + print(f"An error occurred: {e}") + + elif opt == '2': + print("\n----------------------------------------------------") + print("Batch Test Sweep Settings:") + print("Select class mode:") + print(" [0] Sweep 'cut (挥手)' only") + print(" [1] Sweep 'grip (抓握)' only") + print(" [2] Sweep 'draw_o (画圈)' only") + print(" [3] Sweep mixed samples (interleaved across all classes)") + + try: + mode_choice = input("Select mode [0-3]: ").strip() + if not mode_choice: + continue + mode_choice = int(mode_choice) + if mode_choice not in [0, 1, 2, 3]: + print("Invalid choice.") + continue + + count_input = input("How many samples to test? (default 20): ").strip() + count = int(count_input) if count_input else 20 + + delay_input = input("Delay between samples in seconds? (default 0.5): ").strip() + delay = float(delay_input) if delay_input else 0.5 + + # Build list of samples to test: (class_idx, sample_idx) + test_queue = [] + if mode_choice in [0, 1, 2]: + # Specific class + samples_avail = len(X_by_label[mode_choice]) + indices = random.sample(range(samples_avail), min(count, samples_avail)) + for idx in indices: + test_queue.append((mode_choice, idx)) + else: + # Mixed classes + for _ in range(count): + c_idx = random.choice([0, 1, 2]) + s_idx = random.randint(0, len(X_by_label[c_idx]) - 1) + test_queue.append((c_idx, s_idx)) + + print(f"\nStarting batch test of {len(test_queue)} samples...") + print("Press Ctrl+C to abort batch run.") + + results = [] # List of dicts: {gt, pred, success, latency} + + for step, (gt_class, sample_idx) in enumerate(test_queue): + print(f"\n[{step+1}/{len(test_queue)}] Sending {LABEL_MAP[gt_class]} sample #{sample_idx}...") + + # Select 1 complex sample: shape (50, 114) + complex_data = X_by_label[gt_class][sample_idx] # Shape: (50, 114) complex + + # Serialize raw complex data (interleave imaginary and real parts) + raw_floats = np.empty((50, 228), dtype=np.float32) + raw_floats[:, 0::2] = complex_data.imag + raw_floats[:, 1::2] = complex_data.real + flat_features = raw_floats.flatten() + + # Framing + header = bytes([0xAA, 0xBB, 0xCC, 0xDD]) + data_bytes = struct.pack('<11400f', *flat_features) + length_bytes = struct.pack(' 0: + try: + line = ser.readline().decode('utf-8', errors='ignore').strip() + if line: + print(f" [Board Log] {line}") + if "Inference Result:" in line: + success = True + # Parse prediction class + if "Wave/Cut" in line or "挥手" in line: + pred_class = 0 + elif "Grip" in line or "抓握" in line: + pred_class = 1 + elif "Circle/Draw_o" in line or "画圈" in line: + pred_class = 2 + elif "Unknown" in line or "未知" in line: + pred_class = 3 + + # Parse latency + time_match = re.search(r'Time taken:\s*(\d+)\s*us', line) + if time_match: + latency_us = int(time_match.group(1)) + break + elif "Inference failed" in line: + break + except Exception as read_err: + print(f" Error reading logs: {read_err}") + break + time.sleep(0.01) + + correct = (pred_class == gt_class) if success else False + results.append({ + 'gt': gt_class, + 'pred': pred_class, + 'success': success, + 'correct': correct, + 'latency': latency_us + }) + + if success: + status_str = "\033[92mCORRECT\033[0m" if correct else f"\033[91mWRONG\033[0m (Predicted: {LABEL_MAP.get(pred_class, 'unknown')})" + print(f" >> Outcome: {status_str} | Latency: {latency_us / 1000.0:.2f} ms") + else: + print(" >> Outcome: \033[91mTIMEOUT/FAILED\033[0m") + + time.sleep(delay) + + # Print batch summary report + print("\n" + "="*52) + print(" BATCH TEST SWEEP REPORT ") + print("" + "="*52) + total_sent = len(results) + successful = sum(1 for r in results if r['success']) + correct = sum(1 for r in results if r['correct']) + + accuracy = (correct / total_sent * 100.0) if total_sent > 0 else 0 + avg_latency = (sum(r['latency'] for r in results if r['success']) / successful / 1000.0) if successful > 0 else 0 + + print(f"Total Samples Sent: {total_sent}") + print(f"Successful Inferences: {successful}") + print(f"Correct Predictions: {correct}") + print(f"Overall Accuracy: \033[93m{accuracy:.2f}%\033[0m") + print(f"Average Inference Time: {avg_latency:.2f} ms") + print("-"*52) + + # Print per-class details + for c_idx, c_name in LABEL_MAP.items(): + class_results = [r for r in results if r['gt'] == c_idx] + class_total = len(class_results) + class_correct = sum(1 for r in class_results if r['correct']) + class_acc = (class_correct / class_total * 100.0) if class_total > 0 else 0 + if class_total > 0: + print(f"Class '{c_name}': Accuracy = {class_acc:.2f}% ({class_correct}/{class_total})") + print("="*52) + + except KeyboardInterrupt: + print("\nBatch test sweep aborted by user.") + except ValueError: + print("Invalid setting input.") + + elif opt == '3': + print("Closing port and exiting...") + break + else: + print("Invalid option.") + + ser.close() + +if __name__ == '__main__': + main() diff --git a/src/firmware-code/p4_remote_wifi/transform_tool/test_csi_fps.py b/src/firmware-code/p4_remote_wifi/transform_tool/test_csi_fps.py new file mode 100644 index 0000000..7742826 --- /dev/null +++ b/src/firmware-code/p4_remote_wifi/transform_tool/test_csi_fps.py @@ -0,0 +1,132 @@ +import sys +import os +import time +import re +import serial +import serial.tools.list_ports + +def get_serial_ports(): + return serial.tools.list_ports.comports() + +def main(): + print("====================================================") + print(" WiFi CSI Frame Rate (FPS) Tester ") + print("====================================================") + + ports = get_serial_ports() + if not ports: + print("No serial ports found! Please connect your ESP32-C6 (or CSI transmitting board).") + sys.exit(1) + + print("Available Serial Ports:") + for idx, port in enumerate(ports): + print(f"[{idx}] {port.device} - {port.description}") + + while True: + try: + choice = input(f"\nSelect port [0-{len(ports)-1}]: ").strip() + idx = int(choice) + if 0 <= idx < len(ports): + port_name = ports[idx].device + break + else: + print("Invalid selection.") + except ValueError: + print("Please enter a valid number.") + + # Baud rates choices + print("\nBaud Rate Options:") + print("[0] 115200 (Default)") + print("[1] 921600 (High Speed)") + print("[2] Custom") + baud_choice = input("Select baud rate [0-2]: ").strip() + if baud_choice == '1': + baud_rate = 921600 + elif baud_choice == '2': + try: + baud_rate = int(input("Enter custom baud rate: ").strip()) + except ValueError: + baud_rate = 115200 + else: + baud_rate = 115200 + + print(f"\nOpening {port_name} at {baud_rate} baud...") + try: + ser = serial.Serial(port_name, baud_rate, timeout=0.5) + print("Port successfully opened! Listening for CSI frames...") + except Exception as e: + print(f"Error opening port: {e}") + sys.exit(1) + + # Matching regex patterns identical to get_tool_fusion.py + pattern = re.compile(r'data:\s*\[([^\]]+)\]') + fallback_pattern = re.compile(r'\[([^\]]+)\]') + + print("\nPress Ctrl+C to stop the test.\n") + print("-" * 55) + print(f"{'Time':<12} | {'FPS (Hz)':<10} | {'Total Frames':<12} | {'Avg Subcarriers/Frame'}") + print("-" * 55) + + total_frames = 0 + fps_counter = 0 + subcarrier_counts = [] + + start_time = time.time() + last_report_time = start_time + + try: + while True: + if ser.in_waiting > 0: + try: + line = ser.readline().decode('utf-8', errors='ignore').strip() + if not line: + continue + + match = pattern.search(line) or fallback_pattern.search(line) + if match: + data_str = match.group(1) + # Count subcarriers (interleaved, so divide by 2 for complex count) + data_list = [x.strip() for x in data_str.split(',') if x.strip()] + subcarriers_len = len(data_list) // 2 + + fps_counter += 1 + total_frames += 1 + subcarrier_counts.append(subcarriers_len) + except Exception as e: + # Ignore decode or read errors during high speed stream + pass + + # Report every 1.0 second + current_time = time.time() + if current_time - last_report_time >= 1.0: + elapsed = current_time - last_report_time + fps = fps_counter / elapsed + + avg_subcarriers = 0 + if subcarrier_counts: + avg_subcarriers = sum(subcarrier_counts) / len(subcarrier_counts) + + local_time_str = time.strftime("%H:%M:%S", time.localtime(current_time)) + print(f"{local_time_str:<12} | {fps:<10.1f} | {total_frames:<12} | {avg_subcarriers:.1f}") + + # Reset interval stats + fps_counter = 0 + subcarrier_counts = [] + last_report_time = current_time + + time.sleep(0.001) # small sleep to prevent high CPU load + + except KeyboardInterrupt: + print("\nTest stopped by user.") + + duration = time.time() - start_time + print("-" * 55) + print(f"Test Duration: {duration:.1f} seconds") + print(f"Total Received: {total_frames} frames") + if duration > 0: + print(f"Average FPS: {total_frames / duration:.1f} Hz") + + ser.close() + +if __name__ == '__main__': + main()