Scene composition (connectomekg.scene)

connectomekg.scene

connectomekg/scene.py

Builds the real-geometry 3-D views of a connectome: view A, every neuron's marked point as a dim context cloud, and view B, the skeletons of a query's neurons drawn at full brightness inside it, and view C, neuropils linked by the signal flow their neurons carry between them. Both views can also draw the neuropil surface meshes connkg meshes caches (:mod:connectomekg.neuropil_meshes). Unlike the fleet's other viz3d consumers (gutenberg_kg, pycode_kg, genealogy_kg), this graph already has space -- every neuron carries real x/y/z coordinates and the download holds a traced skeleton for most of them -- so there is no tree to grow here, no kg_utils.viz3d.organic call anywhere in this file. What is reused from that engine is just the camera rule (frame_tree, in cli/cmd_viz3d.py) and seed_from_key for a stable per-type color.

Split the way pycode_kg.scene3d and genealogy_kg.scene split their own layout from composition, so most of this is testable without PyVista:

• :class:WorldFrame, :func:world_frame, :func:context_points, :func:circuit_neurons, :func:neuropil_flow and :func:flow_arc are pure NumPy plus SQL -- no PyVista import.
• :func:build_brain_scene composes those into a caller-supplied pv.Plotter. This half needs the viz3d extra.

Colors come from :mod:connectomekg.colors, never from :mod:connectomekg.viz -- viz.py imports plotly at module scope, and this module must stay importable with only the viz3d extra installed, no viz. The same reason genealogy_kg keeps its palette in theme.py rather than viz.py.

NeuronGroup(label, neuron_ids, color) dataclass

Neurons drawn together in one color, under one label.

The circuit view otherwise groups by cell type and colors by name, which is right when the question is "show me LC4" and wrong when it is "show me the answer": a path's hops are an order, and the neurons in one hop rarely share a type. A group lets the caller say what belongs together.

Parameters:

Name	Type	Description	Default
label	str	Names the group in the scene's title and its actor names.	required
neuron_ids	Sequence[str]	Neuron node ids in this group.	required
color	str	#RRGGBB for every neuron in it.	required

NeuropilFlow(names, bases, centroids_nm, n_synapses, pairs) dataclass

Neuropil positions and the signal flow between them (view C).

Flow from neuropil A to B is carried by neurons: each neuron's output synapses in B, apportioned by the share of its input synapses that lie in A, summed over neurons, with A = B excluded. See docs/rendering.md, "What flow measures".

Parameters:

Name	Type	Description	Default
names	list[str]	Neuropil abbreviations, e.g. "ME_R", sorted.	required
bases	list[str]	Each neuropil's side-free base, e.g. "ME", same order.	required
centroids_nm	ndarray	(n, 3) synapse-weighted centroid of the marked points of every neuron in each neuropil; a row is NaN when no neuron in it has coordinates.	required
n_synapses	ndarray	(n,) each neuropil's synapse count.	required
pairs	list[tuple[str, str, float]]	(source, target, flow) for every directed pair with nonzero flow, strongest first, ties broken by name.	required

SceneInfo(title, points, n_context, n_circuit, n_skeletons, missing_skeletons, soma_fallbacks, view='circuit', n_flow_pairs=0, n_flow_total=0, n_neuropil_meshes=0, n_context_somas=0, picks=PickTargets.empty()) dataclass

What a composed brain scene contains.

Parameters:

Name	Type	Description	Default
title	str	Human-readable summary for a window title or CLI echo.	required
points	ndarray	Every drawn point, world units, for camera framing via kg_utils.viz3d.frame_tree.	required
n_context	int	Neurons drawn in the context cloud.	required
n_circuit	int	Neurons resolved for the circuit view.	required
n_skeletons	int	Circuit neurons drawn from a loaded skeleton file.	required
missing_skeletons	list[int]	Circuit neurons' root ids with no skeleton file loaded (no data_dir, or the file was absent) -- drawn as a larger fallback sphere at the marked point instead.	required
soma_fallbacks	int	Circuit neurons whose skeleton had no Label 1 row, so the soma sphere sits at the root point instead.	required
view	str	The view composed, one of :data:VIEWS.	'circuit'
n_flow_pairs	int	Flow arcs drawn (view C).	0
n_flow_total	int	Directed neuropil pairs with nonzero flow (view C).	0
n_neuropil_meshes	int	Neuropil surface meshes drawn; 0 when none are cached.	0
n_context_somas	int	Context-cloud neurons drawn at a real soma rather than at FlyWire's marked point; 0 until connkg skeletons has back-filled somas.	0
picks	PickTargets	Every circuit point with the neuron that owns it, so a click in the viewer resolves to a neuron. Empty for the flow view.	empty()

WorldFrame(center, scale=NM_PER_WORLD_UNIT, marker_scale=1.0) dataclass

The nm-to-world-units mapping a scene is framed in.

Parameters:

Name	Type	Description	Default
center	ndarray	(3,) nm; the point that maps to the world origin.	required
scale	float	Nanometres per world unit.	NM_PER_WORLD_UNIT
marker_scale	float	Multiplier on marker sizes (neuropil spheres, flow tubes, soma spheres), so a dataset that frames larger than FAFB draws them the same size on screen; 1 for FAFB. The context cloud is left at FAFB's size: scaled up, its dots speckle the spheres inside it.	1.0

to_world(points_nm)

Map nm points to world units, dorsal up, section axis away from camera.

(x, y, z) nm becomes world ((x - cx) / S, (z - cz) / S, -(y - cy) / S) -- FAFB's y grows ventrally, so negating it after centring puts dorsal at larger world z.

Parameters:

Name	Type	Description	Default
points_nm	ndarray	(n, 3) or (3,) nm coordinates.	required

Returns:

Type	Description
ndarray	(n, 3) world coordinates.

Source code in src/connectomekg/scene.py

def to_world(self, points_nm: np.ndarray) -> np.ndarray:
    """Map nm points to world units, dorsal up, section axis away from camera.

    ``(x, y, z)`` nm becomes world
    ``((x - cx) / S, (z - cz) / S, -(y - cy) / S)`` -- FAFB's ``y`` grows
    ventrally, so negating it after centring puts dorsal at larger world
    ``z``.

    :param points_nm: ``(n, 3)`` or ``(3,)`` nm coordinates.
    :return: ``(n, 3)`` world coordinates.
    """
    pts = np.atleast_2d(np.asarray(points_nm, dtype=np.float64))
    cx, cy, cz = self.center
    x = (pts[:, 0] - cx) / self.scale
    y = (pts[:, 2] - cz) / self.scale
    z = -(pts[:, 1] - cy) / self.scale
    return np.stack([x, y, z], axis=1)

