Compiling SankeyDefinitions

The compile_sankey_definition() function converts a Sankey diagram definitions to a WeaverSpec. This can then be passed to execute_weave() to apply to a specific Datasets, or saved to JSON format and applied to datasets later using the Python or Javascript implementations.

compile_sankey_definition(sankey_definition: SankeyDefinition, measures='value', link_width=None, link_color=None, palette=None, add_elsewhere_waypoints=True, dimension_tables=None)

Compile a SankeyDefinition into a WeaverSpec with routing tree.

This pre-expands all partitions and builds a decision tree for efficient flow routing at execution time.

Parameters:

• sankey_definition (SankeyDefinition) – The high-level definition of the Sankey diagram.

• measures (str, list, or dict) – Measures to aggregate.

• link_width (str, optional) – Measure name to use for link width.

• link_color (str or ColorScale, optional) – Color scale for links.

• palette (str or list, optional) – Color palette.

• add_elsewhere_waypoints (bool) – Whether to add waypoints for elsewhere flows.

• dimension_tables (dict) – Dimension tables for resolving query strings.

Returns:

The compiled spec with routing tree.

Return type:

WeaverSpec

execute_weave(spec, dataset)

Execute a WeaverSpec against flow data to produce SankeyData.

Parameters:

• spec (WeaverSpec) – The compiled spec with routing tree.

• dataset (Dataset or DataFrame) – The flow data.

Returns:

The resulting Sankey diagram data with nodes and links.

Return type:

SankeyData

Compiled specs

Spec schema for compiled Sankey definitions.

This module defines the JSON-serializable spec format that represents a fully compiled Sankey diagram definition, with selections/partitions expanded and converted to explicit filters. The spec can be executed against flow data to produce SankeyData results.

class BundleSpec(id: str | int, source: str, target: str)

Specification for a bundle (provenance for edges).

classmethod from_json(data: dict) → BundleSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class CategoricalColorSpec(attribute: str, lookup: Dict[str, str], default: str, type: str = 'categorical')

Categorical color scale specification.

classmethod from_json(data: dict) → CategoricalColorSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class DisplaySpec(link_width: str, link_color: CategoricalColorSpec | QuantitativeColorSpec)

Display configuration for the Sankey diagram.

classmethod from_json(data: dict) → DisplaySpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class EdgeSpec(source: str | None, target: str | None, type: str, time: str, bundle_ids: List[str | int])

Specification for a single edge in the Sankey diagram.

classmethod from_json(data: dict) → EdgeSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class GroupSpec(id: str, title: str, nodes: List[str])

Specification for a process group.

classmethod from_json(data: dict) → GroupSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class MeasureSpec(column: str, aggregation: str)

Specification for a measure to aggregate from flow data.

classmethod from_json(data: dict) → MeasureSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class NodeSpec(title: str, type: str, group: str | None, style: str, direction: str, hidden: bool = False)

Specification for a single node in the Sankey diagram.

classmethod from_json(data: dict) → NodeSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class QuantitativeColorSpec(attribute: str, palette: List[str], domain: Tuple[float, float], type: str = 'quantitative', intensity: str | None = None)

Quantitative color scale specification.

classmethod from_json(data: dict) → QuantitativeColorSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.

class WeaverSpec(version: str, nodes: Dict[str, NodeSpec], groups: List[GroupSpec], bundles: List[BundleSpec], ordering: List[List[List[str]]], edges: List[EdgeSpec], measures: List[MeasureSpec], display: DisplaySpec, routing_tree: Node[tuple[int, ...]])

Complete specification for a compiled Sankey diagram.

This is the top-level spec that can be serialized to JSON and executed against flow data to produce a SankeyData object.

classmethod from_json(data: dict) → WeaverSpec

Create from JSON dict.

to_json() → dict

Convert to JSON-serializable dict.