Buckets:
| #!/usr/bin/env python3 | |
| """audit_fix — 逐条核对之前报的"四处不一致"，区分真缺陷和我自己的索引误报。 | |
| A. omnibrain 各视图的窗口/comps 文件实际数量 vs 索引里的"对象行数" | |
| B. omnibrain-multi 的 labels.json vs index_timeseries.parquet 实际行数 | |
| C. primate_dmfc_rsg 的 NWB 能否读出真实形状 | |
| D. 残留临时文件 | |
| """ | |
| import glob | |
| import json | |
| import re | |
| from collections import Counter | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| ROOT = Path("/domains") | |
| def section(t): | |
| print("\n" + "=" * 70) | |
| print(t) | |
| print("=" * 70) | |
| section("A. omnibrain 视图：文件数 vs 对象行数") | |
| VIEWS = ["omnibrain/spike_view", "omnibrain/spike_view_macaque", | |
| "omnibrain/spike_view_human", "omnibrain/eeg_view"] | |
| for v in VIEWS: | |
| files = sorted(glob.glob(str(ROOT / v / "**" / "*.npz"), recursive=True)) | |
| kinds = Counter() | |
| shapes = Counter() | |
| for f in files: | |
| z = np.load(f, mmap_mode="r") | |
| for k in z.keys(): | |
| kinds[k] += 1 | |
| shapes[(k, tuple(z[k].shape))] += 1 | |
| print(f"\n[{v}]") | |
| print(f" files={len(files)} keys={dict(kinds)}") | |
| for k in ("windows", "comps"): | |
| sub = {s: c for (kk, s), c in shapes.items() if kk == k} | |
| if not sub: | |
| continue | |
| dup = {s: c for s, c in sub.items() if c > 1} | |
| print(f" {k}: 文件={sum(sub.values())} 不同形状={len(sub)} 重复形状={dup}") | |
| # 会话编号连续性 | |
| if "windows" in kinds: | |
| ids = sorted(int(m.group(1)) for f in files | |
| if "windows" in Path(f).stem | |
| for m in [re.search(r"(\d+)$", Path(f).stem)] if m) | |
| if ids: | |
| gaps = [i for i in range(min(ids), max(ids) + 1) if i not in ids] | |
| print(f" windows 会话号 {min(ids)}–{max(ids)} 缺={gaps}") | |
| section("B. omnibrain-multi：labels.json vs 实际索引") | |
| lab_p = ROOT / "omnibrain-multi/labels.json" | |
| idx_p = ROOT / "omnibrain-multi/index_timeseries.parquet" | |
| lab = json.loads(lab_p.read_text(encoding="utf-8")) | |
| idx = pd.read_parquet(idx_p) | |
| print(f"index_timeseries.parquet: rows={len(idx)} cols={list(idx.columns)}") | |
| print(f"labels.json sample.count = {lab['labels']['sample']['count']}") | |
| print(f"labels.json count_by_source = {lab['labels']['sample']['count_by_source']}") | |
| if "source" in idx.columns: | |
| print(f"实际 by source = {idx['source'].value_counts().to_dict()}") | |
| for src, v in (lab.get("sources") or {}).items(): | |
| print(f" labels.sources[{src}].n_windows = {v.get('n_windows')}") | |
| section("C. primate_dmfc_rsg：NWB 真实形状") | |
| try: | |
| import h5py | |
| ok = True | |
| except Exception as e: | |
| print("h5py 不可用:", e) | |
| ok = False | |
| if ok: | |
| for f in sorted(glob.glob(str(ROOT / "primate/dmfc_rsg/*.nwb"))): | |
| with h5py.File(f, "r") as h: | |
| cands = [] | |
| def walk(name, obj): | |
| if isinstance(obj, h5py.Dataset) and obj.ndim >= 2: | |
| cands.append((name, obj.shape, str(obj.dtype))) | |
| h.visititems(walk) | |
| cands.sort(key=lambda t: -int(np.prod(t[1]))) | |
| print(f"\n[{Path(f).name}] datasets(ndim>=2) top3:") | |
| for name, shp, dt in cands[:3]: | |
| print(f" {name} {shp} {dt}") | |
| section("D. 残留临时文件") | |
| tmp = glob.glob(str(ROOT / "**" / "*.tmp"), recursive=True) + \ | |
| glob.glob(str(ROOT / "**" / "*~"), recursive=True) | |
| print("tmp/backup files:", tmp if tmp else "none") | |
Xet Storage Details
- Size:
- 3.48 kB
- Xet hash:
- 227c7b98deff469d3ec247777428293622b7cce4408c534445b652e9e75bbe58
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.