add_floor(plotter, background=BACKGROUND)

Put a shadow-receiving floor under the composed scene, lit from above.

Adds a floor plane (actor floor) a step off the background just below plotter.bounds, replaces the lights with a shadow-casting key spotlight above the scene plus a weak headlight, and enables shadow mapping. The floor is only visible from a camera that looks down on it, so pair it with a non-zero elevation in :func:aim_camera, and call it after framing: the floor is far larger than the scene and would otherwise decide the framing.

Parameters:

Name	Type	Description	Default
plotter	Plotter	Plotter with the scene composed and the camera framed.	required
background	str	The scene's background, which the floor is shaded from (:func:floor_color).	BACKGROUND

Source code in src/connectomekg/scene.py

def add_floor(plotter: pv.Plotter, background: str = BACKGROUND) -> None:
    """Put a shadow-receiving floor under the composed scene, lit from above.

    Adds a floor plane (actor ``floor``) a step off the background just below
    ``plotter.bounds``, replaces the lights with a shadow-casting key
    spotlight above the scene plus a weak headlight, and enables shadow
    mapping. The floor is only visible from a camera that looks down on it,
    so pair it with a non-zero ``elevation`` in :func:`aim_camera`, and call
    it after framing: the floor is far larger than the scene and would
    otherwise decide the framing.

    :param plotter: Plotter with the scene composed and the camera framed.
    :param background: The scene's background, which the floor is shaded
        from (:func:`floor_color`).
    """
    import pyvista as pv  # noqa: PLC0415 - the viz3d-render-only import boundary

    xmin, xmax, ymin, ymax, zmin, zmax = plotter.bounds
    cx, cy = (xmin + xmax) / 2, (ymin + ymax) / 2
    floor = pv.Plane(
        center=(cx, cy, zmin - _FLOOR_DROP),
        direction=(0, 0, 1),
        i_size=_FLOOR_SIZE,
        j_size=_FLOOR_SIZE,
        i_resolution=1,
        j_resolution=1,
    )
    # Darker than the background, not equal to it: a floor the same color as
    # the void behind it shows no horizon and no lit pool, so the shadow has
    # nothing to be a shadow *on*.
    # Culled from behind, so orbiting under the scene does not put an opaque
    # 120-unit plane between the camera and the subject. A floor is a surface
    # to stand on, not a wall, and it has no underside worth seeing.
    plotter.add_mesh(
        floor,
        color=floor_color(background),
        ambient=0.12,
        diffuse=0.95,
        specular=0.0,
        culling="back",
        name="floor",
    )

    # The three-point rig stays. Only a positional light can cast a shadow in
    # VTK, so one is added for that job -- but replacing the rig with it, as
    # this used to, left everything the narrow cone missed unlit: the neuropil
    # shells at 10 % opacity and the whole-brain cloud simply vanished, which
    # is a poor trade for a shadow. Lighting and shadow-casting are separate
    # jobs here, and the key is dimmer than it was because it is no longer
    # also responsible for lighting the scene.
    add_studio_lighting(plotter)
    dx, dy = _KEY_LIGHT_OFFSET
    key = pv.Light(
        position=(cx + dx, cy + dy, zmax + _KEY_LIGHT_HEIGHT),
        focal_point=(cx, cy, zmin),
        light_type="scene light",
        intensity=_KEY_LIGHT_INTENSITY,
    )
    key.positional = True
    key.cone_angle = _KEY_LIGHT_CONE
    plotter.add_light(key)
    _shade_skeleton_lines(plotter)
    plotter.enable_shadows()  # ty: ignore[missing-argument]
    # PyVista exposes no setter for the shadow map size.
    shadow_pass = plotter.renderer._render_passes._shadow_map_pass
    if shadow_pass is not None:
        shadow_pass.GetShadowMapBakerPass().SetResolution(_SHADOW_MAP_RESOLUTION)
        _draw_translucency_after_shadows(plotter, shadow_pass)

add_studio_lighting(plotter)

Light a floorless scene so its geometry reads as solid.

Replaces PyVista's five on-axis camera lights with three offset ones; see :data:_STUDIO_LIGHTS for why the offset is the whole point and why they stay camera-relative. Skeletons drawn as lines are unaffected either way -- a line has no surface to shade, so --tubes is what makes lighting visible at all on a circuit.

Parameters:

Name	Type	Description	Default
plotter	Plotter	Plotter with the scene composed.	required

Source code in src/connectomekg/scene.py

def add_studio_lighting(plotter: pv.Plotter) -> None:
    """Light a floorless scene so its geometry reads as solid.

    Replaces PyVista's five on-axis camera lights with three offset ones; see
    :data:`_STUDIO_LIGHTS` for why the offset is the whole point and why they
    stay camera-relative. Skeletons drawn as lines are unaffected either way
    -- a line has no surface to shade, so ``--tubes`` is what makes lighting
    visible at all on a circuit.

    :param plotter: Plotter with the scene composed.
    """
    import pyvista as pv  # noqa: PLC0415 - the viz3d-render-only import boundary

    plotter.remove_all_lights()
    for position, intensity in _STUDIO_LIGHTS:
        light = pv.Light(position=position, focal_point=(0.0, 0.0, 0.0), light_type="camera light")
        light.intensity = intensity
        plotter.add_light(light)

aim_camera(plotter, points, *, fov=14.0, elevation=0.0, spec=None)

Point the camera at a composed scene and frame it tightly at that view.

kg_utils.viz3d.frame_tree sets the view direction (from the front, dorsal up), the camera tilts up by elevation degrees so it looks down on the scene, and quiltwright.frame_and_focus then fits the scene at that final direction and puts the focal plane at the harmonic mean of its near and far depths. reset_camera() would fit the un-tilted bounds instead, which leaves a tilted scene small in frame.

