Programmatic Access

Install dependencies

# Run once (comment out after first execution)
!pip install caveclient cloud-volume

import caveclient
from caveclient import CAVEclient, chunkedgraph as cg
from cloudvolume import CloudVolume
import numpy as np
import pandas as pd

Authenticate

Run the cell below once — it opens a browser tab. Copy the string labelled token and paste it into the second cell.

global_url = "https://global.brain-wire-test.org/"
auth = caveclient.auth.AuthClient(server_address=global_url)

# Opens a browser page — copy the token string shown there
auth.setup_token(make_new=False)

# Paste your token below and run this cell
auth.save_token(token="YOUR_TOKEN_HERE", overwrite=True)

Connect to the Fish1 server

global_url = "https://global.brain-wire-test.org/"
local_url  = "https://pcgv3local.brain-wire-test.org"
datastack  = "fish1_full"
dataset    = "fish1_v250915"

client = CAVEclient(datastack_name=datastack, server_address=global_url)
client.materialize.version = client.materialize.most_recent_version()
print(f"Using materialization version: {client.materialize.version}")

cggraph = cg.ChunkedGraphClient(
    server_address=local_url,
    table_name=dataset,
    auth_client=caveclient.auth.AuthClient(token=client.auth.token),
)

cv = client.info.segmentation_cloudvolume()
print("Connected successfully.")

All annotations are stored in versioned, queryable tables accessed via client.materialize.

all_tables = client.annotation.get_tables()
print(all_tables)

client.materialize.tables.somas().query(limit=10)

By 3-D coordinate

Given a voxel coordinate (e.g. from Neuroglancer), retrieve the supervoxel and root ID and confirm it is the most recent root.

def get_latest_root_id(x, y, z, resolution=(8, 8, 30)):
    """Return (supervoxel_id, root_id) for a voxel coordinate."""
    supervoxel_id = cv.download_point(
        pt=(x, y, z), size=1, agglomerate=False, coord_resolution=resolution
    )
    supervoxel_id = np.int64(supervoxel_id[0, 0, 0, 0])
    print(f"supervoxel_id : {supervoxel_id}")

    root_id = cggraph.get_root_id(supervoxel_id)
    print(f"root_id       : {root_id}")

    if not cggraph.is_latest_roots([root_id]):
        root_id = cggraph.get_latest_roots(root_id)
        print(f"updated root  : {root_id}  (segment was edited)")
    else:
        print("root_id is current")

    return supervoxel_id, root_id

supervoxel_id, root_id = get_latest_root_id(75765, 38366, 6682)

By lore ID

lore_id = "173502"  # ← change this

row = client.materialize.tables.somas(id=lore_id).query().iloc[0]

root_id    = int(row["pt_root_id"])
pt_vox     = row["pt_position"]
cell_type  = row["cell_type"]
type_label = {"exc": "excitatory", "inh": "inhibitory"}.get(cell_type, "unknown")

print(f"Lore ID   : {lore_id}")
print(f"Root ID   : {root_id}")
print(f"Position  : {pt_vox}  (voxels @ 16×16×30 nm)")
print(f"Cell type : {type_label}")

Root ID → lore ID (reverse lookup)

row = client.materialize.tables.somas(pt_root_id=root_id).query()

if len(row):
    print(f"Root {root_id} → lore ID {row.iloc[0]['id']}")
else:
    print(f"No soma annotation found for root {root_id}")

Get all supervoxels for a root ID

def get_supervoxels(root_id, client):
    return client.chunkedgraph.get_leaves(root_id, stop_layer=1)

svids = get_supervoxels(root_id, client)
print(f"Root {root_id} → {len(svids)} supervoxels")

Synapse tables encode axon-to-dendrite connections with partner root IDs, spatial positions, and excitatory/inhibitory labels (tag='1' = inhibitory, tag='2' = excitatory).

Fetch incoming and outgoing synapses

lore_id = "173502"  # ← change this

row     = client.materialize.tables.somas(id=lore_id).query().iloc[0]
root_id = int(row["pt_root_id"])

incoming = client.materialize.tables.synapses_axde_label(post_pt_root_id=root_id).query()
inh_in   = incoming[incoming["tag"] == "1"]
exc_in   = incoming[incoming["tag"] == "2"]
outgoing = client.materialize.tables.synapses_axde_label(pre_pt_root_id=root_id).query()

print(f"Incoming synapses : {len(incoming):>5}  ({len(inh_in)} inhibitory, {len(exc_in)} excitatory)")
print(f"  unique pre-synaptic partners : {incoming['pre_pt_root_id'].nunique()}")
print(f"Outgoing synapses : {len(outgoing):>5}")

Assemble a full synapse table with size IDs and e/i labels

The full synapse record is spread across three tables. The cell below joins them and exports to CSV.

pre_ids  = client.materialize.tables.synapses_axde_pre_synapse_id(
               id=incoming.id.values.tolist()).query()[["id", "tag"]]
post_ids = client.materialize.tables.synapses_axde_post_synapse_id(
               id=incoming.id.values.tolist()).query()[["id", "tag"]]

merged = (
    incoming
    .merge(pre_ids,  on="id", suffixes=("", "_pre"))
    .merge(post_ids, on="id", suffixes=("", "_post"))
)

synapse_table = merged[
    ["pre_pt_position", "tag_pre", "post_pt_position", "tag_post", "tag"]
].rename(columns={"tag_pre": "pre_site_id", "tag_post": "post_site_id", "tag": "ei_label"})

