Buckets:
| #!/usr/bin/env python3 | |
| """build_l2_missing — 补齐 L2 缺的三个成员。 | |
| 1. primate_dmfc_rsg (NWB)：Units 表 = ragged spike_times + 每单元 obs_intervals。 | |
| 按各单元共有的观测区间拼接，20 ms 分箱，得到 (总bin, n_units)。 | |
| 2. omnibrain_multi：1,409,850 × 64 × 8 窗口，按 (source, group_key) 分组算协方差类不变量。 | |
| 每组最多取 5000 个窗口，控制协方差计算量。 | |
| 3. counts_v4_ragged：175 段 uint8 + 索引。这是同一批数据的另一种编码， | |
| 行里打 representation=ragged_uint8 与 duplicate_of=<源文件>，避免下游重复计数。 | |
| 只补模型无关量 + 协方差类；τ 只在 dmfc_rsg 上算（它有时间序列）。 | |
| """ | |
| from __future__ import annotations | |
| import glob | |
| import json | |
| import time | |
| from pathlib import Path | |
| import h5py | |
| import numpy as np | |
| import pandas as pd | |
| ROOT = Path("/domains") | |
| OUT = ROOT / "framework" | |
| MAX_LAG = 400 | |
| CAP_WINDOWS = 5000 | |
| def n2(n): | |
| k = 1 | |
| while k < n: | |
| k <<= 1 | |
| return k | |
| def curve_parts(x, max_lag): | |
| T, d = x.shape | |
| xc = x - x.mean(axis=0, keepdims=True) | |
| N = n2(2 * T) | |
| F = np.fft.rfft(xc, n=N, axis=0) | |
| Fs = F.sum(axis=1) | |
| tot = np.fft.irfft(Fs * np.conj(Fs), n=N)[:max_lag + 1] | |
| dia = np.fft.irfft((np.abs(F) ** 2).sum(axis=1), n=N)[:max_lag + 1] | |
| nrm = np.arange(T, T - max_lag - 1, -1) | |
| tot = tot / np.maximum(nrm, 1) | |
| dia = dia / np.maximum(nrm, 1) | |
| t0 = max(tot[0], 1e-12) | |
| return (tot - dia) / t0, dia / max(dia[0], 1e-12) | |
| def cross_time(curve, bin_ms, frac): | |
| if len(curve) < 3: | |
| return None, "too_short" | |
| y1 = float(curve[1]) | |
| if not np.isfinite(y1) or y1 <= 0: | |
| return None, "bad_start" | |
| thr = frac * y1 | |
| idx = np.flatnonzero(curve[2:] <= thr) | |
| if len(idx) == 0: | |
| return None, "censored" | |
| l = int(idx[0]) + 2 | |
| y0, yb = float(curve[l - 1]), float(curve[l]) | |
| f = (y0 - thr) / (y0 - yb) if y0 > yb else 0.0 | |
| return float((l - 1 + f) * bin_ms), "ok" | |
| def cov_metrics(X): | |
| X = np.asarray(X, dtype=np.float64) | |
| if X.ndim != 2 or X.shape[0] < 5 or X.shape[1] < 2: | |
| return dict(pr=None, erank=None, spec_slope=None) | |
| Xc = X - X.mean(axis=0, keepdims=True) | |
| C = (Xc.T @ Xc) / max(X.shape[0] - 1, 1) | |
| ev = np.clip(np.linalg.eigvalsh(C)[::-1], 0, None) | |
| s = ev.sum() | |
| if s <= 0: | |
| return dict(pr=None, erank=None, spec_slope=None) | |
| pr = float(s * s / max((ev ** 2).sum(), 1e-30)) | |
| p = ev / s | |
| p = p[p > 0] | |
| erank = float(np.exp(-(p * np.log(p)).sum())) | |
| k = min(20, len(ev)) | |
| rr = np.arange(1, k + 1) | |
| ok = ev[:k] > 0 | |
| slope = float(np.polyfit(np.log(rr[ok]), np.log(ev[:k][ok]), 1)[0]) if ok.sum() > 2 else None | |
| return dict(pr=pr, erank=erank, spec_slope=slope) | |
| BASE = dict(member=None, slice=None, kind=None, n_units=None, n_bins=None, bin_ms=None, | |
| rate_hz=None, note=None, pr=None, erank=None, spec_slope=None, | |
| tau_ms=None, rho1=None, snr=None, n_points=0, max_lag_ms=None, | |
| neg_frac=None, tau_status="not_computed", tau_available=False, | |
| pooled_acf_lag1=None, pooled_acf_half_ms=None, pooled_acf_zero_ms=None, | |
| pooled_acf_half_status=None, pooled_acf_zero_status=None, | |
| representation=None, duplicate_of=None) | |
| rows = [] | |
| t00 = time.time() | |
| def add(**kw): | |
| r = dict(BASE) | |
| r.update(kw) | |
| r["tau_available"] = r["tau_status"] == "ok" | |
| rows.append(r) | |
| # ---------------------------------------------------------------- 1. NWB | |
| print("[1] dmfc_rsg NWB", flush=True) | |
| for f in sorted(glob.glob(str(ROOT / "primate/dmfc_rsg/*.nwb"))): | |
| name = Path(f).stem | |
| with h5py.File(f, "r") as h: | |
| st = h["units/spike_times"][:] | |
| sti = h["units/spike_times_index"][:] | |
| oi = h["units/obs_intervals"][:] | |
| oii = h["units/obs_intervals_index"][:] | |
| n_units = len(sti) | |
| starts = np.concatenate([[0], sti[:-1]]) | |
| per_unit_spikes = [st[a:b] for a, b in zip(starts, sti)] | |
| istarts = np.concatenate([[0], oii[:-1]]) | |
| per_unit_ints = [oi[a:b] for a, b in zip(istarts, oii)] | |
| # 取所有单元共有的区间（按 1 ms 取整比对） | |
| def key(iv): | |
| return (round(float(iv[0]), 3), round(float(iv[1]), 3)) | |
| common = None | |
| for ivs in per_unit_ints: | |
| s = {key(iv) for iv in ivs} | |
| common = s if common is None else (common & s) | |
| common = sorted(common) if common else [] | |
| if len(common) < 2: | |
| print(f" {name}: common intervals={len(common)}，退回全局跨度", flush=True) | |
| lo = min(float(iv[0]) for ivs in per_unit_ints for iv in ivs) | |
| hi = max(float(iv[1]) for ivs in per_unit_ints for iv in ivs) | |
| common = [(lo, hi)] | |
| bin_s = 0.020 | |
| edges_list, total = [], 0 | |
| for a, b in common: | |
| nb = max(int(np.floor((b - a) / bin_s)), 1) | |
| edges_list.append((a, nb)) | |
| total += nb | |
| X = np.zeros((total, n_units), dtype=np.float32) | |
| off = 0 | |
| for a, nb in edges_list: | |
| e = a + np.arange(nb + 1) * bin_s | |
| for j, sp in enumerate(per_unit_spikes): | |
| c, _ = np.histogram(sp, bins=e) | |
| X[off:off + nb, j] = c | |
| off += nb | |
| print(f" {name}: units={n_units} intervals={len(common)} bins={total} " | |
| f"({total*20/1000:.0f}s)", flush=True) | |
| ml = min(MAX_LAG, X.shape[0] - 2) | |
| rho, acf = curve_parts(X, ml) | |
| rec = dict(member="primate_dmfc_rsg", slice=name, kind="nwb_binned", | |
| n_units=n_units, n_bins=int(total), bin_ms=20.0, | |
| rate_hz=float(X.mean() / 0.02), | |
| note=f"concatenated_{len(common)}_intervals", representation="nwb", | |
| max_lag_ms=ml * 20.0, rho1=float(rho[1]), | |
| pooled_acf_lag1=float(acf[1])) | |
| rec.update(cov_metrics(X)) | |
| rec["pooled_acf_half_ms"], rec["pooled_acf_half_status"] = cross_time(acf, 20.0, 0.5) | |
| rec["pooled_acf_zero_ms"], rec["pooled_acf_zero_status"] = cross_time(acf, 20.0, 0.0) | |
| rec["tau_status"] = "not_computed" # τ 需要零模型，本轮不重算 | |
| add(**rec) | |
| # ---------------------------------------------------------------- 2. omnibrain-multi | |
| print("\n[2] omnibrain_multi", flush=True) | |
| idx = pd.read_parquet(ROOT / "omnibrain-multi/index_timeseries.parquet") | |
| ts = np.load(ROOT / "omnibrain-multi/timeseries.npy", mmap_mode="r") | |
| rng = np.random.default_rng(0) | |
| for (src, gk), g in idx.groupby(["source", "group_key"], sort=False): | |
| pos = g.index.to_numpy() | |
| if len(pos) > CAP_WINDOWS: | |
| pos = rng.choice(pos, CAP_WINDOWS, replace=False) | |
| X = np.asarray(ts[np.sort(pos)], dtype=np.float32).reshape(-1, 64) | |
| rec = dict(member="omnibrain_multi", slice=f"{src}|{gk}", kind="windows64x8", | |
| n_units=64, n_bins=8, bin_ms=100.0, | |
| note=f"n_win={len(pos)}", representation="windowed_100ms", | |
| tau_status="not_estimable_8bins", | |
| pooled_acf_half_status="not_estimable_8bins", | |
| pooled_acf_zero_status="not_estimable_8bins") | |
| rec.update(cov_metrics(X)) | |
| add(**rec) | |
| print(f" groups={idx.groupby(['source','group_key']).ngroups} [{time.time()-t00:.0f}s]", | |
| flush=True) | |
| # ---------------------------------------------------------------- 3. counts-v4 | |
| INCLUDE_COUNTS_V4 = False # 175 段是既有成员的另一种编码，不进入 L2 | |
| print("\n[3] counts_v4_ragged", "SKIPPED" if not INCLUDE_COUNTS_V4 else "", flush=True) | |
| ci = pd.read_parquet(ROOT / "counts-v4/counts_index.parquet") if INCLUDE_COUNTS_V4 else None | |
| flat = np.load(ROOT / "counts-v4/counts_ragged.npy", mmap_mode="r") | |
| for _, r in (ci.iterrows() if ci is not None else []): | |
| nb, nu = int(r.n_bins_20ms), int(r.n_units) | |
| off, nel = int(r.offset), int(r.n_elements) | |
| if nel != nb * nu: | |
| print(f" skip seg {r.segment_idx}: n_elements={nel} != {nb}*{nu}", flush=True) | |
| continue | |
| # counts_ragged.npy 是 uint8 整数计数，不要再乘 8 | |
| x = np.asarray(flat[off:off + nel], dtype=np.float32).reshape(nb, nu) | |
| bin_ms = float(r.bin_ms) | |
| ml = min(MAX_LAG, nb - 2) | |
| if ml < 20: | |
| continue | |
| rho, acf = curve_parts(x, ml) | |
| rec = dict(member="counts_v4_ragged", slice=f"seg{int(r.segment_idx):03d}", | |
| kind="ragged_uint8", n_units=nu, n_bins=nb, bin_ms=bin_ms, | |
| rate_hz=float(x.mean() / (bin_ms / 1000.0)), | |
| note=f"{r.source}|{Path(str(r.file)).stem}", representation="ragged_uint8", | |
| duplicate_of=str(r.file), max_lag_ms=ml * bin_ms, rho1=float(rho[1]), | |
| pooled_acf_lag1=float(acf[1]), tau_status="not_computed", | |
| pooled_acf_half_ms=None, pooled_acf_zero_ms=None) | |
| rec.update(cov_metrics(x)) | |
| rec["pooled_acf_half_ms"], rec["pooled_acf_half_status"] = cross_time(acf, bin_ms, 0.5) | |
| rec["pooled_acf_zero_ms"], rec["pooled_acf_zero_status"] = cross_time(acf, bin_ms, 0.0) | |
| add(**rec) | |
| print(f" segments={0 if ci is None else len(ci)} [{time.time()-t00:.0f}s]", flush=True) | |
| # ---------------------------------------------------------------- merge | |
| new = pd.DataFrame(rows) | |
| print(f"\nnew rows={len(new)} members={new.member.nunique()} [{time.time()-t00:.0f}s]") | |
| P = OUT / "l2_invariants.parquet" | |
| old = pd.read_parquet(P) | |
| keep = old[~old.member.isin(new.member.unique())] | |
| out = pd.concat([keep, new], ignore_index=True, sort=False) | |
| out.to_parquet(P, index=False) | |
| out.to_csv(OUT / "l2_invariants.csv", index=False) | |
| D = OUT / "l2_decay.parquet" | |
| d = pd.read_parquet(D) | |
| dnew = new[["member", "slice", "pooled_acf_lag1", "pooled_acf_half_ms", | |
| "pooled_acf_zero_ms", "pooled_acf_half_status", "pooled_acf_zero_status", | |
| "max_lag_ms"]].copy() | |
| dmm = pd.concat([d[~d.member.isin(new.member.unique())], dnew], ignore_index=True, | |
| sort=False) | |
| dmm.to_parquet(D, index=False) | |
| print(f"l2_invariants rows={len(out)} members={out.member.nunique()}") | |
| print(out.groupby("member").agg(n=("slice", "size"), tau_ok=("tau_available", "sum"), | |
| pr=("pr", "median"), erank=("erank", "median")).to_string()) | |
Xet Storage Details
- Size:
- 10 kB
- Xet hash:
- b8d8986d45f978c1ec30c070c6ca6f703b331db7d1ce1974b0e6903e21c1eeee
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.