frame_and_focus measures plotter.bounds, so call this before :func:add_floor. It also fits to the window's aspect: pass the quilt or still spec you will render, and the window is set to the aspect render_quilt captures each view at (16:9 for 16-landscape, whose tiles are 4:3). Afterwards, pass fov=None to render_quilt and depth_report: the camera is locked.

Parameters:

Name	Type	Description	Default
plotter	Plotter	Plotter with the scene composed.	required
points	ndarray	World points the view direction is computed from, usually SceneInfo.points.	required
fov	float	Vertical field of view to lock the camera to, in degrees.	14.0
elevation	float	Degrees to tilt the camera up from the front view.	0.0
spec	QuiltSpec | None	The quiltwright.QuiltSpec the scene will be rendered at; None frames at the window's current aspect, as an interactive viewer wants.	None

Returns:

Type	Description
tuple[float, float, float]	(near, far, focal_distance) from frame_and_focus.

Source code in src/connectomekg/scene.py

def aim_camera(
    plotter: pv.Plotter,
    points: np.ndarray,
    *,
    fov: float = 14.0,
    elevation: float = 0.0,
    spec: QuiltSpec | None = None,
) -> tuple[float, float, float]:
    """Point the camera at a composed scene and frame it tightly at that view.

    ``kg_utils.viz3d.frame_tree`` sets the view direction (from the front,
    dorsal up), the camera tilts up by *elevation* degrees so it looks down on
    the scene, and ``quiltwright.frame_and_focus`` then fits the scene at that
    final direction and puts the focal plane at the harmonic mean of its near
    and far depths. ``reset_camera()`` would fit the un-tilted bounds instead,
    which leaves a tilted scene small in frame.

    ``frame_and_focus`` measures ``plotter.bounds``, so call this before
    :func:`add_floor`. It also fits to the window's aspect: pass the quilt or
    still *spec* you will render, and the window is set to the aspect
    ``render_quilt`` captures each view at (16:9 for ``16-landscape``, whose
    tiles are 4:3). Afterwards, pass ``fov=None`` to ``render_quilt`` and
    ``depth_report``: the camera is locked.

    :param plotter: Plotter with the scene composed.
    :param points: World points the view direction is computed from,
        usually ``SceneInfo.points``.
    :param fov: Vertical field of view to lock the camera to, in degrees.
    :param elevation: Degrees to tilt the camera up from the front view.
    :param spec: The ``quiltwright.QuiltSpec`` the scene will be rendered at;
        ``None`` frames at the window's current aspect, as an interactive
        viewer wants.
    :return: ``(near, far, focal_distance)`` from ``frame_and_focus``.
    """
    from kg_utils.viz3d import frame_tree  # noqa: PLC0415 - the viz3d-render-only import boundary
    from quiltwright import frame_and_focus  # noqa: PLC0415

    frame = frame_tree(points, fov=fov)
    camera = plotter.camera
    camera.position = frame.position
    camera.focal_point = frame.focal_point
    camera.up = frame.up
    if elevation:
        camera.Elevation(elevation)
        camera.OrthogonalizeViewUp()
    return frame_and_focus(plotter, fov=fov, spec=spec)

auto_skeleton_step(n_neurons, *, budget=SCENE_POINT_BUDGET)

The stride to draw n_neurons at so the scene stays near budget points.

A stride of 4 is the finest the skeleton cache holds, so that is the floor and a small scene simply gets it. Past roughly a thousand neurons the stride grows to keep the point count flat, which is what lets :data:connectomekg.validation.MAX_SCENE_NEURONS be five thousand rather than five hundred: the cost of a scene stops tracking the neuron count.

Parameters:

Name	Type	Description	Default
n_neurons	int	Neurons the scene will draw.	required
budget	int	Points to aim for across all of them.	SCENE_POINT_BUDGET

Returns:

Type	Description
int	A stride between :data:DEFAULT_CACHE_STEP and :data:connectomekg.validation.MAX_SKELETON_STEP.

Source code in src/connectomekg/scene.py

def auto_skeleton_step(n_neurons: int, *, budget: int = SCENE_POINT_BUDGET) -> int:
    """The stride to draw *n_neurons* at so the scene stays near ``budget`` points.

    A stride of 4 is the finest the skeleton cache holds, so that is the floor
    and a small scene simply gets it. Past roughly a thousand neurons the
    stride grows to keep the point count flat, which is what lets
    :data:`connectomekg.validation.MAX_SCENE_NEURONS` be five thousand rather
    than five hundred: the cost of a scene stops tracking the neuron count.

    :param n_neurons: Neurons the scene will draw.
    :param budget: Points to aim for across all of them.
    :return: A stride between :data:`DEFAULT_CACHE_STEP` and
        :data:`connectomekg.validation.MAX_SKELETON_STEP`.
    """
    if n_neurons <= 0:
        return DEFAULT_CACHE_STEP
    wanted = DEFAULT_CACHE_STEP * n_neurons * _POINTS_PER_NEURON_AT_STRIDE_4 / max(budget, 1)
    return int(min(MAX_SKELETON_STEP, max(DEFAULT_CACHE_STEP, round(wanted))))

build_brain_scene(plotter, kg, *, specs=(), view='circuit', data_dir=None, color_by='super_class', skeleton_step=None, tubes=False, top=100, neuropils=True, cloud=None, groups=None, progress=None, background=BACKGROUND)

Compose the whole-brain context cloud plus a circuit or neuropil flow into plotter.

Parameters:

