Human Activity Recognition Using Smartphones Dataset

数据集地址http://archive.ics.uci.edu/ml/datasets/Human+Activity+Recognition+Using+Smartphones

        对于智能家居、健康监护等场景来说，如何保护用户隐私一直是个痛点。基于摄像头的方案虽然直观，但隐私问题挥之不去。相比之下，雷达技术因其非侵入式、保护隐私的特性而备受关注。

原始数据有9个输入通道。当前只采用1个通道数据

只用 body_acc_x（一个通道），把每个 128 点的时间窗做 STFT（短时傅里叶变换）→ 时频谱（log-幅值谱），然后用一个小型 2D CNN 做 6 类分类（UCI HAR 的 6 个动作）。你只需要把 DATA_DIR 改成你本地 UCI HAR Dataset 的路径即可。

# -*- coding: utf-8 -*-
"""
UCI HAR - single channel (body_acc_x) -> STFT spectrogram -> 2D CNN classification
Author: you
"""

import os
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import matplotlib.pyplot as plt  # ← 新增：用于保存时频谱图片

# ---------------------------
# 0. 可复现实验
# ---------------------------
SEED = 2025
torch.manual_seed(SEED)
np.random.seed(SEED)

# ---------------------------
# 1. 仅加载 body_acc_x + 标签
# ---------------------------
def load_body_acc_x(root, split="train"):
    """
    root: 指向 'UCI HAR Dataset' 目录
    split: 'train' 或 'test'
    返回:
      X: (N, 128) float32
      y: (N,) int64 in [0..5]
    """
    inertial_dir = os.path.join(root, split, "Inertial Signals")
    assert os.path.isdir(inertial_dir), f"Not found: {inertial_dir}"

    # 标签：1..6 -> 0..5
    y_path = os.path.join(root, split, f"y_{split}.txt")
    y = np.loadtxt(y_path, dtype=int) - 1

    # 仅 body_acc_x
    fname = "body_acc_x_train.txt" if split == "train" else "body_acc_x_test.txt"
    X = np.loadtxt(os.path.join(inertial_dir, fname)).astype(np.float32)  # (N, 128)

    assert X.shape[0] == y.shape[0], f"X and y size mismatch: {X.shape[0]} vs {y.shape[0]}"
    return X, y.astype(np.int64)

# ---------------------------
# 2. 批量 STFT -> 时频谱
# ---------------------------
def stft_spectrogram_batch(x_np, n_fft=64, win_len=32, hop_len=8, log_eps=1e-6):
    """
    x_np: (N, T) numpy float32
    返回: torch.Tensor (N, 1, F, T_frames)，log(1+|STFT|)
    说明：
      - UCI HAR 每窗 T=128, 采样率约 50Hz（这里仅相对频率，不影响 CNN）
      - n_fft >= win_len；win_len=32, hop=8 => 时间帧数较充分（约 13 帧）
    """
    x = torch.from_numpy(x_np)  # (N, T)
    # 去均值(去漂移/重力趋势对频谱的影响)
    x = x - x.mean(dim=1, keepdim=True)

    window = torch.hann_window(win_len)
    # stft: (N, F, T_frames), complex
    Xc = torch.stft(
        x, n_fft=n_fft, hop_length=hop_len, win_length=win_len,
        window=window, center=False, return_complex=True
    )
    mag = torch.abs(Xc)                    # (N, F, T_frames)
    spec = torch.log1p(mag + log_eps)     # 稳定的对数幅度
    spec = spec.unsqueeze(1)               # (N, 1, F, T_frames)
    return spec

# ---------------------------
# 3. 数据集（用预计算的谱，做标准化）
# ---------------------------
class SpecTensorDataset(Dataset):
    def __init__(self, spec_tensor, labels, mean=None, std=None):
        """
        spec_tensor: (N, 1, F, T_frames) torch.float32
        labels: (N,) numpy 或 torch
        mean, std: 用训练集计算的全局均值和方差（标量）
        """
        self.X = spec_tensor
        self.y = torch.as_tensor(labels, dtype=torch.long)

        if mean is None or std is None:
            mean = self.X.mean()
            std = self.X.std() + 1e-8
        self.mean = mean
        self.std = std

    def __len__(self):
        return self.X.shape[0]

    def __getitem__(self, idx):
        x = (self.X[idx] - self.mean) / self.std
        return x, self.y[idx]

# ---------------------------
# 4. 一个轻量 2D CNN
# ---------------------------
class SpecCNN(nn.Module):
    def __init__(self, num_classes=6):
        super().__init__()
        self.net = nn.Sequential(
            nn.Conv2d(1, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),                # 下采样 1/2

            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2),                # 再下采样 1/2

            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(inplace=True),
            nn.AdaptiveAvgPool2d((4, 4))    # 自适应汇聚到固定大小
        )
        self.head = nn.Sequential(
            nn.Flatten(),
            nn.Dropout(0.4),
            nn.Linear(128 * 4 * 4, 256),
            nn.ReLU(inplace=True),
            nn.Dropout(0.4),
            nn.Linear(256, num_classes)
        )

    def forward(self, x):
        x = self.net(x)
        x = self.head(x)
        return x

