从 8 个分片到热图：10 分钟 CSI 数据集解析实战

作者 Wu Jun · 更新于 2026-10-29 · Wi-Fi CSI, 数据集, Python, 解析, 热图 · 约 6 分钟

这份数据集不是"下载完就结束"的压缩包：它是 8 个二进制分片，需要先校验、再重组、再按记录格式解析。 本文用一段可以照着敲的 Python 代码，把 Zenodo 上的 10 分钟数据变成幅度矩阵和热图，并验证记录数、 时长与整包 SHA-256。所有数字都来自本项目实测 ^[1]，数据与脚本已公开 ^{[2] [3]}。

1. 准备：下载与依赖

从 Zenodo 记录 https://zenodo.org/records/23434789 下载 8 个分片、manifest.json 与 REASSEMBLE.md，放到同一个目录。Python 侧只需要标准库 + NumPy + Matplotlib：

python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install numpy matplotlib

2. 先校验，再分析

不要跳过校验。分片缺失或下载损坏时，后面的解析结果会"看起来正常但数值错误"：

cat csi-10m.part-*.bin > csi_10m.bin
shasum -a 256 csi_10m.bin
# 期望：527466ceb8b5f835e55ff348f02808736b947db39e7baac1f34e375de146a440

也可以逐个分片与 manifest.json 里的 sha256 对比；整包校验通过后再进入下一步。

3. 读取记录：外层容器与内层头

文件是"记录流"，每条记录由两部分组成：

• 外层：double file_ts + uint16 length（10 字节，小端）；
• 内层：length 字节 payload，前 24 字节是头，后面是 int16 I/Q 交错。

这份数据集使用 v1 的 24 字节头。读取时同时保留 chain_info，因为 bit15 才是"本次测量 的最后一条链"这一边界信号：

import struct
from collections import Counter

import numpy as np

REC = struct.Struct("<dH")             # 外层：文件时间戳 + payload 长度
HDR = struct.Struct("<I6sbBBBBBHHHH")  # 内层 24 字节头

def parse(path):
    records, measurements, pending = [], [], []
    with open(path, "rb") as f:
        while True:
            raw = f.read(REC.size)
            if not raw:
                break
            if len(raw) != REC.size:
                raise EOFError("truncated outer record")

            file_ts, length = REC.unpack(raw)
            payload = f.read(length)
            if len(payload) != length:
                raise EOFError(f"truncated payload: want {length}, got {len(payload)}")
            if length < HDR.size:
                raise ValueError("payload smaller than CSI header")

            ts, ta, rssi, snr, bw, ch, mode, _pad, n, tx, rx, chain = (
                HDR.unpack_from(payload)
            )
            iq = np.frombuffer(
                payload, dtype="<i2", count=n * 2, offset=HDR.size
            )
            item = {
                "file_ts": file_ts, "ts": ts, "ta": ta, "rssi": rssi,
                "snr": snr, "n": n, "tx": tx, "rx": rx,
                "chain": chain, "iq": iq,
            }
            records.append(item)
            pending.append(item)

            if chain & 0x8000:
                measurements.append(pending)
                pending = []

    if pending:
        print("warning: incomplete tail records:", len(pending))
    return records, measurements

records, measurements = parse("csi_10m.bin")
print("records:", len(records))
print("complete measurements:", len(measurements))
print("chains per group:", Counter(map(len, measurements)))
print("first header:", {k: v for k, v in records[0].items() if k != "iq"})

预期结果：

records: 57293
complete measurements: 7609
chains per group: Counter({8: 6714, 4: 895})
warning: incomplete tail records: 1

ts 是微秒；主数据有 256 个子载波，另有 4 条 64 子载波记录。不要假设每次测量都必然 有完整的 8 条链。这份抓包里既有 8 链组，也有 4 链组，末尾还有 1 条未闭合记录。

4. 每次测量只选一条有效链

同一次测量的多条链共享接近相同的时间点，不能把 57,293 条记录当成 57,293 个独立采样。 下面的代码优先选择 tx0/rx0；若它为空，再选择平均幅度最大的有效链：

def amplitude(iq):
    return np.hypot(iq[0::2].astype(float), iq[1::2].astype(float))

