Buckets:

illusion-brain
/

brain-domains

Files

xet

illusion-brain/brain-domains / scripts /build_slice_catalog.py

redmingwei

6 days ago

download

raw

10 kB

#!/usr/bin/env python3
"""build_slice_catalog — L3 检索层：把每个切片做成可检索记录
不改写数据、不生成任务样本，只做"能查"。核心不是多一个 parquet，而是：
**检索出来的每一行都带着该行的适用性限制**，否则查的人会把"不可用的字段"
当成可用。因此 catalog 里除数值外还有 caveats（逐行计算）与 tier（L0–L4 到位情况）。
两级粒度：
  level=slice  来自 l2_invariants（成员 × 切片）
  level=region 来自 l2_region_invariants（成员 × 切片 × 区，逐单元真标签）
输出：
  framework/slice_catalog.parquet
  framework/retrieval_index.json（字段表 + 词表 + caveat 规则 + 示例查询命中数）
"""
from __future__ import annotations

import json
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path("/domains")
OUT = ROOT / "framework"

# ---------------- 成员元数据（物种/模态/是否有逐单元区标签）
# unknown 表示本会话没有可靠来源，宁可留空也不猜
MEMBER_META = {
    "steinmetz": ("mouse", "spikes", True),
    "counts_v4": ("mouse+macaque", "counts", False),
    "primate_macaque_nlb": ("macaque", "spikes", True),
    "primate_dmfc_rsg": ("macaque", "spikes", False),
    "primate_human_eeg": ("human", "eeg", False),
    "omnibrain_multi": ("mouse+macaque+human", "spikes", False),
    "omnibrain_spike_view": ("mouse+macaque+human", "spikes", False),
    "omnibrain_spike_view_human": ("human", "spikes", False),
    "omnibrain_spike_view_macaque": ("macaque", "spikes", False),
    "omnibrain_eeg_view": ("human", "eeg", False),
    "omnibrain_graph_view": ("mouse", "connectome", False),
    "omnibrain4d_eeg_train": ("human", "eeg", False),
    "omnibrain4d_eeg_test": ("human", "eeg", False),
    "flywire": ("fly", "connectome", False),
}

# 逐行 caveat 规则：条件 + 说明
REGION_HIGH_DISPERSION = {"SCs", "ILA", "CA3", "PL", "MOs"}


def caveats_for(row) -> list[str]:
    c = []
    mem = str(row["member"])
    modality = str(row["modality"])
    if modality == "eeg":
        c.append("EEG 的跨通道共同分量已被白化：rho1/rho_half 不适用，只能看单通道 ACF")
    if mem == "omnibrain_multi" and "human" in str(row["species"]):
        c.append("人类颅内部分无区标签（NWB 无 electrodes 表），region 只能为空")
    if str(row["region_confidence"]) == "none":
        c.append("无区标签")
    if pd.notna(row.get("region")) and str(row["region"]) in REGION_HIGH_DISPERSION \
            and mem == "steinmetz":
        c.append("该区跨 session 离散度 >1.4（log 尺度），不要按区引用绝对值")
    st = row.get("rho_half_status")
    if pd.notna(st) and st != "ok":
        c.append(f"rho_half 不可估（{st}），只有上界/下界")
    tt = row.get("tau_status")
    if pd.notna(tt) and tt != "ok":
        c.append(f"τ 不可用（{tt}）")
    if pd.notna(row.get("n_units")) and float(row["n_units"]) < 8:
        c.append("单元数 <8，任何区级量的方差都很大")
    # 注意：PR 不能跨成员比较是**全局**限制，写在 retrieval_index.gaps 里，
    # 不挂到每一行——否则每行都至少带一条 caveat，逐行警告会被淹没。
    return c


# ---------------- 1. 切片级
L = pd.read_parquet(OUT / "l2_invariants.parquet")
keep = ["member", "slice", "n_bins", "bin_ms", "rate_hz", "rho1", "pooled_acf_lag1",
        "rho_half_ms", "rho_half_status", "rho_zero_ms", "pr", "erank", "spec_slope",
        "tau_ms", "tau_status", "acf1_half_ms", "acf1_half_status"]
