Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
5676685
feat: add visualize command and nested extraction tests
cattabiani May 11, 2026
1a90879
Fix NaN in external id mapping when no overlap exists
cattabiani May 11, 2026
fb2b6c2
Fix external nodes file missing merged IDs from multiple edges
cattabiani May 11, 2026
50f9e8a
Skip external_ populations in _write_subcircuit_virtual
cattabiani May 11, 2026
e152fcd
Refactor: split _write_subcircuit_virtual into _filter_virtual_typed_…
cattabiani May 11, 2026
a398e70
Merge _make_parent_the_original_mapping and _add_mapping_to_original …
cattabiani May 11, 2026
bb657e4
Unify write path: _filter_virtual_typed_subcircuit and _gather_new_ex…
cattabiani May 11, 2026
2436f24
Merge external populations in nested subcircuit extraction
cattabiani May 11, 2026
6ab4231
Combine original_id into single list, drop original2_id
cattabiani May 11, 2026
4d7ccf7
Add D-level equivalence checks and refactor into _assert_equivalent h…
cattabiani May 11, 2026
e6749c4
Single-pass multi-input edge writing, remove multi-pass workarounds
cattabiani May 11, 2026
36691bb
Refactor visualize: replace _classify_population with PopulationType …
cattabiani May 12, 2026
8071171
Improve test comments: document node fates per subset, use _assert_ci…
cattabiani May 12, 2026
b8f55a5
Rename test circuits to numbered scheme (c1, c2_c1, ...) to avoid con…
cattabiani May 12, 2026
19fca5e
Move output file ownership into _copy_filtered_edges, remove _get_nod…
cattabiani May 12, 2026
b6ac4c3
Move lint to Python 3.12 runner (ruff dropped 3.10 support)
cattabiani May 12, 2026
8f3d896
Format visualize.py (ruff)
cattabiani May 12, 2026
2a4574f
Inline _write_indexes: remove trivial wrapper, call libsonata directly
cattabiani May 12, 2026
5d30483
Update tasks.md, eliminate node_pop_name_mapping_secondary, add docst…
cattabiani May 12, 2026
25aab5d
WIP: decouple _gather from _filter state (task 17)
cattabiani May 12, 2026
6f47e9c
Decouple _gather from _filter: pass per-population offset, remove copies
cattabiani May 12, 2026
85b09b0
Clean up _gather: remove redundant mkdir, fix misleading debug log
cattabiani May 13, 2026
03b101e
Simplify: remove unnecessary if-guard on offset addition
cattabiani May 13, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ build
.tox
.idea
.vscode
.kiro
venv
dist
brainbuilder/version.py
19 changes: 19 additions & 0 deletions brainbuilder/app/sonata.py
Original file line number Diff line number Diff line change
Expand Up @@ -449,3 +449,22 @@ def resize_datatypes(file, population_name, population_type, attributes):

click.secho(f"The following updates were performed:\n{updates}")
click.secho(f"One should run `h5repack {file} output.h5` to repack the file", fg="green")


@app.command()
@click.argument("circuit-config", type=REQUIRED_PATH)
@click.option("-o", "--output", help="Save to file (e.g., circuit.png). Otherwise opens viewer.")
@click.option("-t", "--title", help="Title for the graph (default: parent directory name)")
def visualize(circuit_config, output, title):
"""Display a graph of circuit connectivity grouped by population.

Requires: pip install brainbuilder[viz] and system graphviz.
"""
from brainbuilder.utils.sonata.visualize import draw_circuit

if not title:
title = Path(circuit_config).parent.name

draw_circuit(circuit_config, output_path=output, title=title)
if output:
click.echo(f"Saved to {output}")
717 changes: 443 additions & 274 deletions brainbuilder/utils/sonata/split_population.py

Large diffs are not rendered by default.

167 changes: 167 additions & 0 deletions brainbuilder/utils/sonata/visualize.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
# SPDX-License-Identifier: Apache-2.0
"""Visualize a SONATA circuit as a graph with populations as clusters."""

import json
from collections import Counter
from enum import StrEnum
from pathlib import Path

import bluepysnap
import h5py

# Populations with more nodes than this threshold are shown as a single
# cluster node with population-level edges instead of individual nodes.
_MAX_NODES_DETAILED = 10


class PopulationType(StrEnum):
"""Classification of a node population for visualization purposes."""

BIOPHYSICAL = "biophysical"
VIRTUAL = "virtual"
EXTERNAL = "external"

@property
def color(self) -> str:
return {
PopulationType.BIOPHYSICAL: "lightyellow",
PopulationType.VIRTUAL: "lightblue",
PopulationType.EXTERNAL: "lightsalmon",
}.get(self, "white")

@classmethod
def from_population(cls, pop_name: str, pop_type: str | None) -> "PopulationType":
if pop_name.startswith("external_"):
return cls.EXTERNAL
if pop_type == "virtual":
return cls.VIRTUAL
return cls.BIOPHYSICAL