Name	Type	Description	Default
plotter	Plotter	PyVista plotter to compose into; actors are cleared first.	required
kg	ConnectomeKG	An open ConnectomeKG.	required
specs	Sequence[str]	For view="circuit", specs resolved and unioned via :func:circuit_neurons for the circuit view (view B); empty draws the context cloud alone. For view="flow", specs whose neurons the flow sum is restricted to, with no neuron cap; empty sums every neuron.	()
view	str	"circuit" (view B) or "flow" (view C).	'circuit'
data_dir	str | Path | None	Skeleton download root (fafb_v783); without it every circuit neuron falls back to a marked-point sphere. Unused by the flow view.	None
color_by	str	Context cloud coloring, "super_class" or "sign". The flow view ignores it and draws the cloud in neutral gray.	'super_class'
skeleton_step	int | None	Skeleton simplification stride; None chooses one by neuron count via :func:auto_skeleton_step. Bounded to [1, MAX_SKELETON_STEP] via :func:~kg_utils.validation.bounded_int.	None
tubes	bool	Draw circuit skeletons as tubes instead of lines.	False
top	int	Flow arcs drawn, strongest first, bounded to [1, MAX_FLOW_PAIRS].	100
neuropils	bool	Draw the neuropil surface meshes when the dataset has them cached (connkg meshes); without the cache the scene is drawn without them and progress says how to fetch them.	True
cloud	bool | None	Draw the whole-brain context cloud of marked points; None (the default) draws it only when no neuropil meshes are.	None
progress	Callable[[str], None] | None	Called with a short message at each stage; None (the default) keeps the build silent. The same Callable[[str], None] contract as ConnectomeExtractor's, so the CLI and the viewer share it.	None
background	str	Scene background, a :data:BACKGROUNDS name or #RRGGBB; the context cloud is muted toward it. Pass the same one to :func:add_floor.	BACKGROUND

Returns:

Type	Description
SceneInfo	The composed :class:SceneInfo.

Raises:

Type	Description
ValueError	On an out-of-range argument, an unknown view or color_by, or a circuit over :data:connectomekg.validation.MAX_SCENE_NEURONS.

Source code in src/connectomekg/scene.py