keep = [k for k in keep if k in L.columns]
S = L[keep].copy()
S["level"] = "slice"
S["region"] = np.nan
S["region_confidence"] = "n/a"
S["n_units"] = np.nan
print(f"slice 级: {len(S)} 行")

# ---------------- 2. 区级
R = pd.read_parquet(OUT / "l2_region_invariants.parquet")
keep_r = ["member", "slice", "region", "n_units", "n_bins", "bin_ms", "rate_hz", "rho1",
          "pooled_acf_lag1", "rho_half_ms", "rho_half_status", "rho_zero_ms", "pr",
          "erank", "spec_slope", "label_source"]
keep_r = [k for k in keep_r if k in R.columns]
G = R[keep_r].copy()
G["level"] = "region"
_conf = {"allen_acronym": "anatomical", "dandi_000070_electrodes": "anatomical"}
G["region_confidence"] = G.label_source.map(lambda s: _conf.get(s, str(s)))
for col in ("tau_ms", "tau_status", "acf1_half_ms", "acf1_half_status"):
    G[col] = np.nan
print(f"region 级: {len(G)} 行")

C = pd.concat([S, G], ignore_index=True, sort=False)
C["duration_s"] = C.n_bins.astype(float) * C.bin_ms.astype(float) / 1000.0

# ---------------- 3. 成员元数据 + 来源 pin
meta = C.member.map(lambda m: MEMBER_META.get(m, ("unknown", "unknown", None)))
C["species"] = [m[0] for m in meta]
C["modality"] = [m[1] for m in meta]
C["has_unit_region"] = [m[2] for m in meta]

SRC = pd.read_csv(OUT / "sources.csv")
PIN = dict(zip(SRC.source_id, SRC.pin_value))
STAT = dict(zip(SRC.source_id, SRC.pin_status))
MEM2SRC = {
    "steinmetz": "steinmetz2019",
    "primate_macaque_nlb": "dandi000070",
    "primate_dmfc_rsg": "dandi000130",
    "primate_human_eeg": "things-eeg2",
    "omnibrain_eeg_view": "things-eeg2",
    "omnibrain4d_eeg_train": "things-eeg2",
    "omnibrain4d_eeg_test": "things-eeg2",
    "omnibrain_multi": "dandi000950",
    "flywire": "flywire",
}
C["source_id"] = C.member.map(MEM2SRC)
C["pin_value"] = C.source_id.map(PIN)
C["pin_status"] = C.source_id.map(STAT)
C["unit_id"] = (C.member.astype(str) + "|" + C.slice.astype(str) + "|"
                + C.region.fillna("-").astype(str))

# ---------------- 4. L0–L4 到位情况（每行）
C["tier_L0"] = C.pin_status.fillna("none")
C["tier_L1"] = "dedup_checked"
C["tier_L2"] = np.where(C.rho_half_status == "ok", "rho_half_ok",
                        "rho_half_unavailable")
C["tier_L3"] = "retrieval_only"
C["tier_L4"] = np.where(C.has_unit_region == True, "unit_region_verified",
                        "region_unverified")

C["caveats"] = [caveats_for(r) for _, r in C.iterrows()]
C["n_caveats"] = C.caveats.map(len)

# 数值列四舍五入，避免 inf/巨大的浮点尾巴
for c in ("rho_half_ms", "rho_zero_ms", "acf1_half_ms", "tau_ms", "rate_hz",
          "rho1", "pooled_acf_lag1", "pr", "erank", "spec_slope", "duration_s"):
    if c in C.columns:
        C[c] = C[c].astype(float).round(6)