print(f"Merged table: {len(synapse_table)} rows")
synapse_table.to_csv(f"synapses_{lore_id}_{root_id}.csv", index=False)

Filter synapses by physical size

Each synapse has a bounding box in synapses_axon_to_dendrite_size. Use this to remove small or spurious contacts.

syn_sizes = client.materialize.tables.synapses_axon_to_dendrite_size(
    tag=pre_ids.tag.values.tolist(),
    tag2=post_ids.tag.values.tolist(),
).query()

bbox_start = np.vstack(syn_sizes["pt_position"].values)
bbox_end   = np.vstack(syn_sizes["pt2_position"].values)
dims       = np.abs(bbox_start - bbox_end)
syn_sizes[["size_x", "size_y", "size_z"]] = dims
syn_sizes["volume"] = dims[:, 0] * dims[:, 1] * dims[:, 2]

by_volume = syn_sizes[syn_sizes["volume"] > 1000]
by_dims   = syn_sizes[(syn_sizes["size_x"] > 10) & (syn_sizes["size_y"] > 10) & (syn_sizes["size_z"] > 10)]

print(f"All synapses         : {len(syn_sizes)}")
print(f"After volume filter  : {len(by_volume)}")
print(f"After per-axis filter: {len(by_dims)}")

Skeletons are available in SWC format and can be loaded into tools such as navis or NEURON.

lore_id = "173502"  # ← change this

row     = client.materialize.tables.somas(id=lore_id).query().iloc[0]
root_id = int(row["pt_root_id"])

skel_vol = CloudVolume(
    "precomputed://middleauth+"
    "https://pcgv3local.brain-wire-test.org"
    "/skeletoncache/api/v1/fish1_full/precomputed/"
)
skeleton = skel_vol.skeleton.get(root_id)

out_file = f"skeleton_{lore_id}_{root_id}.swc"
with open(out_file, "w") as f:
    f.write(skeleton.to_swc())
print(f"Saved: {out_file}")
print(skeleton)

synapses = client.materialize.tables.synapses_axde_label(post_pt_root_id=root_id).query()
inh_syn  = synapses[synapses["tag"] == "1"]
exc_syn  = synapses[synapses["tag"] == "2"]

inh_syn.to_csv(f"inh_synapses_{lore_id}_{root_id}.csv", index=False)
exc_syn.to_csv(f"exc_synapses_{lore_id}_{root_id}.csv", index=False)
print(f"Lore {lore_id}: {len(synapses)} total synapses ({len(inh_syn)} inhibitory, {len(exc_syn)} excitatory)")

The function below retrieves all segments that have received at least one proofreading edit (merge or split), with edit counts and soma coordinates. Use limit to cap the number fetched — the dataset has over 42,000 edited segments in total. See also the Proofreading page.

from datetime import datetime, timezone

def get_proofread_segments(client, since=None, limit=None):
    ts_start = since or datetime(2020, 1, 1, tzinfo=timezone.utc)
    ts_end   = datetime.now(tz=timezone.utc)

    _, new_roots = client.chunkedgraph.get_delta_roots(ts_start, ts_end)
    edited_roots = np.unique(new_roots).tolist()
    print(f"Found {len(edited_roots)} edited segments total")

    if limit:
        edited_roots = edited_roots[:limit]
        print(f"Fetching first {limit}")

    BATCH, all_logs = 200, {}
    for i in range(0, len(edited_roots), BATCH):
        batch = edited_roots[i : i + BATCH]
        all_logs.update(client.chunkedgraph.get_tabular_change_log(batch, filtered=False))
        print(f"  fetched {i + len(batch)} / {len(edited_roots)}")

    frames = []
    for rid, df in all_logs.items():
        if df.empty:
            continue
        df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
        frames.append(pd.DataFrame({
            "root_id":       rid,
            "n_merges":      int(df["is_merge"].sum()),
            "n_splits":      int((~df["is_merge"]).sum()),
            "n_total_edits": len(df),
            "first_edit":    df["timestamp"].min(),
            "last_edit":     df["timestamp"].max(),
        }, index=[0]))

    if not frames:
        return pd.DataFrame()

    edit_summary = pd.concat(frames, ignore_index=True)

    soma_locs = client.materialize.query_table(
        "somas",
        filter_in_dict={"pt_root_id": edit_summary["root_id"].tolist()},
        select_columns=["pt_root_id", "pt_position"],
    ).rename(columns={"pt_root_id": "root_id"})
    soma_locs[["x", "y", "z"]] = pd.DataFrame(soma_locs["pt_position"].tolist(), index=soma_locs.index)
    soma_locs.drop(columns=["pt_position"], inplace=True)

    return (
        edit_summary.merge(soma_locs, on="root_id", how="left")
        .sort_values("n_total_edits", ascending=False)
        .reset_index(drop=True)
    )

df = get_proofread_segments(client, limit=500)

print("=" * 50)
print("PROOFREADING SUMMARY")
print("=" * 50)
print(f"  Total edited segments  : {len(df)}")
print(f"  Total merges           : {df.n_merges.sum()}")
print(f"  Total splits           : {df.n_splits.sum()}")
print(f"  Total operations       : {df.n_total_edits.sum()}")
print(f"  Segments with >10 edits: {(df.n_total_edits > 10).sum()}")
print("=" * 50)
print(df.head(10))

# df.to_csv("proofread_segments.csv", index=False)