times, rows = [], []
for group in measurements:
    valid = [item for item in group if item["n"] == 256 and item["iq"].any()]
    if not valid:
        continue

    chosen = next(
        (item for item in valid if (item["tx"], item["rx"]) == (0, 0)),
        None,
    )
    if chosen is None:
        chosen = max(valid, key=lambda item: amplitude(item["iq"]).mean())

    times.append(chosen["file_ts"])
    rows.append(amplitude(chosen["iq"]))

t = np.asarray(times, dtype=float)
t -= t[0]
H = np.vstack(rows)
print("usable measurements:", H.shape)
print("measurement rate:", len(t) / (t[-1] - t[0]), "Hz")

按这个简单筛选得到约 7,608 × 256 的矩阵，时间跨度约 599.92 秒，独立测量率约 12.68 Hz。原分析脚本还会剔除 25 个零值占比过高的异常组，最终使用 7,583 次测量。 把这些差异保留下来，比强行宣称"每组一定有 8 条链"更符合真实数据。

不要把这个平均值直接当成均匀时钟。CSI 质量审计显示： 组间隔中位数为 41ms、p95 为 322ms，且有 57 个间隔超过 1 秒。做低频频谱估计前必须屏蔽长空档， 或只在连续窗口内分析。

5. 生成幅度热图

把每次测量选中的 256 个子载波按时间堆成矩阵：

import matplotlib.pyplot as plt

plt.figure(figsize=(12, 5))
plt.imshow(
    H.T, aspect="auto", origin="lower", cmap="viridis",
    extent=[t[0], t[-1], 0, H.shape[1]],
)
plt.xlabel("time (s)")
plt.ylabel("subcarrier")
plt.title("GL-MT6000 CSI amplitude (one valid chain per measurement)")
plt.colorbar(label="amplitude")
plt.tight_layout()
plt.savefig("csi_heatmap.png", dpi=150)

这张图应该与数据集页中的热图形态一致；如果全是噪声或常数，优先检查分片顺序、24 字节头 偏移和 I/Q 字节序，而不是先调颜色范围。

6. 计算 1 秒滚动运动指标

原始测量间隔并不完全均匀。下面的代码先按时间戳插值到 20 Hz，再对相邻快照的幅度差做 1 秒滑动平均。插值只是统一时间网格，不会凭空增加信息：

FS = 20.0
uniform_t = np.arange(0.0, t[-1], 1.0 / FS)
uniform_H = np.column_stack([
    np.interp(uniform_t, t, H[:, i]) for i in range(H.shape[1])
])

delta = np.abs(np.diff(uniform_H, axis=0)).mean(axis=1)
window = int(FS)
motion = np.convolve(delta, np.ones(window) / window, mode="valid")
motion_t = uniform_t[1 + window // 2: 1 + window // 2 + len(motion)]

plt.figure(figsize=(12, 3))
plt.plot(motion_t, motion)
plt.xlabel("time (s)")
plt.ylabel("1-second mean |Δamplitude|")
plt.tight_layout()
plt.savefig("csi_motion.png", dpi=150)

在真实数据里，人走动时会看到明显的尖峰；静止时曲线接近基线。这个指标只是最便宜的 "有人活动"特征，不等于跌倒检测；呼吸和心率也不能仅靠这个幅度差指标可靠估计。

7. 常见错误

8. 下一步

• 相位处理与呼吸/心率：见 Python 处理链路 与 呼吸检测；
• 数据集说明与引用格式：见 数据集页；
• 采集端复现：见 GL-MT6000 复现教程。

9. 复现验收清单

在继续做算法之前，至少让解析器满足以下条件：

• 8 个分片按文件名排序重组后，整包 SHA-256 与 manifest 一致；
• 解析结果为 57,293 条外层记录，且没有未读取的尾部字节；
• 按 bit15 闭合后得到 7,609 个完整测量组，而不是简单地用记录数除以 8；
• 矩阵为 7,608 × 256 左右，并明确记录剔除异常组的规则；
• 时间跨度约 599.92 秒，独立测量率约 12.68 Hz；
• 最大时间空档、RSSI、SNR 与所用链选择规则写入摘要；
• 同一份输入重复运行得到完全相同的记录数、矩阵形状和 SHA-256。

如果这些数字不一致，先修正解析和分组，不要用深度学习模型或调色参数掩盖数据问题。