Buckets:

illusion-brain
/

brain-domains

Files

xet

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

redmingwei

6 days ago

download

raw

10.7 kB

#!/usr/bin/env python3
"""build_l1_structure — L1 结构层落地（v2：region 分五级可信度）。
region 不是一个可比的键，而是五种不同强度的证据。分级规则（按 mapping_status）：
  exact / exact_single_species  -> anatomical          解剖定位，且有本体节点
  uberon_missing                -> anatomical_unmapped 解剖定位，但无 UBERON 节点
  task_code_reinterpreted       -> inferred            由任务码推定，不是解剖定位
  compound                      -> compound            一个标签覆盖多个区（如 M1_PMd）
  coarse / rollup               -> coarse              粗粒度或上卷
  unassigned / 缺失             -> none                无解剖标签
产出 structure.parquet（带 region_confidence）与 joins.json（带各级计数与建议用法）。
"""
from __future__ import annotations

import json
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path("/domains")
OUT = ROOT / "framework"
rows = []
notes = {}

CONF = {
    "exact": "anatomical",
    "exact_single_species": "anatomical",
    "uberon_missing": "anatomical_unmapped",
    "task_code_reinterpreted": "inferred",
    "compound": "compound",
    "coarse": "coarse",
    "rollup": "coarse",
    "unassigned": "none",
}


def conf_series(s):
    return s.map(CONF).fillna("none")


# 粗标签解析到父区（与 counts-v4 既有映射一致，不引入新精度）
RESOLVE = {"DMFC": "PFC", "Area2": "S1"}


def add(df, level, member, provenance, conf=None):
    d = df.copy()
    for c in ["source", "species", "modality", "region", "region_kind", "domain_l1"]:
        if c not in d.columns:
            d[c] = None
    d["region_resolved"] = d["region"].map(
        lambda r: RESOLVE.get(str(r), r) if pd.notna(r) else r)
    d["region_resolution"] = d["region"].map(
        lambda r: "parent_rollup" if str(r) in RESOLVE else "as_labeled")
    if member == "steinmetz":
        d["region_assignment"] = "per_unit"
    elif member == "counts_v4":
        d["region_assignment"] = "per_segment"
    elif member == "flywire":
        d["region_assignment"] = "per_node"
    elif member == "omnibrain_multi":
        d["region_assignment"] = np.where(
            d["source"].astype(str) == "mouse_steinmetz", "session_majority", "window_area")
    else:
        d["region_assignment"] = "unknown"
    if isinstance(conf, str) and conf in d.columns:
        d["region_confidence"] = conf_series(d[conf])
    else:
        d["region_confidence"] = conf or "none"
    g = (d.groupby(["source", "species", "modality", "region", "region_resolved",
                    "region_resolution", "region_assignment", "region_kind",
                    "region_confidence", "domain_l1"], dropna=False)
         .size().reset_index(name="n"))
    g["level"] = level
    g["member"] = member
    g["provenance"] = provenance
    rows.append(g[["level", "member", "source", "species", "modality", "region",
                   "region_resolved", "region_resolution", "region_assignment",
                   "region_kind", "region_confidence", "domain_l1", "n", "provenance"]])
    notes[member] = {"rows_in": int(len(d)), "groups": int(len(g)),
                     "confidence": d.region_confidence.value_counts().to_dict()}


# 1) 小鼠：单元级，Allen 缩写，属解剖定位
u = pd.read_parquet(ROOT / "steinmetz/units.parquet")
u["source"] = "mouse_steinmetz"
u["species"] = "mus_musculus"
u["modality"] = "extracellular_spike"
u["domain_l1"] = u["functional_domain"]
u["region_kind"] = "anatomy"
add(u, "unit", "steinmetz", "steinmetz/units.parquet", conf="anatomical")

# 2) counts-v4：片段级，带 mapping_status
ci = pd.read_parquet(ROOT / "counts-v4/counts_index.parquet")
add(ci, "segment", "counts_v4", "counts-v4/counts_index.parquet", conf="mapping_status")

# 3) omnibrain-multi：窗级标签，带 mapping_status
mi = pd.read_parquet(ROOT / "omnibrain-multi/index_timeseries.parquet")
for c in ["domain_l1", "region_kind", "mapping_status"]:
    if c not in mi.columns:
        mi[c] = None
