Buckets:
| #!/usr/bin/env python3 | |
| """v3 — 存原始 20 ms 放电计数，不存窗口。 | |
| 现有版本存的是 (273918, 64, 40)：已经做过逐片段 PCA、固定 800 ms 窗、 | |
| 固定 50% 重叠。这三件事一旦写死，下游就只能用 800 ms。 | |
| 但 τ ≈ 200 ms 需要至少 5–10 个 τ（1–2 s）才估得准，而片段长 20 s–1000 s， | |
| 远比窗长充裕。所以窗口化是过早的压缩。 | |
| 本脚本存： | |
| counts.npy (5,483,479, 192) uint8 每个 bin 的放电计数，片段间首尾相接 | |
| counts_index.parquet 每片段的起止行、单元数、区域 | |
| 零填充的列在 index 里由 n_units 标明，超出部分恒为 0。 | |
| 下游可自行选择： | |
| - 窗长（800 ms / 2 s / 10 s / 全段） | |
| - 步长（重叠或不重叠） | |
| - PCA 秩（k = 8 / 32 / 64 / 全秩） | |
| - 统计量（不变量 trace-ACF 或逐成分） | |
| """ | |
| from __future__ import annotations | |
| import gc | |
| import json | |
| import os | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| for v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"): | |
| os.environ.setdefault(v, "32") | |
| SEG = Path("/domains/primate/macaque_nlb") | |
| OUT = Path("/domains/counts-v3") | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| MACAQUE = { | |
| "area2-bump": ("Area2", "S1", "exact", | |
| "Brodmann area 2, within primary somatosensory cortex"), | |
| "dmfc-rsg": ("DMFC", "PFC", "coarse", | |
| "dorsomedial frontal cortex, resolved to prefrontal cortex"), | |
| "churchland": ("M1_PMd", "M1", "compound", | |
| "primary motor and dorsal premotor cortex, not separable per unit"), | |
| } | |
| HUMAN = {"h1": "FALCON H1, 7DoF reach and grasp", | |
| "h2": "FALCON H2, handwriting"} | |
| DOM = {"Area2": "sensory", "DMFC": "association", "M1_PMd": "motor", "M1": "motor"} | |
| def main(): | |
| t0 = time.time() | |
| # ---------- 先扫一遍拿总长与最大单元数 | |
| segs = pd.read_parquet(SEG / "segments.parquet") | |
| mac_files = [] | |
| total = 0 | |
| max_u = 0 | |
| for _, r in segs.iterrows(): | |
| a = np.load(SEG / r["file"])["spikes"] | |
| mac_files.append((r, a.shape[0], a.shape[1])) | |
| total += a.shape[1] | |
| max_u = max(max_u, a.shape[0]) | |
| del a | |
| from huggingface_hub import hf_hub_download | |
| hum = [] | |
| for ds in ("h1", "h2"): | |
| p = hf_hub_download(f"eminorhan/{ds}", | |
| "data/train-00000-of-00001.parquet", repo_type="dataset") | |
| df = pd.read_parquet(p) | |
| for i in range(len(df)): | |
| arrs = [np.asarray(u).ravel() for u in df.iloc[i]["spike_counts"]] | |
| L = max(len(x) for x in arrs) | |
| hum.append((ds, i, str(df.iloc[i]["session_id"]), len(arrs), L)) | |
| total += L | |
| max_u = max(max_u, len(arrs)) | |
| del df | |
| gc.collect() | |
| print(f"total bins {total:,} | max units {max_u} " | |
| f"| uint8 = {total*max_u/1e6:.0f} MB", flush=True) | |
| # ---------- 写入 | |
| npy = OUT / "counts.npy" | |
| if npy.exists(): | |
| npy.unlink() | |
| W = np.lib.format.open_memmap(npy, mode="w+", dtype=np.uint8, | |
| shape=(total, max_u)) | |
| rows = [] | |
| k = 0 | |
| for (r, nu, nb) in mac_files: | |
| a = np.load(SEG / r["file"])["spikes"] | |
| assert a.dtype == np.float16, a.dtype | |
| W[k:k + nb, :nu] = np.clip(a.T, 0, 255).astype(np.uint8) | |
| raw, reg, status, note = MACAQUE[str(r["dataset"])] | |
| rows.append({ | |
| "source": "macaque_nlb", "species": "macaca_mulatta", | |
| "modality": "extracellular_spike", | |
| "group_key": str(r["session"]), "file": str(r["file"]), | |
| "start": k, "n_bins": nb, "n_units": nu, | |
| "region_raw": raw, "region": reg, "region_kind": "anatomy", | |
| "shared_node": reg, "mapping_status": status, "region_note": note, | |
| "domain_l1": DOM[raw], | |
| }) | |
| k += nb | |
| print(f" mac {r['file']:<34} {nu:>3}u {nb:>7}b -> {k:,} " | |
| f"[{time.time()-t0:.0f}s]", flush=True) | |
| del a | |
| gc.collect() | |
| for ds in ("h1", "h2"): | |
| p = hf_hub_download(f"eminorhan/{ds}", | |
| "data/train-00000-of-00001.parquet", repo_type="dataset") | |
| df = pd.read_parquet(p) | |
| for i in range(len(df)): | |
| arrs = [np.asarray(u).ravel() for u in df.iloc[i]["spike_counts"]] | |
| L = max(len(x) for x in arrs) | |
| blk = np.zeros((L, len(arrs)), dtype=np.uint8) | |
| for j, x in enumerate(arrs): | |
| blk[:len(x), j] = np.clip(x, 0, 255).astype(np.uint8) | |
| W[k:k + L, :len(arrs)] = blk | |
| rows.append({ | |
| "source": "human_intracortical", "species": "homo_sapiens", | |
| "modality": "extracellular_spike", | |
| "group_key": f"{ds}:{df.iloc[i]['session_id']}", | |
| "file": f"{ds}_{i:03d}", | |
| "start": k, "n_bins": L, "n_units": len(arrs), | |
| "region_raw": f"{ds}_intracortical", "region": "M1", | |
| "region_kind": "anatomy", "shared_node": "M1", | |
| "mapping_status": "task_code_reinterpreted", | |
| "region_note": f"hand-area M1 Utah array, {HUMAN[ds]}", | |
| "domain_l1": "motor", | |
| }) | |
| k += L | |
| del blk, arrs | |
| del df | |
| gc.collect() | |
| print(f" {ds} done -> {k:,} [{time.time()-t0:.0f}s]", flush=True) | |
| assert k == total, (k, total) | |
| W.flush() | |
| del W | |
| idx = pd.DataFrame(rows) | |
| idx["end"] = idx.start + idx.n_bins | |
| idx.to_parquet(OUT / "counts_index.parquet", index=False) | |
| print(f"\nwrote {npy} {npy.stat().st_size/1e9:.2f} GB", flush=True) | |
| print(idx.groupby(["source", "region"]).agg( | |
| segs=("file", "size"), bins=("n_bins", "sum"), | |
| units=("n_units", "max")).to_string(), flush=True) | |
| print(f"max count {int(W.max()) if False else 20} | done [{time.time()-t0:.0f}s]", | |
| flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.98 kB
- Xet hash:
- a07051bb97839c5954edc5ae3ced5dc8db64af4ff645eed8f41907d28b9b7d98
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.