def build_brain_scene(
    plotter: pv.Plotter,
    kg: ConnectomeKG,
    *,
    specs: Sequence[str] = (),
    view: str = "circuit",
    data_dir: str | Path | None = None,
    color_by: str = "super_class",
    skeleton_step: int | None = None,
    tubes: bool = False,
    top: int = 100,
    neuropils: bool = True,
    cloud: bool | None = None,
    groups: Sequence[NeuronGroup] | None = None,
    progress: Callable[[str], None] | None = None,
    background: str = BACKGROUND,
) -> SceneInfo:
    """Compose the whole-brain context cloud plus a circuit or neuropil flow into *plotter*.

    :param plotter: PyVista plotter to compose into; actors are cleared first.
    :param kg: An open ``ConnectomeKG``.
    :param specs: For ``view="circuit"``, specs resolved and unioned via
        :func:`circuit_neurons` for the circuit view (view B); empty draws the
        context cloud alone. For ``view="flow"``, specs whose neurons the flow
        sum is restricted to, with no neuron cap; empty sums every neuron.
    :param view: ``"circuit"`` (view B) or ``"flow"`` (view C).
    :param data_dir: Skeleton download root (``fafb_v783``); without it every
        circuit neuron falls back to a marked-point sphere. Unused by the
        flow view.
    :param color_by: Context cloud coloring, ``"super_class"`` or ``"sign"``.
        The flow view ignores it and draws the cloud in neutral gray.
    :param skeleton_step: Skeleton simplification stride; ``None`` chooses one
        by neuron count via :func:`auto_skeleton_step`. Bounded to
        ``[1, MAX_SKELETON_STEP]`` via :func:`~kg_utils.validation.bounded_int`.
    :param tubes: Draw circuit skeletons as tubes instead of lines.
    :param top: Flow arcs drawn, strongest first, bounded to
        ``[1, MAX_FLOW_PAIRS]``.
    :param neuropils: Draw the neuropil surface meshes when the dataset has
        them cached (``connkg meshes``); without the cache the scene is drawn
        without them and *progress* says how to fetch them.
    :param cloud: Draw the whole-brain context cloud of marked points;
        ``None`` (the default) draws it only when no neuropil meshes are.
    :param progress: Called with a short message at each stage; ``None`` (the
        default) keeps the build silent. The same
        ``Callable[[str], None]`` contract as ``ConnectomeExtractor``'s, so
        the CLI and the viewer share it.
    :param background: Scene background, a :data:`BACKGROUNDS` name or
        ``#RRGGBB``; the context cloud is muted toward it. Pass the same one
        to :func:`add_floor`.
    :return: The composed :class:`SceneInfo`.
    :raises ValueError: On an out-of-range argument, an unknown ``view`` or
        ``color_by``, or a circuit over :data:`connectomekg.validation.MAX_SCENE_NEURONS`.
    """
    import pyvista as pv  # noqa: PLC0415 - the viz3d-render-only import boundary

    view = require_choice("view", view, VIEWS)
    if skeleton_step is None:
        skeleton_step = 0  # resolved once the circuit's size is known
    else:
        skeleton_step = bounded_int("skeleton_step", skeleton_step, 1, MAX_SKELETON_STEP)
    top = bounded_int("top", top, 1, MAX_FLOW_PAIRS)
    background = resolve_background(background)

    def _say(message: str) -> None:
        if progress is not None:
            progress(message)

    _reset_floor_passes(plotter)
    plotter.clear_actors()
    plotter.set_background(background)  # ty: ignore[invalid-argument-type]
    frame = world_frame(kg.store)
    world_points: list[np.ndarray] = []

    n_meshes = 0
    if neuropils:
        mesh_path = neuropil_mesh_path(kg.db_path)
        if mesh_path.exists():
            meshes = load_neuropil_meshes(mesh_path)
            _say(f"{len(meshes)} neuropil meshes")
            _draw_neuropil_meshes(plotter, meshes, frame, view)
            world_points.extend(frame.to_world(v.astype(np.float64)) for v, _ in meshes.values())
            n_meshes = len(meshes)
        else:
            _say("no neuropil meshes cached; `connkg meshes` fetches them")

    n_context = 0
    n_context_somas = 0
    if cloud is None:
        # The surfaces show the brain's outline more plainly than 139,255
        # dots, so the cloud stays for a scene that has no meshes drawn.
        cloud = n_meshes == 0
    if cloud:
        _say("context point cloud")
        ctx_ids, ctx_points_nm, ctx_colors, n_context_somas = context_points(
            kg.store, color_by=color_by
        )
        ctx_ids, ctx_points_nm, ctx_colors, radius = _context_for_view(
            view, ctx_ids, ctx_points_nm, ctx_colors
        )
        n_context = len(ctx_ids)
        if n_context:
            ctx_world = frame.to_world(ctx_points_nm)
            cloud_mesh = pv.PolyData(ctx_world)
            rgb = np.asarray([_hex_to_rgb(c) for c in ctx_colors], dtype=np.float64)
            rgb = (
                rgb + (np.asarray(_hex_to_rgb(background), dtype=np.float64) - rgb) * _CONTEXT_MUTE
            )
            cloud_mesh.point_data["rgb"] = np.clip(rgb, 0, 255).astype(np.uint8)
            glyphs = cloud_mesh.glyph(
                geom=pv.Sphere(radius=radius, theta_resolution=6, phi_resolution=4),
                orient=False,
                scale=False,
            )
            plotter.add_mesh(glyphs, scalars="rgb", rgb=True, name="context")
            world_points.append(ctx_world)

    ds_row = kg.store.con.execute("SELECT name FROM nodes WHERE kind='dataset'").fetchone()
    dataset_name = ds_row[0] if ds_row else "connectome"

    if view == "flow":
        n_pairs, n_total = _draw_flow(plotter, kg, specs, frame, top, world_points, _say)
        add_studio_lighting(plotter)
        points = np.concatenate(world_points, axis=0) if world_points else np.zeros((1, 3))
        scope = f" via {', '.join(specs)}" if specs else ""
        title = (
            f"{dataset_name} | neuropil flow{scope}: top {n_pairs} of {n_total} pairs "
            f"context={n_context}"
        )
        _say("scene composed")
        return SceneInfo(
            title=title,
            points=points,
            n_context=n_context,
            n_circuit=0,
            n_skeletons=0,
            missing_skeletons=[],
            soma_fallbacks=0,
            view=view,
            n_flow_pairs=n_pairs,
            n_flow_total=n_total,
            n_neuropil_meshes=n_meshes,
            n_context_somas=n_context_somas,
        )

    if groups is not None:
        circuit_ids = sorted({nid for g in groups for nid in g.neuron_ids})
        if len(circuit_ids) > MAX_SCENE_NEURONS:
            raise ValueError(
                f"{len(circuit_ids)} neurons in these groups, over the cap of "
                f"{MAX_SCENE_NEURONS} for one scene"
            )
    else:
        circuit_ids = circuit_neurons(kg, specs) if specs else []
    n_circuit = len(circuit_ids)

    _say(f"resolving {n_circuit} circuit neurons")
    metas: dict[str, dict] = {}
    root_ids: list[int] = []
    for nid in circuit_ids:
        node = kg.store.node(nid)
        if node is None:
            continue
        metas[nid] = node.get("metadata") or {}
        root_id = metas[nid].get("root_id")
        if root_id is not None:
            root_ids.append(int(root_id))

    # Either the caller said what belongs together, or cell type decides.
    if groups is not None:
        draw_groups: list[tuple[str, str, list[tuple[str, dict]]]] = [
            (g.label, g.color, [(nid, metas[nid]) for nid in g.neuron_ids if nid in metas])
            for g in groups
        ]
    else:
        by_type: dict[str, list[tuple[str, dict]]] = {}
        for nid in circuit_ids:
            if nid in metas:
                by_type.setdefault(str(metas[nid].get("cell_type") or "unknown"), []).append(
                    (nid, metas[nid])
                )
        draw_groups = [(n, type_color(n), m) for n, m in by_type.items()]

    if not skeleton_step:
        skeleton_step = auto_skeleton_step(n_circuit)
        _say(f"stride {skeleton_step} for {n_circuit} neurons")

    skeletons_by_root: dict[int, Skeleton] = {}
    missing_skeletons: list[int] = list(root_ids)
    cached_roots: set[int] = set()
    cache_path = skeleton_cache_path(kg.db_path)
    if root_ids and cache_path.exists():
        _say(f"loading {len(root_ids)} skeletons from {cache_path.name}")
        skeletons_by_root, missing_skeletons = load_cached_skeletons(cache_path, root_ids)
        cached_roots = set(skeletons_by_root)
    if data_dir is not None and missing_skeletons:
        _say(f"loading {len(missing_skeletons)} skeletons from {data_dir}")
        from_swc, missing_skeletons = load_skeletons(data_dir, missing_skeletons)
        skeletons_by_root.update(from_swc)
    # A cached skeleton is already simplified, so it is drawn with the stride
    # divided down rather than applied a second time; see
    # connectomekg.skeleton_cache.effective_step.
    cached_step = effective_step(skeleton_step, cache_info(cache_path)[0]) if cached_roots else 1

    n_skeletons = 0
    soma_fallbacks = 0
    collector = _Collector()
    if circuit_ids:
        _say(f"drawing {n_circuit} circuit neurons")
    for label, color, members in draw_groups:
        chain_points: list[np.ndarray] = []
        chain_radii: list[np.ndarray] = []
        chain_cells: list[np.ndarray] = []
        n_chain_points = 0
        soma_world: list[np.ndarray] = []
        fallback_world: list[np.ndarray] = []
        for nid, meta in members:
            root_id = meta.get("root_id")
            skeleton = skeletons_by_root.get(int(root_id)) if root_id is not None else None
            if skeleton is not None:
                n_skeletons += 1
                step = cached_step if skeleton.root_id in cached_roots else skeleton_step
                chains = polylines(skeleton, step=step)
                if chains:
                    drawn = frame.to_world(skeleton.points)
                    # Chains index the whole skeleton, so the points go in as
                    # one block and the cells are renumbered past what is
                    # already in this group's batch.
                    chain_cells.extend(
                        np.concatenate([[len(c)], c + n_chain_points]) for c in chains
                    )
                    chain_points.append(drawn)
                    chain_radii.append(
                        np.clip(
                            skeleton.radius / NM_PER_WORLD_UNIT,
                            _MIN_TUBE_RADIUS,
                            _MAX_TUBE_RADIUS,
                        )
                    )
                    n_chain_points += len(drawn)
                    collector.add(nid, drawn)
                soma_nm, is_soma = soma(skeleton)
                if not is_soma:
                    soma_fallbacks += 1
                soma_world.append(frame.to_world(soma_nm)[0])
            else:
                x, y, z = meta.get("x"), meta.get("y"), meta.get("z")
                if x is None or y is None or z is None:
                    continue
                marked = frame.to_world(np.asarray([x, y, z], dtype=np.float64))[0]
                fallback_world.append(marked)
                # A neuron with no skeleton is one sphere, and that sphere is
                # the only thing there is to click on.
                collector.add(nid, marked)

        if chain_points:
            points = np.concatenate(chain_points, axis=0)
            radii = np.concatenate(chain_radii, axis=0)
            # Built empty and filled, not `PolyData(points)`: that constructor
            # adds a vertex cell per point, and those points then draw beside
            # the lines. They have no normals, so under the floor's shadow
            # pass their shader fails to compile -- the same unlit-branch
            # failure `_shade_skeleton_lines` exists to prevent, which it
            # cannot fix here because it turns *lines* into tubes.
            mesh = pv.PolyData()
            mesh.points = points
            mesh.lines = np.concatenate(chain_cells).astype(np.int64)
            if tubes:
                # A format-1 cache stores no radius, so its skeletons read back
                # as zeros and floor to the constant width they were drawn at.
                if radii.max() > _MIN_TUBE_RADIUS:
                    mesh["radius"] = radii
                    mesh = mesh.tube(
                        scalars="radius",
                        absolute=True,
                        radius=_TUBE_RADIUS,
                        n_sides=_TUBE_SIDES,
                    )
                else:
                    mesh = mesh.tube(radius=_TUBE_RADIUS, n_sides=_TUBE_SIDES)
            plotter.add_mesh(mesh, color=color, line_width=2, name=f"skeleton:{label}")
            world_points.append(points)
        if soma_world:
            arr = np.asarray(soma_world)
            glyph = pv.PolyData(arr).glyph(
                geom=pv.Sphere(radius=_SOMA_RADIUS * frame.marker_scale),
                orient=False,
                scale=False,
            )
            plotter.add_mesh(glyph, color=color, name=f"soma:{label}")
            world_points.append(arr)
        if fallback_world:
            arr = np.asarray(fallback_world)
            glyph = pv.PolyData(arr).glyph(
                geom=pv.Sphere(radius=_FALLBACK_RADIUS * frame.marker_scale),
                orient=False,
                scale=False,
            )
            plotter.add_mesh(glyph, color=color, name=f"fallback:{label}")
            world_points.append(arr)

    add_studio_lighting(plotter)
    points = np.concatenate(world_points, axis=0) if world_points else np.zeros((1, 3))
    cloud_kind = "somas" if n_context_somas else "context"
    title = f"{dataset_name} | {cloud_kind}={n_context} circuit={n_circuit} skeletons={n_skeletons}"
    _say("scene composed")
    return SceneInfo(
        title=title,
        points=points,
        n_context=n_context,
        n_circuit=n_circuit,
        n_skeletons=n_skeletons,
        missing_skeletons=sorted(set(missing_skeletons)),
        soma_fallbacks=soma_fallbacks,
        n_neuropil_meshes=n_meshes,
        n_context_somas=n_context_somas,
        picks=collector.build(),
    )