add(mi, "window", "omnibrain_multi", "omnibrain-multi/index_timeseries.parquet",
    conf="mapping_status")

# 4) FlyWire：super_class 属粗粒度分组
fw = pd.read_parquet(ROOT / "flywire/node_features.parquet")
fw["source"] = "flywire"
fw["species"] = "drosophila_melanogaster"
fw["modality"] = "connectome"
fw["region_kind"] = "anatomy"
fw["domain_l1"] = None
add(fw, "node", "flywire", "flywire/node_features.parquet", conf="coarse")

S = pd.concat(rows, ignore_index=True)
S = S[S.region.notna() | S.domain_l1.notna()]
S.to_parquet(OUT / "structure.parquet", index=False)
S.to_csv(OUT / "structure.csv", index=False)


def keyinfo(key, kind, members, semantics, joinable, caveat, extra=None,
            vocab_rel="shared"):
    per = {}
    for m in members:
        sub = S[(S.member == m) & S[key].notna()]
        per[m] = sorted(set(sub[key].astype(str)))
    union = set().union(*per.values()) if per else set()
    rec = dict(key=key, kind=kind, members=members, n_distinct_union=len(union),
               n_distinct_by_member={m: len(v) for m, v in per.items()},
               vocabulary_relation=vocab_rel,
               values_by_member={m: v[:25] for m, v in per.items()},
               joinable_across_members=joinable,
               semantics=semantics, caveat=caveat)
    if extra:
        rec.update(extra)
    return rec


members_with_region = sorted(S[S.region.notna()].member.unique())
members_with_domain = sorted(S[S.domain_l1.notna()].member.unique())
conf_by_level = (S.groupby(["level", "region_confidence"]).n.sum()
                 .unstack(fill_value=0).to_dict(orient="index"))
conf_by_member = (S.groupby(["member", "region_confidence"]).n.sum()
                  .unstack(fill_value=0).to_dict(orient="index"))

keys = [
    keyinfo("source", "categorical", sorted(S.member.unique()),
            "数据来源（数据集/片段集合）", True,
            "同名 source 在 counts_v4 与 steinmetz 中是同一批记录的不同编码"),
    keyinfo("species", "categorical", sorted(S.member.unique()),
            "物种", True, "只有 3 个值；EEG 与侵入式共用 homo_sapiens"),
    keyinfo("modality", "categorical", sorted(S.member.unique()),
            "记录模态", True, "scalp_eeg 与 intracortical 在 L2 的可用不变量不同"),
    keyinfo("region", "anatomical_node", members_with_region,
            "解剖区（偏序节点，不是层级）", "partial",
            "三套词表（Allen / Brodmann / FlyWire super_class）不能直接相等；"
            "且同一列混了五种证据强度，跨成员比较必须先用 region_confidence 过滤",
            vocab_rel="multi_vocabulary",
            extra={"region_confidence_by_level": {lv: {k: int(v) for k, v in d.items()}
                                            for lv, d in conf_by_level.items()},
                   "region_confidence_by_member": {m: {k: int(v) for k, v in d.items()}
                                                   for m, d in conf_by_member.items()},
                   "mapping_status_raw": (mi["mapping_status"].value_counts().to_dict()
                                          if "mapping_status" in mi.columns else None),
                   "region_assignment": {
                       "note": "标签的归属粒度；与 region_confidence（映射质量）是两个维度",
                       "counts": {k: int(v) for k, v in
                                  S.groupby("region_assignment").n.sum().items()},
                       "session_majority_accuracy": 0.271,
                       "session_majority_evidence": "39 个 session 的主导区占比中位 26.6%；"
                                                   "单元加权正确率 27.1%（其余 72.9% 的单元被归到错误区）；"
                                                   "仅 9/39 个 session 的占比 ≥30%，0/39 ≥70%"},
                   "recommended_filter": "跨物种/跨成员比较只用 region_confidence == 'anatomical'；"
                                          "'inferred' 与 'compound' 必须在图注中单列；region_assignment == 'session_majority' 的数据不可用于区级分析"}),
    keyinfo("domain_l1", "functional_node", members_with_domain,
            "功能域（multi 是 6 类 L1；小鼠 units 表是 13 个 functional_domain）", "partial",
            "两套词表：13 个 functional_domain 未与 6 个 L1 域对齐；与 region 也不可交换",
            vocab_rel="multi_vocabulary"),
    dict(key="unit", kind="instance", members=["steinmetz", "primate_macaque_nlb",
                                               "primate_dmfc_rsg", "counts_v4_ragged"],
         joinable_across_members=False,
         semantics="单个神经元/通道",
         caveat="跨成员的 unit 列不是同一个神经元；禁止拼接或按列对齐"),
    dict(key="window", kind="sampling", members=["omnibrain_multi", "omnibrain_spike_view"],
         joinable_across_members=False,
         semantics="100 ms × 8 bin 的滑动窗",
         caveat="窗是派生采样单位，不同成员的 stride/长度不同，不能当样本键"),
    dict(key="trial", kind="instance", members=["primate_human_eeg", "omnibrain4d_eeg_test"],
         joinable_across_members=False,
         semantics="试次",
         caveat="不同数据集试次编号无共同含义"),
]