# ---------------------------
# 可视化：保存前两条样本的时频谱图（转换后的样本图片）
# ---------------------------
def save_spec_image(spec_tensor, idx, out_path, title_prefix="Train Sample"):
    """
    spec_tensor: (N, 1, F, T_frames) torch.float32
    idx: 样本索引
    out_path: 保存路径
    """
    spec = spec_tensor[idx, 0].cpu().numpy()  # 取出 (F, T_frames)
    plt.figure(figsize=(6, 4))
    plt.imshow(spec, aspect='auto', origin='lower')  # 频率从下到上
    plt.colorbar(label='log(1+|STFT|)')
    plt.xlabel('Time frames')
    plt.ylabel('Frequency bins')
    plt.title(f'{title_prefix} #{idx}')
    plt.tight_layout()
    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
    plt.savefig(out_path, dpi=200)
    plt.close()

# ---------------------------
# 5. 训练与评估
# ---------------------------
def train_one_epoch(model, loader, criterion, optimizer, device):
    model.train()
    total_loss, total_correct, total = 0.0, 0, 0
    for xb, yb in loader:
        xb, yb = xb.to(device), yb.to(device)
        optimizer.zero_grad(set_to_none=True)
        logits = model(xb)
        loss = criterion(logits, yb)
        loss.backward()
        optimizer.step()

        total_loss += loss.item() * yb.size(0)
        total_correct += (logits.argmax(1) == yb).sum().item()
        total += yb.size(0)
    return total_loss / total, total_correct / total

@torch.no_grad()
def evaluate(model, loader, criterion, device):
    model.eval()
    total_loss, total_correct, total = 0.0, 0, 0
    for xb, yb in loader:
        xb, yb = xb.to(device), yb.to(device)
        logits = model(xb)
        loss = criterion(logits, yb)
        total_loss += loss.item() * yb.size(0)
        total_correct += (logits.argmax(1) == yb).sum().item()
        total += yb.size(0)
    return total_loss / total, total_correct / total

# ---------------------------
# 6. 主流程
# ---------------------------
def main():
    # 1) 改成你的本地路径
    DATA_DIR = r"D:\project\deep-learning-for-image-processing-master\data_set\UCI HAR Dataset\UCI HAR Dataset"

    # 2) 仅 body_acc_x
    X_tr, y_tr = load_body_acc_x(DATA_DIR, "train")
    X_te, y_te = load_body_acc_x(DATA_DIR, "test")

    # 3) 批量生成时频谱（log|STFT|）
    #    参数可按需调整：n_fft=64, win_len=32, hop_len=8
    spec_tr = stft_spectrogram_batch(X_tr, n_fft=64, win_len=32, hop_len=8)  # (N,1,F,Tf)
    spec_te = stft_spectrogram_batch(X_te, n_fft=64, win_len=32, hop_len=8)

    # —— 输出两张“转换后的样本图片” —— #
    save_spec_image(spec_tr, 0, "figs/spec_train_sample_1.png", title_prefix="Train Spectrogram")
    save_spec_image(spec_tr, 1, "figs/spec_train_sample_2.png", title_prefix="Train Spectrogram")
    print("Saved spectrogram images to:",
          "figs/spec_train_sample_1.png, figs/spec_train_sample_2.png")

    # 4) 用训练集统计做标准化
    mean = spec_tr.mean()
    std = spec_tr.std() + 1e-8

    train_ds = SpecTensorDataset(spec_tr, y_tr, mean, std)
    test_ds  = SpecTensorDataset(spec_te, y_te, mean, std)

    train_loader = DataLoader(train_ds, batch_size=128, shuffle=True, num_workers=0, drop_last=False)
    test_loader  = DataLoader(test_ds,  batch_size=256, shuffle=False, drop_last=False)

    # 5) 模型/优化器
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = SpecCNN(num_classes=6).to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4)
    scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='max', factor=0.5, patience=3)

    # 6) 训练
    epochs = 80
    best_acc, best_state = 0.0, None
    for ep in range(1, epochs + 1):
        tr_loss, tr_acc = train_one_epoch(model, train_loader, criterion, optimizer, device)
        te_loss, te_acc = evaluate(model, test_loader, criterion, device)
        scheduler.step(te_acc)
        if te_acc > best_acc:
            best_acc, best_state = te_acc, {k: v.cpu().clone() for k, v in model.state_dict().items()}
        print(f"Epoch {ep:02d} | train loss {tr_loss:.4f} acc {tr_acc:.4f} | "
              f"test loss {te_loss:.4f} acc {te_acc:.4f} | best {best_acc:.4f}")

    # 7) 载入最佳模型并最终评估
    if best_state is not None:
        model.load_state_dict(best_state)
    te_loss, te_acc = evaluate(model, test_loader, criterion, device)
    print(f"\nFinal Test  | loss {te_loss:.4f} acc {te_acc:.4f}")

if __name__ == "__main__":
    main()