circuit_neurons(kg, specs)

The union of every spec's neurons, for the circuit view (view B).

Parameters:

Name	Type	Description	Default
kg	ConnectomeKG	An open ConnectomeKG.	required
specs	Sequence[str]	Specs, as accepted by :meth:ConnectomeKG.neurons_of.	required

Returns:

Type	Description
list[str]	Sorted, deduplicated neuron node ids.

Raises:

Type	Description
ValueError	If the union exceeds :data:connectomekg.validation.MAX_SCENE_NEURONS, naming the count and the cap.

Source code in src/connectomekg/scene.py

def circuit_neurons(kg: ConnectomeKG, specs: Sequence[str]) -> list[str]:
    """The union of every spec's neurons, for the circuit view (view B).

    :param kg: An open ``ConnectomeKG``.
    :param specs: Specs, as accepted by :meth:`ConnectomeKG.neurons_of`.
    :return: Sorted, deduplicated neuron node ids.
    :raises ValueError: If the union exceeds :data:`connectomekg.validation.MAX_SCENE_NEURONS`,
        naming the count and the cap.
    """
    ids: set[str] = set()
    for spec in specs:
        ids.update(kg.neurons_of(spec))
    if len(ids) > MAX_SCENE_NEURONS:
        raise ValueError(
            f"{len(ids)} neurons resolved, over the cap of {MAX_SCENE_NEURONS} for one "
            "scene; narrow the spec(s)"
        )
    return sorted(ids)

context_points(store, *, color_by='super_class')

Every neuron's soma, or its marked point, for the whole-brain context cloud (view A).

A neuron carries FlyWire's marked point as x/y/z always, and its real soma as soma_x/soma_y/soma_z once connkg skeletons has back-filled one (:func:connectomekg.skeleton_cache.write_somas). The soma wins where there is one, so a graph with the back-fill draws a soma cloud and one without still draws the marked-point cloud it always did.

Parameters:

Name	Type	Description	Default
store	GraphStore	The graph store.	required
color_by	str	"super_class" or "sign".	'super_class'

Returns:

Type	Description
tuple[list[str], ndarray, list[str], int]	(neuron_ids, points_nm, colors, n_somas) -- points_nm is (n, 3), colors is one #RRGGBB string per neuron, same order as neuron_ids, and n_somas counts how many of those points are a real soma rather than a marked point. Neurons with no coordinates at all are omitted.

Raises:

Type	Description
ValueError	If color_by is not one of the two choices.

Source code in src/connectomekg/scene.py

