From eight parts to a heatmap: parsing the 10-minute CSI dataset

By Wu Jun · Updated 2026-10-29 · Wi-Fi CSI, dataset, Python, parsing, heatmap · 6 min read

The dataset is not a "download and open" archive: it is eight binary parts that must be verified, reassembled and parsed record by record. This walkthrough turns the 10-minute Zenodo capture into an amplitude matrix and a heatmap, and checks the record count, duration and whole-file SHA-256. All numbers come from the project's real capture ^[1]; the data and scripts are public ^{[2] [3]}.

1. Download and dependencies

Download the eight parts, manifest.json and REASSEMBLE.md from https://zenodo.org/records/23434789 into one directory. On the Python side you only need the standard library, NumPy and Matplotlib:

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

2. Verify before analysing

Do not skip verification. A missing or corrupted part can still produce plausible-looking but wrong numbers:

cat csi-10m.part-*.bin > csi_10m.bin
shasum -a 256 csi_10m.bin
# expected: 527466ceb8b5f835e55ff348f02808736b947db39e7baac1f34e375de146a440

You can also compare each part against the sha256 entries in manifest.json.

3. Read the records: outer container and inner header

The file is a record stream. Each record has two layers:

• outer: double file_ts + uint16 length (10 bytes, little endian);
• inner: a length-byte payload whose first 24 bytes are a header, followed by interleaved int16 I/Q.

This dataset uses the v1 24-byte header. Keep chain_info as well, because bit 15 marks the end of a measurement:

import struct
from collections import Counter

import numpy as np

REC = struct.Struct("<dH")             # outer: file timestamp + payload length
HDR = struct.Struct("<I6sbBBBBBHHHH")  # inner 24-byte header

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"})

Expected:

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

ts is in microseconds; the main data has 256 subcarriers, with four 64-subcarrier records. Do not assume that every measurement has all eight chains. This capture contains both 8-chain and 4-chain groups, plus one unterminated record at the end.

4. Select one valid chain per measurement

Chains from the same measurement share almost the same timestamp. The 57,293 records must not be treated as 57,293 independent samples. The code below prefers tx0/rx0, then falls back to the strongest valid chain:

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")

This simple filter produces roughly a 7,608 × 256 matrix over 599.92 seconds, or about 12.68 independent measurements/s. The full analysis script also removes 25 groups with excessive zero values, leaving 7,583 measurements. Keeping those details is more faithful than claiming that every group always contains eight chains.

Do not turn that average back into a uniform clock without inspection. The capture-quality audit reports a median interval of 41 ms, a 95th percentile of 322 ms and 57 gaps above one second. Mask long gaps or analyse contiguous windows before estimating a low-frequency spectrum.

5. Build the amplitude heatmap

Stack the selected 256 subcarriers for each measurement over time:

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)

The result should look like the heatmap on the dataset page. If it is pure noise or a constant, check part order, the 24-byte header offset and I/Q endianness before adjusting colour limits.

6. Compute a one-second rolling motion metric

The raw measurement gaps are not perfectly uniform. Interpolate onto a 20 Hz grid, then average the frame-to-frame amplitude change over one second. Interpolation regularises time; it does not create new information:

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)

Walking produces clear peaks; a quiet room stays near the baseline. This is only a cheap "activity" feature. It is not fall detection, and respiration or heart rate should not be estimated from this amplitude-difference metric alone.

7. Common mistakes

8. Next steps

• Phase processing and respiration/heart rate: Python pipeline and respiration guide;
• Dataset details and citation: dataset page;
• Capture-side reproduction: GL-MT6000 guide.

9. Reproduction acceptance checks

Before building an algorithm on top of the parser, require the following:

• sorting all eight parts and reassembling them produces the whole-file SHA-256 in the manifest;
• the parser reads 57,293 outer records and leaves no unread tail bytes;
• closing groups on bit 15 produces 7,609 complete measurements, rather than dividing the record count by eight;
• the matrix is approximately 7,608 × 256 and the outlier-removal rule is documented;
• the time span is about 599.92 seconds and the independent measurement rate is about 12.68 Hz;
• the summary records the maximum gap, RSSI, SNR and chain-selection policy;
• repeated runs on the same input produce identical record counts, matrix shapes and SHA-256 values.

If those checks disagree, fix parsing and grouping before applying a deep-learning model or changing plot colours.