axes = ["source", "region", "domain_l1"]
d = mi.dropna(subset=["region"])
obs = d.groupby(axes, dropna=False).ngroups
theo = int(np.prod([d[a].nunique() for a in axes]))
holes = dict(axes=axes, observed=int(obs), theoretical=int(theo),
             empty_fraction=round(1 - obs / theo, 3) if theo else None,
             n_per_axis={a: int(d[a].nunique()) for a in axes},
             source="omnibrain-multi/index_timeseries.parquet")

joins = dict(schema_version="1.1",
             note="L1 只放结构键与覆盖计数，不含活动数据；join 合法性按语义判定，不按列名",
             keys=keys, holes=holes,
             levels={lvl: int(g.member.nunique()) for lvl, g in S.groupby("level")},
             members=notes)
(OUT / "joins.json").write_text(json.dumps(joins, indent=2, ensure_ascii=False),
                                encoding="utf-8")

print(f"structure rows={len(S)}")
print("confidence by level:",
      pd.DataFrame(conf_by_level).fillna(0).astype(int).to_string())
print("\nby member:")
print(conf_by_member and pd.DataFrame(conf_by_member).fillna(0).astype(int).to_string())
print("\nholes:", holes["empty_fraction"], holes["observed"], "/", holes["theoretical"])

Xet Storage Details