def _load_id_mapping(circuit_config_path):
"""Load id_mapping from circuit provenance if available.

Returns:
dict: pop_name -> list of parent_ids, or None if no mapping exists.
"""
config = json.loads(Path(circuit_config_path).read_text())
mapping_path = config.get("components", {}).get("provenance", {}).get("id_mapping")
if not mapping_path:
return None

mapping_file = Path(circuit_config_path).parent / mapping_path
if not mapping_file.exists():
return None

mapping = json.loads(mapping_file.read_text())
return mapping


def draw_circuit(
circuit_config_path, output_path=None, max_nodes_detailed=_MAX_NODES_DETAILED, title=None
):
"""Draw a SONATA circuit using graphviz with populations as clusters.

For small populations (<=max_nodes_detailed), individual nodes and edges
are shown. For large populations, a single summary node is shown with
population-level edges. Duplicate edges are collapsed with a count label.

If an id_mapping exists in provenance, node labels show the parent (original)
IDs instead of the local IDs. External populations get a distinct color.

Args:
circuit_config_path: Path to circuit_config.json.
output_path: If provided, save the rendered image to this path.
Otherwise, render to a temp file and open it.
max_nodes_detailed: Populations with more nodes than this are shown
as a single summary node.

Requires:
pip install brainbuilder[viz]
System graphviz: brew install graphviz (macOS) / apt install graphviz (Linux)
"""
try:
import graphviz
except ImportError as e:
raise ImportError(
"graphviz Python package is required for visualization. "
"Install with: pip install brainbuilder[viz]\n"
"Also requires system graphviz: brew install graphviz"
) from e

circuit = bluepysnap.Circuit(str(circuit_config_path))
id_mapping = _load_id_mapping(circuit_config_path)

dot = graphviz.Digraph("circuit", format="png")
dot.attr(rankdir="LR")
dot.attr("node", shape="circle", fontsize="9", width="0.3", height="0.3")
if title:
dot.attr(label=title, labelloc="t", fontsize="14")

detailed_pops = set()

for pop_name, pop in circuit.nodes.items():
pop_type = PopulationType.from_population(pop_name, pop.type)

# Get original IDs for labels if mapping exists
parent_ids = None
if id_mapping and pop_name in id_mapping:
entry = id_mapping[pop_name]
parent_ids = entry.get("original_id", entry.get("parent_id"))

if pop.size <= max_nodes_detailed:
detailed_pops.add(pop_name)
with dot.subgraph(name=f"cluster_{pop_name}") as sub:
sub.attr(
label=f"{pop_name} ({pop_type}, {pop.size})",
style="filled",
color=pop_type.color,
)
prev = None
for i in range(pop.size):
label = str(parent_ids[i]) if parent_ids else str(i)
sub.node(f"{pop_name}__{i}", label=f"<<B>{label}</B>>")
if prev is not None:
sub.edge(prev, f"{pop_name}__{i}", style="invis", weight="10")
prev = f"{pop_name}__{i}"
else:
dot.node(
f"{pop_name}__summary",
label=f"{pop_name}\n({pop_type}, {pop.size})",
shape="box",
style="filled",
fillcolor=pop_type.color,
)

# Edges — group duplicates and show count
for edge_name, edge in circuit.edges.items():
src_name = edge.source.name
tgt_name = edge.target.name
src_detailed = src_name in detailed_pops
tgt_detailed = tgt_name in detailed_pops

with h5py.File(edge.h5_filepath, "r") as h5:
sgids = h5[f"edges/{edge_name}/source_node_id"][:]
tgids = h5[f"edges/{edge_name}/target_node_id"][:]

if src_detailed and tgt_detailed:
edge_counts = Counter(zip(sgids.tolist(), tgids.tolist()))
for (s, t), count in edge_counts.items():
attrs = {}
if count > 1:
attrs["label"] = str(count)
attrs["fontsize"] = "8"
dot.edge(f"{src_name}__{s}", f"{tgt_name}__{t}", **attrs)
else:
src_node = f"{src_name}__summary" if not src_detailed else f"{src_name}__{sgids[0]}"
tgt_node = f"{tgt_name}__summary" if not tgt_detailed else f"{tgt_name}__{tgids[0]}"
dot.edge(src_node, tgt_node, label=str(len(sgids)), fontsize="8")

if output_path:
dot.render(outfile=output_path, cleanup=True)
else:
import tempfile

filename = title.replace(" ", "_") if title else "circuit"
filepath = Path(tempfile.gettempdir()) / filename
dot.render(filename=str(filepath), view=True, cleanup=True)
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ include = ["brainbuilder*"]
[project.optional-dependencies]
all = [] # for compatibility
reindex = [] # for compatibility
viz = ["graphviz"]

[project.urls]
Homepage = "https://openbraininstitute/brainbuilder"
Expand Down
Loading
Loading