def context_points(
    store: GraphStore, *, color_by: str = "super_class"
) -> tuple[list[str], np.ndarray, list[str], int]:
    """Every neuron's soma, or its marked point, for the whole-brain context cloud (view A).

    A neuron carries FlyWire's marked point as ``x``/``y``/``z`` always, and
    its real soma as ``soma_x``/``soma_y``/``soma_z`` once ``connkg skeletons``
    has back-filled one (:func:`connectomekg.skeleton_cache.write_somas`). The
    soma wins where there is one, so a graph with the back-fill draws a soma
    cloud and one without still draws the marked-point cloud it always did.

    :param store: The graph store.
    :param color_by: ``"super_class"`` or ``"sign"``.
    :return: ``(neuron_ids, points_nm, colors, n_somas)`` -- ``points_nm`` is
        ``(n, 3)``, ``colors`` is one ``#RRGGBB`` string per neuron, same
        order as ``neuron_ids``, and ``n_somas`` counts how many of those
        points are a real soma rather than a marked point. Neurons with no
        coordinates at all are omitted.
    :raises ValueError: If ``color_by`` is not one of the two choices.
    """
    color_by = require_choice("color_by", color_by, ("super_class", "sign"))
    rows = store.con.execute(
        f"SELECT id, "
        f"COALESCE(json_extract(metadata,'$.soma_x'), json_extract(metadata,'$.x')), "
        f"COALESCE(json_extract(metadata,'$.soma_y'), json_extract(metadata,'$.y')), "
        f"COALESCE(json_extract(metadata,'$.soma_z'), json_extract(metadata,'$.z')), "
        f"json_extract(metadata,'$.super_class'), json_extract(metadata,'$.sign'), "
        f"json_extract(metadata,'$.has_soma') FROM nodes WHERE {_SQL_NEURON_XYZ}"
    ).fetchall()
    ids = [r[0] for r in rows]
    points = (
        np.asarray([(r[1], r[2], r[3]) for r in rows], dtype=np.float64)
        if rows
        else np.empty((0, 3), dtype=np.float64)
    )
    if color_by == "super_class":
        colors = [SUPER_CLASS_COLOR.get(r[4] or "", UNKNOWN_COLOR) for r in rows]
    else:
        colors = [SIGN_COLOR.get(int(r[5]) if r[5] is not None else 0, SIGN_COLOR[0]) for r in rows]
    return ids, points, colors, sum(1 for r in rows if r[6])

floor_color(background)

The floor under background: :data:FLOOR_COLOR under the default, else a step darker, or a step lighter under a dark background.

Parameters:

Name	Type	Description	Default
background	str	The scene background, #RRGGBB.	required

Returns:

Type	Description
str	The floor color, #RRGGBB.

Source code in src/connectomekg/scene.py

def floor_color(background: str) -> str:
    """The floor under *background*: :data:`FLOOR_COLOR` under the default,
    else a step darker, or a step lighter under a dark background.

    :param background: The scene background, ``#RRGGBB``.
    :return: The floor color, ``#RRGGBB``.
    """
    if background.upper() == BACKGROUND:
        return FLOOR_COLOR
    dark = sum(_hex_to_rgb(background)) / 3 < 64
    return _blend(background, "#FFFFFF" if dark else "#000000", _FLOOR_STEP)

flow_arc(start, end, *, bow=_FLOW_BOW, n_points=_FLOW_ARC_POINTS)

A quadratic arc from start to end, bowed to one side of its direction.

The bow points along direction x world-up (world +x when the direction is vertical), so the arc for A -> B and the arc for B -> A bow to opposite sides instead of overdrawing.

Parameters:

Name	Type	Description	Default
start	ndarray	(3,) world start point.	required
end	ndarray	(3,) world end point.	required
bow	float	Control point offset as a fraction of the chord length.	_FLOW_BOW
n_points	int	Points along the arc, including both ends.	_FLOW_ARC_POINTS

Returns:

Type	Description
ndarray	(n_points, 3) world points.

Source code in src/connectomekg/scene.py

def flow_arc(
    start: np.ndarray,
    end: np.ndarray,
    *,
    bow: float = _FLOW_BOW,
    n_points: int = _FLOW_ARC_POINTS,
) -> np.ndarray:
    """A quadratic arc from *start* to *end*, bowed to one side of its direction.

    The bow points along ``direction x world-up`` (world ``+x`` when the
    direction is vertical), so the arc for A -> B and the arc for B -> A bow to
    opposite sides instead of overdrawing.

    :param start: ``(3,)`` world start point.
    :param end: ``(3,)`` world end point.
    :param bow: Control point offset as a fraction of the chord length.
    :param n_points: Points along the arc, including both ends.
    :return: ``(n_points, 3)`` world points.
    """
    start = np.asarray(start, dtype=np.float64)
    end = np.asarray(end, dtype=np.float64)
    chord = end - start
    length = float(np.linalg.norm(chord))
    side = np.cross(chord, (0.0, 0.0, 1.0))
    if np.linalg.norm(side) < 1e-9 * max(length, 1.0):
        side = np.array([1.0, 0.0, 0.0])
    side = side / np.linalg.norm(side)
    control = (start + end) / 2 + side * bow * length
    t = np.linspace(0.0, 1.0, n_points)[:, None]
    return (1 - t) ** 2 * start + 2 * (1 - t) * t * control + t**2 * end

neuropil_flow(store, neuron_ids=None)

Aggregate IN_NEUROPIL evidence into neuropil centroids and flow.

Centroids always use every neuron, so neuropils sit in the same place in every flow render; only the flow sum is restricted by neuron_ids.

Parameters:

Name	Type	Description	Default
store	GraphStore	The graph store.	required
neuron_ids	Iterable[str] | None	Neuron node ids whose flow is summed; None sums every neuron.	None

Returns:

Type	Description
NeuropilFlow	The :class:NeuropilFlow.

Source code in src/connectomekg/scene.py