order = ["unit_id", "level", "member", "species", "modality", "region",
         "region_confidence", "has_unit_region", "slice", "n_units", "n_bins", "bin_ms",
         "duration_s", "rate_hz", "rho1", "pooled_acf_lag1", "rho_half_ms",
         "rho_half_status", "rho_zero_ms", "pr", "erank", "spec_slope", "tau_ms",
         "tau_status", "acf1_half_ms", "acf1_half_status", "source_id", "pin_value",
         "pin_status", "tier_L0", "tier_L1", "tier_L2", "tier_L3", "tier_L4",
         "n_caveats", "caveats"]
C[order].to_parquet(OUT / "slice_catalog.parquet", index=False)
print(f"catalog: {len(C)} 行, {C.unit_id.nunique()} 个唯一 unit_id")
print("species 分布:", C.drop_duplicates("member").species.value_counts().to_dict())
print("无元数据的成员:", sorted(set(C[C.species == "unknown"].member)))

# ---------------- 5. 示例查询（命中数写进 JSON，便于文档引用真实数字）
def q(label, **kw):
    m = pd.Series(True, index=C.index)
    for k, v in kw.items():
        if isinstance(v, (list, tuple, set)):
            m &= C[k].isin(list(v))
        else:
            m &= (C[k] == v)
    sub = C[m]
    print(f"  {label:38s} -> {len(sub):4d} 行")
    return int(len(sub))


print("\n示例查询命中数:")
ex = {
    "mouse_CA1_region_level": q("小鼠 CA1（区级）", member="steinmetz", level="region",
                                region="CA1"),
    "macaque_rho_half_ok": q("猕猴 rho_half 可估（区级）", member="primate_macaque_nlb",
                             level="region", rho_half_status="ok"),
    "has_unit_region": q("有逐单元区标签的成员", has_unit_region=True),
    "no_caveat": q("无任何 caveat", n_caveats=0),
}
ex["long_record_gt_1000s"] = int((C[C.level == "slice"].duration_s > 1000).sum())
print(f"  {'时长 >1000 s':38s} -> {ex['long_record_gt_1000s']:4d} 行")