Size:: 10.7 kB
Xet hash:: 44d714aa2fc46edea9cbc6e31578133099a9f3f4b705f02d52f2e226116c4084

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

	#!/usr/bin/env python3
	"""build_l1_structure — L1 结构层落地（v2：region 分五级可信度）。

	region 不是一个可比的键，而是五种不同强度的证据。分级规则（按 mapping_status）：

	exact / exact_single_species -> anatomical 解剖定位，且有本体节点
	uberon_missing -> anatomical_unmapped 解剖定位，但无 UBERON 节点
	task_code_reinterpreted -> inferred 由任务码推定，不是解剖定位
	compound -> compound 一个标签覆盖多个区（如 M1_PMd）
	coarse / rollup -> coarse 粗粒度或上卷
	unassigned / 缺失 -> none 无解剖标签

	产出 structure.parquet（带 region_confidence）与 joins.json（带各级计数与建议用法）。
	"""
	from __future__ import annotations

	import json
	from pathlib import Path

	import numpy as np
	import pandas as pd

	ROOT = Path("/domains")
	OUT = ROOT / "framework"
	rows = []
	notes = {}

	CONF = {
	"exact": "anatomical",
	"exact_single_species": "anatomical",
	"uberon_missing": "anatomical_unmapped",
	"task_code_reinterpreted": "inferred",
	"compound": "compound",
	"coarse": "coarse",
	"rollup": "coarse",
	"unassigned": "none",
	}


	def conf_series(s):
	return s.map(CONF).fillna("none")


	# 粗标签解析到父区（与 counts-v4 既有映射一致，不引入新精度）
	RESOLVE = {"DMFC": "PFC", "Area2": "S1"}


	def add(df, level, member, provenance, conf=None):
	d = df.copy()
	for c in ["source", "species", "modality", "region", "region_kind", "domain_l1"]:
	if c not in d.columns:
	d[c] = None
	d["region_resolved"] = d["region"].map(
	lambda r: RESOLVE.get(str(r), r) if pd.notna(r) else r)
	d["region_resolution"] = d["region"].map(
	lambda r: "parent_rollup" if str(r) in RESOLVE else "as_labeled")
	if member == "steinmetz":
	d["region_assignment"] = "per_unit"
	elif member == "counts_v4":
	d["region_assignment"] = "per_segment"
	elif member == "flywire":
	d["region_assignment"] = "per_node"
	elif member == "omnibrain_multi":
	d["region_assignment"] = np.where(
	d["source"].astype(str) == "mouse_steinmetz", "session_majority", "window_area")
	else:
	d["region_assignment"] = "unknown"
	if isinstance(conf, str) and conf in d.columns:
	d["region_confidence"] = conf_series(d[conf])
	else:
	d["region_confidence"] = conf or "none"
	g = (d.groupby(["source", "species", "modality", "region", "region_resolved",
	"region_resolution", "region_assignment", "region_kind",
	"region_confidence", "domain_l1"], dropna=False)
	.size().reset_index(name="n"))
	g["level"] = level
	g["member"] = member
	g["provenance"] = provenance
	rows.append(g[["level", "member", "source", "species", "modality", "region",
	"region_resolved", "region_resolution", "region_assignment",
	"region_kind", "region_confidence", "domain_l1", "n", "provenance"]])
	notes[member] = {"rows_in": int(len(d)), "groups": int(len(g)),
	"confidence": d.region_confidence.value_counts().to_dict()}


	# 1) 小鼠：单元级，Allen 缩写，属解剖定位
	u = pd.read_parquet(ROOT / "steinmetz/units.parquet")
	u["source"] = "mouse_steinmetz"
	u["species"] = "mus_musculus"
	u["modality"] = "extracellular_spike"
	u["domain_l1"] = u["functional_domain"]
	u["region_kind"] = "anatomy"
	add(u, "unit", "steinmetz", "steinmetz/units.parquet", conf="anatomical")

	# 2) counts-v4：片段级，带 mapping_status
	ci = pd.read_parquet(ROOT / "counts-v4/counts_index.parquet")
	add(ci, "segment", "counts_v4", "counts-v4/counts_index.parquet", conf="mapping_status")

	# 3) omnibrain-multi：窗级标签，带 mapping_status
	mi = pd.read_parquet(ROOT / "omnibrain-multi/index_timeseries.parquet")
	for c in ["domain_l1", "region_kind", "mapping_status"]:
	if c not in mi.columns:
	mi[c] = None
	add(mi, "window", "omnibrain_multi", "omnibrain-multi/index_timeseries.parquet",
	conf="mapping_status")

	# 4) FlyWire：super_class 属粗粒度分组
	fw = pd.read_parquet(ROOT / "flywire/node_features.parquet")
	fw["source"] = "flywire"
	fw["species"] = "drosophila_melanogaster"
	fw["modality"] = "connectome"
	fw["region_kind"] = "anatomy"
	fw["domain_l1"] = None
	add(fw, "node", "flywire", "flywire/node_features.parquet", conf="coarse")

	S = pd.concat(rows, ignore_index=True)
	S = S[S.region.notna() \| S.domain_l1.notna()]
	S.to_parquet(OUT / "structure.parquet", index=False)
	S.to_csv(OUT / "structure.csv", index=False)


	def keyinfo(key, kind, members, semantics, joinable, caveat, extra=None,
	vocab_rel="shared"):
	per = {}
	for m in members:
	sub = S[(S.member == m) & S[key].notna()]
	per[m] = sorted(set(sub[key].astype(str)))
	union = set().union(*per.values()) if per else set()
	rec = dict(key=key, kind=kind, members=members, n_distinct_union=len(union),
	n_distinct_by_member={m: len(v) for m, v in per.items()},
	vocabulary_relation=vocab_rel,
	values_by_member={m: v[:25] for m, v in per.items()},
	joinable_across_members=joinable,
	semantics=semantics, caveat=caveat)
	if extra:
	rec.update(extra)
	return rec


	members_with_region = sorted(S[S.region.notna()].member.unique())
	members_with_domain = sorted(S[S.domain_l1.notna()].member.unique())
	conf_by_level = (S.groupby(["level", "region_confidence"]).n.sum()
	.unstack(fill_value=0).to_dict(orient="index"))
	conf_by_member = (S.groupby(["member", "region_confidence"]).n.sum()
	.unstack(fill_value=0).to_dict(orient="index"))

	keys = [
	keyinfo("source", "categorical", sorted(S.member.unique()),
	"数据来源（数据集/片段集合）", True,
	"同名 source 在 counts_v4 与 steinmetz 中是同一批记录的不同编码"),
	keyinfo("species", "categorical", sorted(S.member.unique()),
	"物种", True, "只有 3 个值；EEG 与侵入式共用 homo_sapiens"),
	keyinfo("modality", "categorical", sorted(S.member.unique()),
	"记录模态", True, "scalp_eeg 与 intracortical 在 L2 的可用不变量不同"),
	keyinfo("region", "anatomical_node", members_with_region,
	"解剖区（偏序节点，不是层级）", "partial",
	"三套词表（Allen / Brodmann / FlyWire super_class）不能直接相等；"
	"且同一列混了五种证据强度，跨成员比较必须先用 region_confidence 过滤",
	vocab_rel="multi_vocabulary",
	extra={"region_confidence_by_level": {lv: {k: int(v) for k, v in d.items()}
	for lv, d in conf_by_level.items()},
	"region_confidence_by_member": {m: {k: int(v) for k, v in d.items()}
	for m, d in conf_by_member.items()},
	"mapping_status_raw": (mi["mapping_status"].value_counts().to_dict()
	if "mapping_status" in mi.columns else None),
	"region_assignment": {
	"note": "标签的归属粒度；与 region_confidence（映射质量）是两个维度",
	"counts": {k: int(v) for k, v in
	S.groupby("region_assignment").n.sum().items()},
	"session_majority_accuracy": 0.271,
	"session_majority_evidence": "39 个 session 的主导区占比中位 26.6%；"
	"单元加权正确率 27.1%（其余 72.9% 的单元被归到错误区）；"
	"仅 9/39 个 session 的占比 ≥30%，0/39 ≥70%"},
	"recommended_filter": "跨物种/跨成员比较只用 region_confidence == 'anatomical'；"
	"'inferred' 与 'compound' 必须在图注中单列；region_assignment == 'session_majority' 的数据不可用于区级分析"}),
	keyinfo("domain_l1", "functional_node", members_with_domain,
	"功能域（multi 是 6 类 L1；小鼠 units 表是 13 个 functional_domain）", "partial",
	"两套词表：13 个 functional_domain 未与 6 个 L1 域对齐；与 region 也不可交换",
	vocab_rel="multi_vocabulary"),
	dict(key="unit", kind="instance", members=["steinmetz", "primate_macaque_nlb",
	"primate_dmfc_rsg", "counts_v4_ragged"],
	joinable_across_members=False,
	semantics="单个神经元/通道",
	caveat="跨成员的 unit 列不是同一个神经元；禁止拼接或按列对齐"),
	dict(key="window", kind="sampling", members=["omnibrain_multi", "omnibrain_spike_view"],
	joinable_across_members=False,
	semantics="100 ms × 8 bin 的滑动窗",
	caveat="窗是派生采样单位，不同成员的 stride/长度不同，不能当样本键"),
	dict(key="trial", kind="instance", members=["primate_human_eeg", "omnibrain4d_eeg_test"],
	joinable_across_members=False,
	semantics="试次",
	caveat="不同数据集试次编号无共同含义"),
	]

	axes = ["source", "region", "domain_l1"]
	d = mi.dropna(subset=["region"])
	obs = d.groupby(axes, dropna=False).ngroups
	theo = int(np.prod([d[a].nunique() for a in axes]))
	holes = dict(axes=axes, observed=int(obs), theoretical=int(theo),
	empty_fraction=round(1 - obs / theo, 3) if theo else None,
	n_per_axis={a: int(d[a].nunique()) for a in axes},
	source="omnibrain-multi/index_timeseries.parquet")

	joins = dict(schema_version="1.1",
	note="L1 只放结构键与覆盖计数，不含活动数据；join 合法性按语义判定，不按列名",
	keys=keys, holes=holes,
	levels={lvl: int(g.member.nunique()) for lvl, g in S.groupby("level")},
	members=notes)
	(OUT / "joins.json").write_text(json.dumps(joins, indent=2, ensure_ascii=False),
	encoding="utf-8")

	print(f"structure rows={len(S)}")
	print("confidence by level:",
	pd.DataFrame(conf_by_level).fillna(0).astype(int).to_string())
	print("\nby member:")
	print(conf_by_member and pd.DataFrame(conf_by_member).fillna(0).astype(int).to_string())
	print("\nholes:", holes["empty_fraction"], holes["observed"], "/", holes["theoretical"])