def neuropil_flow(store: GraphStore, neuron_ids: Iterable[str] | None = None) -> NeuropilFlow:
    """Aggregate ``IN_NEUROPIL`` evidence into neuropil centroids and flow.

    Centroids always use every neuron, so neuropils sit in the same place in
    every flow render; only the flow sum is restricted by *neuron_ids*.

    :param store: The graph store.
    :param neuron_ids: Neuron node ids whose flow is summed; ``None`` sums
        every neuron.
    :return: The :class:`NeuropilFlow`.
    """
    con = store.con
    np_rows = con.execute(
        "SELECT id, name, json_extract(metadata,'$.base'), json_extract(metadata,'$.n_synapses') "
        "FROM nodes WHERE kind='neuropil' ORDER BY name"
    ).fetchall()
    names = [r[1] for r in np_rows]
    name_of = {r[0]: r[1] for r in np_rows}
    index = {name: i for i, name in enumerate(names)}

    positions = {
        r[0]: (r[1], r[2], r[3])
        for r in con.execute(
            f"SELECT id, json_extract(metadata,'$.x'), json_extract(metadata,'$.y'), "
            f"json_extract(metadata,'$.z') FROM nodes WHERE {_SQL_NEURON_XYZ}"
        )
    }
    by_neuron: dict[str, list[tuple[str, int, int]]] = defaultdict(list)
    for nid, np_id, pre, post in con.execute(
        "SELECT src, dst, json_extract(evidence,'$.pre'), json_extract(evidence,'$.post') "
        "FROM edges WHERE rel='IN_NEUROPIL'"
    ):
        if np_id in name_of:
            by_neuron[nid].append((name_of[np_id], int(pre or 0), int(post or 0)))

    weighted = np.zeros((len(names), 3), dtype=np.float64)
    weights = np.zeros(len(names), dtype=np.float64)
    for nid, rows in by_neuron.items():
        pos = positions.get(nid)
        if pos is None:
            continue
        for name, pre, post in rows:
            i = index[name]
            weighted[i] += (pre + post) * np.asarray(pos, dtype=np.float64)
            weights[i] += pre + post
    with np.errstate(invalid="ignore", divide="ignore"):
        centroids = weighted / weights[:, None]
    centroids[weights == 0] = np.nan

    selected = (
        by_neuron if neuron_ids is None else {n: by_neuron[n] for n in neuron_ids if n in by_neuron}
    )
    flow: dict[tuple[str, str], float] = defaultdict(float)
    for rows in selected.values():
        total_post = sum(post for _, _, post in rows)
        if total_post == 0:
            continue
        for source, _, post in rows:
            if post == 0:
                continue
            share = post / total_post
            for target, pre, _ in rows:
                if pre and target != source:
                    flow[(source, target)] += pre * share

    pairs = sorted(((a, b, w) for (a, b), w in flow.items()), key=lambda p: (-p[2], p[0], p[1]))
    return NeuropilFlow(
        names=names,
        bases=[r[2] or r[1] for r in np_rows],
        centroids_nm=centroids,
        n_synapses=np.asarray([r[3] or 0 for r in np_rows], dtype=np.float64),
        pairs=pairs,
    )

region_color(neuropil)

The color for a neuropil: its brain region's color.

Parameters:

Name	Type	Description	Default
neuropil	str	FlyWire abbreviation, with or without side suffix.	required

Returns:

Type	Description
str	A #RRGGBB color from :data:connectomekg.colors.REGION_COLOR, or the unknown gray for a neuropil with no region.

Source code in src/connectomekg/scene.py

def region_color(neuropil: str) -> str:
    """The color for a neuropil: its brain region's color.

    :param neuropil: FlyWire abbreviation, with or without side suffix.
    :return: A ``#RRGGBB`` color from :data:`connectomekg.colors.REGION_COLOR`,
        or the unknown gray for a neuropil with no region.
    """
    return REGION_COLOR.get(neuropil_region(neuropil) or "", UNKNOWN_COLOR)

resolve_background(value)

A background color from a :data:BACKGROUNDS name or a #RRGGBB string.

Parameters:

Name	Type	Description	Default
value	str	A :data:BACKGROUNDS name, or a hex color.	required

Returns:

Type	Description
str	The color as upper-case #RRGGBB.

Raises:

Type	Description
ValueError	If value is neither.

Source code in src/connectomekg/scene.py

def resolve_background(value: str) -> str:
    """A background color from a :data:`BACKGROUNDS` name or a ``#RRGGBB`` string.

    :param value: A :data:`BACKGROUNDS` name, or a hex color.
    :return: The color as upper-case ``#RRGGBB``.
    :raises ValueError: If *value* is neither.
    """
    text = value.strip()
    if text.lower() in BACKGROUNDS:
        return BACKGROUNDS[text.lower()]
    if len(text) == 7 and text[0] == "#":
        try:
            int(text[1:], 16)
        except ValueError:
            pass
        else:
            return text.upper()
    raise ValueError(
        f"background must be one of {', '.join(BACKGROUNDS)} or #RRGGBB, not {value!r}"
    )

type_color(name)

A deterministic color for a cell type name.

Indexes a fixed qualitative palette via seed_from_key, so the same type name always draws the same color -- across renders, sessions and a printed figure.

Parameters:

Name	Type	Description	Default
name	str	Cell type name, e.g. "LC4".	required

Returns:

Type	Description
str	A #RRGGBB color.

Source code in src/connectomekg/scene.py

def type_color(name: str) -> str:
    """A deterministic color for a cell type name.

    Indexes a fixed qualitative palette via ``seed_from_key``, so the same
    type name always draws the same color -- across renders, sessions and a
    printed figure.

    :param name: Cell type name, e.g. ``"LC4"``.
    :return: A ``#RRGGBB`` color.
    """
    return _TYPE_PALETTE[seed_from_key(name) % len(_TYPE_PALETTE)]

world_frame(store)

The scene's :class:WorldFrame, centered on the median neuron position.

Parameters:

Name	Type	Description	Default
store	GraphStore	The graph store.	required

Returns:

Type	Description
WorldFrame	A frame centered on the median of every neuron's marked point (skipping neurons with no coordinates), at :data:NM_PER_WORLD_UNIT, with a marker_scale of the positions' framed extent over FAFB's, never below 1.

Raises:

Type	Description
ValueError	If no neuron in the store has coordinates.

Source code in src/connectomekg/scene.py

def world_frame(store: GraphStore) -> WorldFrame:
    """The scene's :class:`WorldFrame`, centered on the median neuron position.

    :param store: The graph store.
    :return: A frame centered on the median of every neuron's marked point
        (skipping neurons with no coordinates), at :data:`NM_PER_WORLD_UNIT`,
        with a ``marker_scale`` of the positions' framed extent over FAFB's,
        never below 1.
    :raises ValueError: If no neuron in the store has coordinates.
    """
    rows = store.con.execute(
        f"SELECT json_extract(metadata,'$.x'), json_extract(metadata,'$.y'), "
        f"json_extract(metadata,'$.z') FROM nodes WHERE {_SQL_NEURON_XYZ}"
    ).fetchall()
    if not rows:
        raise ValueError("no neuron in this graph has x/y/z coordinates")
    points = np.asarray(rows, dtype=np.float64)
    center = np.median(points, axis=0)
    # Screen width is nm x and screen height nm y (see WorldFrame.to_world).
    low, high = np.percentile(points, [1, 99], axis=0)
    framed = max((high[0] - low[0]) / _REFERENCE_ASPECT, high[1] - low[1])
    marker_scale = min(max(framed / _REFERENCE_FRAMED_NM, 1.0), _MAX_MARKER_SCALE)
    return WorldFrame(center=center, marker_scale=float(marker_scale))