payload = {
    "schema_version": "1.0",
    "n_rows": int(len(C)),
    "n_slice_level": int((C.level == "slice").sum()),
    "n_region_level": int((C.level == "region").sum()),
    "levels": ["slice", "region"],
    "fields": {
        "unit_id": "member|slice|region 的唯一键",
        "level": "slice=成员×切片；region=成员×切片×区（逐单元真标签）",
        "has_unit_region": "该成员是否有逐单元区标签（True 才能做区级分析）",
        "rho_half_ms": "跨单元共享相关的半衰（模型无关）；EEG 不适用",
        "tau_status": "plateau 拟合的状态；非 ok 时 tau_ms 不可用",
        "tier_L2": "该行 L2 是否可用（rho_half_ok / rho_half_unavailable）",
        "tier_L4": "区身份是否逐单元核验过（unit_region_verified / region_unverified）",
        "caveats": "该行的适用性限制，检索后必须先读",
    },
    "vocabularies": {
        "species": sorted(set(C.species.dropna())),
        "modality": sorted(set(C.modality.dropna())),
        "member": sorted(set(C.member.astype(str))),
        "region": sorted(set(C.region.dropna().astype(str))),
    },
    "gaps": {
        "eeg_rho_unavailable": "EEG 成员的 rho1/rho_half 被白化，不可用",
        "human_intracortical_no_region": "人类颅内无区标签（NWB 无 electrodes 表）",
        "pr_not_comparable": "PR 不能跨成员比较",
        "no_quality_model": "三条质量轴均为负结果，catalog 不含 quality 分",
    },
    "example_queries": ex,
}
(OUT / "retrieval_index.json").write_text(
    json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
print("\n已写 slice_catalog.parquet 与 retrieval_index.json")
print("caveat 条数分布:", C.n_caveats.value_counts().sort_index().to_dict())

Xet Storage Details

Size:: 10 kB
Xet hash:: f49b36dd78ab878c9a38b63337eed99be23fb00a9050ad63c0979414b58ceb77

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	#!/usr/bin/env python3
	"""build_slice_catalog — L3 检索层：把每个切片做成可检索记录

	不改写数据、不生成任务样本，只做"能查"。核心不是多一个 parquet，而是：
	检索出来的每一行都带着该行的适用性限制，否则查的人会把"不可用的字段"
	当成可用。因此 catalog 里除数值外还有 caveats（逐行计算）与 tier（L0–L4 到位情况）。

	两级粒度：
	level=slice 来自 l2_invariants（成员 × 切片）
	level=region 来自 l2_region_invariants（成员 × 切片 × 区，逐单元真标签）

	输出：
	framework/slice_catalog.parquet
	framework/retrieval_index.json（字段表 + 词表 + caveat 规则 + 示例查询命中数）
	"""
	from __future__ import annotations

	import json
	from pathlib import Path

	import numpy as np
	import pandas as pd

	ROOT = Path("/domains")
	OUT = ROOT / "framework"

	# ---------------- 成员元数据（物种/模态/是否有逐单元区标签）
	# unknown 表示本会话没有可靠来源，宁可留空也不猜
	MEMBER_META = {
	"steinmetz": ("mouse", "spikes", True),
	"counts_v4": ("mouse+macaque", "counts", False),
	"primate_macaque_nlb": ("macaque", "spikes", True),
	"primate_dmfc_rsg": ("macaque", "spikes", False),
	"primate_human_eeg": ("human", "eeg", False),
	"omnibrain_multi": ("mouse+macaque+human", "spikes", False),
	"omnibrain_spike_view": ("mouse+macaque+human", "spikes", False),
	"omnibrain_spike_view_human": ("human", "spikes", False),
	"omnibrain_spike_view_macaque": ("macaque", "spikes", False),
	"omnibrain_eeg_view": ("human", "eeg", False),
	"omnibrain_graph_view": ("mouse", "connectome", False),
	"omnibrain4d_eeg_train": ("human", "eeg", False),
	"omnibrain4d_eeg_test": ("human", "eeg", False),
	"flywire": ("fly", "connectome", False),
	}

	# 逐行 caveat 规则：条件 + 说明
	REGION_HIGH_DISPERSION = {"SCs", "ILA", "CA3", "PL", "MOs"}


	def caveats_for(row) -> list[str]:
	c = []
	mem = str(row["member"])
	modality = str(row["modality"])
	if modality == "eeg":
	c.append("EEG 的跨通道共同分量已被白化：rho1/rho_half 不适用，只能看单通道 ACF")
	if mem == "omnibrain_multi" and "human" in str(row["species"]):
	c.append("人类颅内部分无区标签（NWB 无 electrodes 表），region 只能为空")
	if str(row["region_confidence"]) == "none":
	c.append("无区标签")
	if pd.notna(row.get("region")) and str(row["region"]) in REGION_HIGH_DISPERSION \
	and mem == "steinmetz":
	c.append("该区跨 session 离散度 >1.4（log 尺度），不要按区引用绝对值")
	st = row.get("rho_half_status")
	if pd.notna(st) and st != "ok":
	c.append(f"rho_half 不可估（{st}），只有上界/下界")
	tt = row.get("tau_status")
	if pd.notna(tt) and tt != "ok":
	c.append(f"τ 不可用（{tt}）")
	if pd.notna(row.get("n_units")) and float(row["n_units"]) < 8:
	c.append("单元数 <8，任何区级量的方差都很大")
	# 注意：PR 不能跨成员比较是全局限制，写在 retrieval_index.gaps 里，
	# 不挂到每一行——否则每行都至少带一条 caveat，逐行警告会被淹没。
	return c


	# ---------------- 1. 切片级
	L = pd.read_parquet(OUT / "l2_invariants.parquet")
	keep = ["member", "slice", "n_bins", "bin_ms", "rate_hz", "rho1", "pooled_acf_lag1",
	"rho_half_ms", "rho_half_status", "rho_zero_ms", "pr", "erank", "spec_slope",
	"tau_ms", "tau_status", "acf1_half_ms", "acf1_half_status"]
	keep = [k for k in keep if k in L.columns]
	S = L[keep].copy()
	S["level"] = "slice"
	S["region"] = np.nan
	S["region_confidence"] = "n/a"
	S["n_units"] = np.nan
	print(f"slice 级: {len(S)} 行")

	# ---------------- 2. 区级
	R = pd.read_parquet(OUT / "l2_region_invariants.parquet")
	keep_r = ["member", "slice", "region", "n_units", "n_bins", "bin_ms", "rate_hz", "rho1",
	"pooled_acf_lag1", "rho_half_ms", "rho_half_status", "rho_zero_ms", "pr",
	"erank", "spec_slope", "label_source"]
	keep_r = [k for k in keep_r if k in R.columns]
	G = R[keep_r].copy()
	G["level"] = "region"
	_conf = {"allen_acronym": "anatomical", "dandi_000070_electrodes": "anatomical"}
	G["region_confidence"] = G.label_source.map(lambda s: _conf.get(s, str(s)))
	for col in ("tau_ms", "tau_status", "acf1_half_ms", "acf1_half_status"):
	G[col] = np.nan
	print(f"region 级: {len(G)} 行")

	C = pd.concat([S, G], ignore_index=True, sort=False)
	C["duration_s"] = C.n_bins.astype(float) * C.bin_ms.astype(float) / 1000.0

	# ---------------- 3. 成员元数据 + 来源 pin
	meta = C.member.map(lambda m: MEMBER_META.get(m, ("unknown", "unknown", None)))
	C["species"] = [m[0] for m in meta]
	C["modality"] = [m[1] for m in meta]
	C["has_unit_region"] = [m[2] for m in meta]

	SRC = pd.read_csv(OUT / "sources.csv")
	PIN = dict(zip(SRC.source_id, SRC.pin_value))
	STAT = dict(zip(SRC.source_id, SRC.pin_status))
	MEM2SRC = {
	"steinmetz": "steinmetz2019",
	"primate_macaque_nlb": "dandi000070",
	"primate_dmfc_rsg": "dandi000130",
	"primate_human_eeg": "things-eeg2",
	"omnibrain_eeg_view": "things-eeg2",
	"omnibrain4d_eeg_train": "things-eeg2",
	"omnibrain4d_eeg_test": "things-eeg2",
	"omnibrain_multi": "dandi000950",
	"flywire": "flywire",
	}
	C["source_id"] = C.member.map(MEM2SRC)
	C["pin_value"] = C.source_id.map(PIN)
	C["pin_status"] = C.source_id.map(STAT)
	C["unit_id"] = (C.member.astype(str) + "\|" + C.slice.astype(str) + "\|"
	+ C.region.fillna("-").astype(str))

	# ---------------- 4. L0–L4 到位情况（每行）
	C["tier_L0"] = C.pin_status.fillna("none")
	C["tier_L1"] = "dedup_checked"
	C["tier_L2"] = np.where(C.rho_half_status == "ok", "rho_half_ok",
	"rho_half_unavailable")
	C["tier_L3"] = "retrieval_only"
	C["tier_L4"] = np.where(C.has_unit_region == True, "unit_region_verified",
	"region_unverified")

	C["caveats"] = [caveats_for(r) for _, r in C.iterrows()]
	C["n_caveats"] = C.caveats.map(len)

	# 数值列四舍五入，避免 inf/巨大的浮点尾巴
	for c in ("rho_half_ms", "rho_zero_ms", "acf1_half_ms", "tau_ms", "rate_hz",
	"rho1", "pooled_acf_lag1", "pr", "erank", "spec_slope", "duration_s"):
	if c in C.columns:
	C[c] = C[c].astype(float).round(6)

	order = ["unit_id", "level", "member", "species", "modality", "region",
	"region_confidence", "has_unit_region", "slice", "n_units", "n_bins", "bin_ms",
	"duration_s", "rate_hz", "rho1", "pooled_acf_lag1", "rho_half_ms",
	"rho_half_status", "rho_zero_ms", "pr", "erank", "spec_slope", "tau_ms",
	"tau_status", "acf1_half_ms", "acf1_half_status", "source_id", "pin_value",
	"pin_status", "tier_L0", "tier_L1", "tier_L2", "tier_L3", "tier_L4",
	"n_caveats", "caveats"]
	C[order].to_parquet(OUT / "slice_catalog.parquet", index=False)
	print(f"catalog: {len(C)} 行, {C.unit_id.nunique()} 个唯一 unit_id")
	print("species 分布:", C.drop_duplicates("member").species.value_counts().to_dict())
	print("无元数据的成员:", sorted(set(C[C.species == "unknown"].member)))

	# ---------------- 5. 示例查询（命中数写进 JSON，便于文档引用真实数字）
	def q(label, **kw):
	m = pd.Series(True, index=C.index)
	for k, v in kw.items():
	if isinstance(v, (list, tuple, set)):
	m &= C[k].isin(list(v))
	else:
	m &= (C[k] == v)
	sub = C[m]
	print(f" {label:38s} -> {len(sub):4d} 行")
	return int(len(sub))


	print("\n示例查询命中数:")
	ex = {
	"mouse_CA1_region_level": q("小鼠 CA1（区级）", member="steinmetz", level="region",
	region="CA1"),
	"macaque_rho_half_ok": q("猕猴 rho_half 可估（区级）", member="primate_macaque_nlb",
	level="region", rho_half_status="ok"),
	"has_unit_region": q("有逐单元区标签的成员", has_unit_region=True),
	"no_caveat": q("无任何 caveat", n_caveats=0),
	}
	ex["long_record_gt_1000s"] = int((C[C.level == "slice"].duration_s > 1000).sum())
	print(f" {'时长 >1000 s':38s} -> {ex['long_record_gt_1000s']:4d} 行")

	payload = {
	"schema_version": "1.0",
	"n_rows": int(len(C)),
	"n_slice_level": int((C.level == "slice").sum()),
	"n_region_level": int((C.level == "region").sum()),
	"levels": ["slice", "region"],
	"fields": {
	"unit_id": "member\|slice\|region 的唯一键",
	"level": "slice=成员×切片；region=成员×切片×区（逐单元真标签）",
	"has_unit_region": "该成员是否有逐单元区标签（True 才能做区级分析）",
	"rho_half_ms": "跨单元共享相关的半衰（模型无关）；EEG 不适用",
	"tau_status": "plateau 拟合的状态；非 ok 时 tau_ms 不可用",
	"tier_L2": "该行 L2 是否可用（rho_half_ok / rho_half_unavailable）",
	"tier_L4": "区身份是否逐单元核验过（unit_region_verified / region_unverified）",
	"caveats": "该行的适用性限制，检索后必须先读",
	},
	"vocabularies": {
	"species": sorted(set(C.species.dropna())),
	"modality": sorted(set(C.modality.dropna())),
	"member": sorted(set(C.member.astype(str))),
	"region": sorted(set(C.region.dropna().astype(str))),
	},
	"gaps": {
	"eeg_rho_unavailable": "EEG 成员的 rho1/rho_half 被白化，不可用",
	"human_intracortical_no_region": "人类颅内无区标签（NWB 无 electrodes 表）",
	"pr_not_comparable": "PR 不能跨成员比较",
	"no_quality_model": "三条质量轴均为负结果，catalog 不含 quality 分",
	},
	"example_queries": ex,
	}
	(OUT / "retrieval_index.json").write_text(
	json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
	print("\n已写 slice_catalog.parquet 与 retrieval_index.json")
	print("caveat 条数分布:", C.n_caveats.value_counts().sort_index().to_dict())