graflo.plot.plotter

Graph visualization utilities for schema and data structures.

This module provides utilities for visualizing graph database schemas, relationships, and data structures using NetworkX and Graphviz. It includes functionality for plotting vertex collections, resources, and their relationships.

Key Components

• SchemaPlotter: Main class for schema visualization
• AuxNodeType: Enum for different node types in visualizations
• Color and shape mappings for different node types
• Tree assembly and graph generation utilities

Example

plotter = ManifestPlotter("config.json", "output/") plotter.plot_vc2fields() # Plot vertex collections and their fields plotter.plot_resources() # Plot resource relationships plotter.plot_vc2vc() # Plot vertex collection relationships

Attributes

ACTOR_BORDER = 'brown' module-attribute

ACTOR_LEGEND_BY_CLASS = {DescendActor: 'descend', VertexActor: 'vertex', EdgeActor: 'edge', TransformActor: 'transform', VertexRouterActor: 'vertex router'} module-attribute

EDGE_STYLE_BY_SOURCE_TYPE = {AuxNodeType.VERTEX: 'solid'} module-attribute

NODE_COLOR_BY_TYPE = {AuxNodeType.RESOURCE: fillcolor_palette['blue'], AuxNodeType.FIELD_DEFINITION: fillcolor_palette['red'], AuxNodeType.VERTEX_BLANK: 'white', AuxNodeType.VERTEX: fillcolor_palette['green'], AuxNodeType.INDEX: 'orange', AuxNodeType.TRANSFORM: fillcolor_palette['stone'], AuxNodeType.FIELD: fillcolor_palette['violet']} module-attribute

NODE_LEGEND_BY_TYPE = {AuxNodeType.VERTEX: 'vertex type', AuxNodeType.VERTEX_BLANK: 'blank vertex', AuxNodeType.RESOURCE: 'resource', AuxNodeType.FIELD_DEFINITION: 'identity field', AuxNodeType.FIELD: 'field', AuxNodeType.INDEX: 'index', AuxNodeType.TRANSFORM: 'transform'} module-attribute

NODE_SHAPE_BY_TYPE = {AuxNodeType.RESOURCE: 'box', AuxNodeType.VERTEX_BLANK: 'box', AuxNodeType.FIELD_DEFINITION: 'trapezium', AuxNodeType.TRANSFORM: 'oval', AuxNodeType.VERTEX: 'ellipse', AuxNodeType.INDEX: 'polygon', AuxNodeType.FIELD: 'octagon'} module-attribute

edge_status = EDGE_STYLE_BY_SOURCE_TYPE module-attribute

fillcolor_palette = {'violet': '#DDD0E5', 'green': '#BEDFC8', 'blue': '#B7D1DF', 'red': '#EBA59E', 'peach': '#FFE5B4', 'stone': '#E4E1D7', 'periwinkle': '#C7CEEA'} module-attribute

logger = logging.getLogger(__name__) module-attribute

map_class2color = {DescendActor: 'white', VertexActor: fillcolor_palette['green'], EdgeActor: fillcolor_palette['periwinkle'], TransformActor: fillcolor_palette['stone'], VertexRouterActor: fillcolor_palette['peach']} module-attribute

map_type2color = NODE_COLOR_BY_TYPE module-attribute

map_type2shape = NODE_SHAPE_BY_TYPE module-attribute

partition_color_palette = ['#FBE3A0', '#A8DADC', '#F4B6C2', '#C9E4A6', '#FFC8A2', '#B5C7F2', '#E0C3A8', '#D6B4E8'] module-attribute

Classes

AuxNodeType

Bases: BaseEnum

Node types for graph visualization.

This enum defines the different types of nodes that can appear in the visualization graphs, each with specific visual properties.

Attributes:

Name	Type	Description
FIELD		Regular field node
FIELD_DEFINITION		Field definition node
INDEX		Index field node
RESOURCE		Resource node
TRANSFORM		Transform node
VERTEX		Vertex node
VERTEX_BLANK		Empty vertex node

Source code in graflo/plot/plotter.py

class AuxNodeType(BaseEnum):
    """Node types for graph visualization.

    This enum defines the different types of nodes that can appear in the
    visualization graphs, each with specific visual properties.

    Attributes:
        FIELD: Regular field node
        FIELD_DEFINITION: Field definition node
        INDEX: Index field node
        RESOURCE: Resource node
        TRANSFORM: Transform node
        VERTEX: Vertex node
        VERTEX_BLANK: Empty vertex node
    """

    FIELD = "field"
    FIELD_DEFINITION = "field_definition"
    INDEX = "index"
    RESOURCE = "resource"
    TRANSFORM = "transform"
    VERTEX = "vertex"
    VERTEX_BLANK = "vertex_blank"

Attributes

FIELD = 'field' class-attribute instance-attribute

FIELD_DEFINITION = 'field_definition' class-attribute instance-attribute

INDEX = 'index' class-attribute instance-attribute

RESOURCE = 'resource' class-attribute instance-attribute

TRANSFORM = 'transform' class-attribute instance-attribute

VERTEX = 'vertex' class-attribute instance-attribute

VERTEX_BLANK = 'vertex_blank' class-attribute instance-attribute

ManifestPlotter

Main class for schema visualization.

This class provides methods to visualize different aspects of a graph database schema, including vertex collections, resources, and their relationships.

Attributes:

Name	Type	Description
fig_path		Path to save visualizations
config		Schema configuration
schema		Schema instance
name		Schema name
prefix		Prefix for output files

Source code in graflo/plot/plotter.py

class ManifestPlotter:
    """Main class for schema visualization.

    This class provides methods to visualize different aspects of a graph database
    schema, including vertex collections, resources, and their relationships.

    Attributes:
        fig_path: Path to save visualizations
        config: Schema configuration
        schema: Schema instance
        name: Schema name
        prefix: Prefix for output files
    """

    def __init__(
        self,
        config_filename: str | os.PathLike[str] | None = None,
        fig_path: str | os.PathLike[str] = ".",
        output_format: str = "pdf",
        output_dpi: int | None = None,
        graph_manifest: GraphManifest | None = None,
    ):
        """Initialize the schema plotter.

        Args:
            config_filename: Optional path to schema configuration file
            fig_path: Path to save visualizations
            graph_manifest: Optional pre-built graph manifest
        """
        self.fig_path = fig_path
        self.output_format = output_format.lower()
        self.output_dpi = output_dpi

        manifest: GraphManifest
        if graph_manifest is not None:
            manifest = graph_manifest
            self.config = None
        elif config_filename is not None:
            self.config = FileHandle.load(config_filename)
            manifest = GraphManifest.from_config(self.config)
        else:
            raise ValueError(
                "ManifestPlotter requires either `config_filename` or `graph_manifest`."
            )

        manifest.finish_init()
        self.schema = manifest.require_schema()
        self.ingestion_model = manifest.require_ingestion_model()

        # A label, not a filename: merged names carry "+" and may carry spaces.
        self.name = Schema._slug_filename_token(self.schema.metadata.name)
        self.prefix = self.name
        # The one index over the schema's declared adjacency. Built here so
        # every plot reads the same traversal rather than each deriving its
        # own from `vertex_config` / `edge_config`.
        self.schema_graph = SchemaGraph.from_schema(self.schema)

    def _figure_path(self, stem: str) -> str:
        return os.path.join(self.fig_path, f"{stem}.{self.output_format}")

    def _versioned_stem(self, stem: str) -> str:
        metadata = getattr(self.schema, "metadata", None)
        version = getattr(metadata, "version", None)
        if version is None:
            return stem
        return f"{stem}-v{version}"

    def _draw(self, ag, stem: str, prog: str = "dot") -> None:
        draw(
            ag,
            self._figure_path(stem),
            output_format=self.output_format,
            prog=prog,
            dpi=self.output_dpi,
        )

    def _discover_edges_from_resources(
        self,
    ) -> tuple[dict[EdgeId, Edge], dict[EdgeId, str], dict[EdgeId, bool]]:
        """Discover edges from resources by walking through ActorWrappers.

        This method finds all EdgeActors in resources and extracts their edges,
        which may include edges with dynamic relations (EdgeActor derivation) that
        aren't fully represented in edge_config.

        Returns:
            discovered_edges: map edge_id → Edge
            relation_source_by_edge_id: document field for per-row relation (plot hint).
            relation_from_key_by_edge_id: True when an edge step derives relation from keys.
        """
        discovered_edges: dict[EdgeId, Edge] = {}
        relation_source_by_edge_id: dict[EdgeId, str] = {}
        relation_from_key_by_edge_id: dict[EdgeId, bool] = {}

        for resource_config in self.ingestion_model.resources:
            actors = ActorWrapper(*resource_config.pipeline).collect_actors()

            for actor in actors:
                if isinstance(actor, EdgeActor):
                    edge = actor.edge
                    # Dynamic EdgeActors resolve types at row time; skip for plot discovery.
                    if edge is None:
                        continue
                    edge_id = edge.edge_id
                    # Store the edge, preferring already discovered edges from edge_config
                    # but allowing resource edges to supplement
                    if edge_id not in discovered_edges:
                        discovered_edges[edge_id] = edge
                    if actor.relation_field is not None:
                        relation_source_by_edge_id[edge_id] = actor.relation_field
                    if actor.derivation.relation_from_key:
                        relation_from_key_by_edge_id[edge_id] = True

        return (
            discovered_edges,
            relation_source_by_edge_id,
            relation_from_key_by_edge_id,
        )

    @staticmethod
    def _edge_label(
        edge: Edge,
        *,
        relation_source_field: str | None = None,
        relation_from_key: bool = False,
    ) -> str | None:
        """Build the human-readable edge label for plotting."""
        if edge.relation is not None:
            return edge.relation
        if relation_source_field is not None:
            return f"[{relation_source_field}]"
        if relation_from_key:
            return "[key]"
        return None

    @staticmethod
    def _merge_edges(primary_edges: dict, supplemental_edges: dict) -> dict:
        """Merge edge dictionaries without replacing existing primary edges."""
        merged = dict(primary_edges)
        for edge_id, edge in supplemental_edges.items():
            merged.setdefault(edge_id, edge)
        return merged

    @staticmethod
    def _filter_edges_with_known_vertices(
        edges: dict,
        known_vertices: set[str],
    ) -> tuple[dict, dict]:
        """Split edges by whether both endpoints are known vertices."""
        valid_edges = {}
        invalid_edges = {}
        for edge_id, edge in edges.items():
            source, target, _ = edge_id
            if source in known_vertices and target in known_vertices:
                valid_edges[edge_id] = edge
            else:
                invalid_edges[edge_id] = edge
        return valid_edges, invalid_edges

    @staticmethod
    def _style_nodes(
        graph: nx.MultiDiGraph | nx.DiGraph,
        *,
        force_labels: bool = False,
        partition_by_vertex: dict[str, str] | None = None,
        partition_color_map: dict[str, str] | None = None,
    ) -> None:
        """Apply common visual styles to graph nodes."""
        for node_id in graph.nodes():
            props = graph.nodes()[node_id]
            node_type = props.get("type")
            if node_type is None:
                logger.warning(
                    "Skipping node style; missing 'type' for node %s", node_id
                )
                continue

            shape = NODE_SHAPE_BY_TYPE.get(node_type)
            color = NODE_COLOR_BY_TYPE.get(node_type)
            if shape is None or color is None:
                logger.warning(
                    "Skipping node style; unsupported node type %s for node %s",
                    node_type,
                    node_id,
                )
                continue

            updates: dict[str, object] = {
                "shape": shape,
                "color": color,
                "fillcolor": color,
                "style": "filled",
            }
            if force_labels and "label" in props:
                updates["forcelabel"] = True

            if (
                partition_by_vertex
                and partition_color_map
                and isinstance(node_id, str)
                and node_id.startswith(f"{AuxNodeType.VERTEX}:")
            ):
                vertex_name = node_id.split(":", maxsplit=1)[1]
                group = partition_by_vertex.get(vertex_name)
                if group is not None and group in partition_color_map:
                    updates["fillcolor"] = partition_color_map[group]

            for key, value in updates.items():
                graph.nodes[node_id][key] = value

    @staticmethod
    def _style_edges(
        graph: nx.MultiDiGraph | nx.DiGraph,
        *,
        default_style: str = "solid",
        source_type_style_map: dict[AuxNodeType, str] | None = None,
    ) -> None:
        """Apply common visual styles to graph edges."""
        if graph.is_multigraph():
            edge_iterator = graph.edges(keys=True)
        else:
            edge_iterator = ((source, target, None) for source, target in graph.edges())

        for source, target, key in edge_iterator:
            if graph.is_multigraph():
                edge_data = graph.edges[source, target, key]
            else:
                edge_data = graph.edges[source, target]
            updates: dict[str, object] = {
                "arrowhead": "vee",
                "style": default_style,
            }
            if source_type_style_map is not None:
                source_props = graph.nodes[source]
                source_type = source_props.get("type")
                if source_type in source_type_style_map:
                    updates["style"] = source_type_style_map[source_type]
            if "label" in edge_data:
                updates["label"] = edge_data["label"]
            for key_name, value in updates.items():
                if graph.is_multigraph():
                    graph.edges[source, target, key][key_name] = value
                else:
                    graph.edges[source, target][key_name] = value

    @staticmethod
    def _add_partition_subgraphs(
        ag,
        graph: nx.MultiDiGraph | nx.DiGraph,
        partition_by_vertex: dict[str, str],
    ) -> None:
        """Add Graphviz subgraphs grouping vertices by partition labels."""
        partition_to_nodes: dict[str, list[str]] = {}
        for node_id in graph.nodes():
            if not isinstance(node_id, str) or not node_id.startswith(
                f"{AuxNodeType.VERTEX}:"
            ):
                continue
            vertex_name = node_id.split(":", maxsplit=1)[1]
            group = partition_by_vertex.get(vertex_name)
            if group is None:
                continue
            partition_to_nodes.setdefault(group, []).append(node_id)

        for group in sorted(partition_to_nodes):
            ag.add_subgraph(
                partition_to_nodes[group],
                name=f"cluster_partition_{group}",
                rank="same",
                label=str(group),
            )

    def _add_resource_vertex_links(
        self,
        graph: nx.MultiDiGraph,
        resource,
    ) -> None:
        """Add a resource node and its vertex links to a graph."""
        resource_id = get_auxnode_id(
            AuxNodeType.RESOURCE,
            resource=resource.name,
            resource_type="resource",
        )
        graph.add_node(
            resource_id,
            type=AuxNodeType.RESOURCE,
            label=resource.name,
        )

        vertex_reasons = self._extract_resource_vertex_reasons(resource)
        for vertex_name, reasons in sorted(vertex_reasons.items()):
            vertex_id = get_auxnode_id(AuxNodeType.VERTEX, vertex=vertex_name)
            graph.add_node(
                vertex_id,
                type=AuxNodeType.VERTEX,
                label=vertex_name,
            )
            graph.add_edge(resource_id, vertex_id, label=", ".join(sorted(reasons)))

    @staticmethod
    def _node_id_to_vertex(node_id: str) -> str:
        """Extract vertex name from an AuxNodeType.VERTEX node ID."""
        return node_id.split(":", maxsplit=1)[1]

    def _build_partition_color_map(
        self,
        partition_values: set[str],
        partition_colors: dict[str, str] | None,
    ) -> dict[str, str]:
        """Build a deterministic color map for partition labels."""
        color_map = {}
        if partition_colors is not None:
            color_map.update(partition_colors)

        auto_keys = sorted(partition_values - set(color_map.keys()))
        for idx, key in enumerate(auto_keys):
            color_map[key] = partition_color_palette[idx % len(partition_color_palette)]
        return color_map

    @staticmethod
    def _partition_legend(
        graph: nx.MultiDiGraph | nx.DiGraph,
        partition_by_vertex: dict[str, str],
        partition_color_map: dict[str, str],
    ) -> list[tuple[str, dict[str, str]]]:
        """Legend entries for the vertex groups drawn in *graph*, by group name."""
        groups = {
            partition_by_vertex[vertex]
            for node_id in graph.nodes()
            if isinstance(node_id, str)
            and node_id.startswith(f"{AuxNodeType.VERTEX}:")
            and (vertex := node_id.split(":", maxsplit=1)[1]) in partition_by_vertex
        }
        entries = []
        for group in sorted(groups, key=_group_order):
            if group not in partition_color_map:
                continue
            meaning = group
            if group == f"{_LEVEL_PREFIX}0":
                meaning = "level 0 (sources)"
            elif group.startswith(_LEVEL_PREFIX):
                meaning = f"level {group.removeprefix(_LEVEL_PREFIX)}"
            entries.append(
                (
                    meaning,
                    {
                        "shape": NODE_SHAPE_BY_TYPE[AuxNodeType.VERTEX],
                        "style": "filled",
                        "color": NODE_COLOR_BY_TYPE[AuxNodeType.VERTEX],
                        "fillcolor": partition_color_map[group],
                    },
                )
            )
        return entries

    def _infer_vertex_levels(self, edges: list[tuple[str, str]]) -> dict[str, int]:
        """Infer robust DAG-like levels by collapsing SCCs then layering the DAG."""
        graph = nx.DiGraph()
        graph.add_nodes_from(self.schema_graph.vertex_types)
        graph.add_edges_from(edges)

        condensation = nx.condensation(graph)
        component_level: dict[int, int] = {}
        for component in nx.topological_sort(condensation):
            predecessors = list(condensation.predecessors(component))
            if not predecessors:
                component_level[component] = 0
                continue
            component_level[component] = (
                max(component_level[pred] for pred in predecessors) + 1
            )

        levels = {}
        member_map = condensation.graph["mapping"]
        for vertex in graph.nodes:
            levels[vertex] = component_level[member_map[vertex]]
        return levels

    def plot_vc2fields(self):
        """Plot vertex collections and their fields.

        Creates a visualization showing the relationship between vertex collections
        and their fields, including index fields. The visualization is saved as
        a PDF file.
        """
        g = nx.DiGraph()
        nodes = []
        edges = []
        vconf = self.schema.core_schema.vertex_config
        vertex_prefix_dict = shortest_unique_prefix_map(
            [v for v in self.schema.core_schema.vertex_config.vertex_set]
        )

        kwargs = {"vfield": True, "vertex_sh": vertex_prefix_dict}
        for k in vconf.vertex_set:
            index_fields = vconf.identity_fields(k)
            fields = vconf.property_names(k)
            kwargs["vertex"] = k
            nodes_collection = [
                (
                    get_auxnode_id(AuxNodeType.VERTEX, **kwargs),
                    {
                        "type": AuxNodeType.VERTEX,
                        "label": get_auxnode_id(
                            AuxNodeType.VERTEX, label=True, **kwargs
                        ),
                    },
                )
            ]
            nodes_fields = [
                (
                    get_auxnode_id(AuxNodeType.FIELD, field=item, **kwargs),
                    {
                        "type": (
                            AuxNodeType.FIELD_DEFINITION
                            if item in index_fields
                            else AuxNodeType.FIELD
                        ),
                        "label": get_auxnode_id(
                            AuxNodeType.FIELD, field=item, label=True, **kwargs
                        ),
                    },
                )
                for item in fields
            ]
            nodes += nodes_collection
            nodes += nodes_fields
            edges += [(x[0], y[0]) for x, y in product(nodes_collection, nodes_fields)]

        g.add_nodes_from(nodes)
        g.add_edges_from(edges)

        self._style_nodes(g, force_labels=True)
        self._style_edges(g)

        ag = to_dot(g)

        for k in vconf.vertex_set:
            level_index = [
                get_auxnode_id(
                    AuxNodeType.FIELD,
                    vertex=k,
                    field=item,
                    vfield=True,
                    vertex_sh=vertex_prefix_dict,
                )
                for item in vconf.identity_fields(k)
            ]
            index_subgraph = ag.add_subgraph(level_index, name=f"cluster_{k}:def")
            index_subgraph.graph_attr["style"] = "filled"
            index_subgraph.graph_attr["label"] = "definition"

        ag = ag.unflatten(levels=5, fans=True, chain=3)
        add_legend(ag, type_legend(g))
        self._draw(ag, self._versioned_stem(f"{self.prefix}_vc2fields"))

    def plot_resources(self):
        """Plot resource relationships.

        Creates visualizations for each resource in the schema, showing their
        internal structure and relationships. Each resource is saved as a
        separate PDF file.
        """
        resource_prefix_dict = shortest_unique_prefix_map(
            [resource.name for resource in self.ingestion_model.resources]
        )
        vertex_prefix_dict = shortest_unique_prefix_map(
            [v for v in self.schema.core_schema.vertex_config.vertex_set]
        )
        kwargs = {"vertex_sh": vertex_prefix_dict, "resource_sh": resource_prefix_dict}

        for resource_config in self.ingestion_model.resources:
            kwargs["resource"] = resource_config.name
            assemble_tree(
                ActorWrapper(*resource_config.pipeline),
                self._figure_path(f"{self.name}.resource-{resource_config.name}"),
                output_format=self.output_format,
                output_dpi=self.output_dpi,
            )

    def plot_source2vc(self):
        """Plot source to vertex collection mappings.

        Creates a visualization showing the relationship between source resources
        and vertex collections. The visualization is saved as a PDF file.
        """
        g = nx.MultiDiGraph()
        for resource in self.ingestion_model.resources:
            self._add_resource_vertex_links(g, resource)

        self._style_nodes(g, force_labels=True)
        self._style_edges(g)

        ag = to_dot(g)
        ag.graph_attr["rankdir"] = "LR"
        ag.graph_attr["splines"] = "spline"

        resource_nodes = [
            get_auxnode_id(
                AuxNodeType.RESOURCE,
                resource=resource.name,
                resource_type="resource",
            )
            for resource in self.ingestion_model.resources
        ]
        vertex_nodes = [
            get_auxnode_id(AuxNodeType.VERTEX, vertex=v)
            for v in sorted(self.schema.core_schema.vertex_config.vertex_set)
            if get_auxnode_id(AuxNodeType.VERTEX, vertex=v) in g.nodes
        ]
        if resource_nodes:
            ag.add_subgraph(resource_nodes, name="cluster_resources", rank="same")
        if vertex_nodes:
            ag.add_subgraph(vertex_nodes, name="cluster_vertices", rank="same")
        add_legend(ag, type_legend(g))

        self._draw(ag, self._versioned_stem(f"{self.prefix}_source2vc"))

    def _extract_resource_vertex_reasons(self, resource) -> dict[str, set[str]]:
        """Collect vertex references for a resource with lightweight reason labels."""
        vertex_reasons: dict[str, set[str]] = {}
        known_vertices = set(self.schema.core_schema.vertex_config.vertex_set)
        actors = ActorWrapper(*resource.pipeline).collect_actors()

        def _add(vertex_name: str, reason: str) -> None:
            if vertex_name not in known_vertices:
                return
            if vertex_name not in vertex_reasons:
                vertex_reasons[vertex_name] = set()
            vertex_reasons[vertex_name].add(reason)

        for actor in actors:
            for vertex_name in actor.references_vertices():
                _add(vertex_name, type(actor).__name__)

            if isinstance(actor, VertexRouterActor):
                for vertex_name in actor.type_map.values():
                    _add(vertex_name, "VertexRouterActor(type_map)")
                for vertex_name in actor.vertex_from_map:
                    _add(vertex_name, "VertexRouterActor(vertex_from_map)")
                for vertex_name in actor.config.vertex_types or []:
                    _add(vertex_name, "VertexRouterActor(vertex_types)")

        return vertex_reasons

    def plot_source2vc_detailed(self):
        """Plot per-resource source-to-vertex mappings as dedicated detail pages."""
        for resource in self.ingestion_model.resources:
            g = nx.MultiDiGraph()
            self._add_resource_vertex_links(g, resource)

            self._style_nodes(g, force_labels=True)
            self._style_edges(g)

            ag = to_dot(g)
            ag.graph_attr["rankdir"] = "LR"
            add_legend(ag, type_legend(g))
            self._draw(
                ag,
                self._versioned_stem(f"{self.name}.resource2vc-{resource.name}"),
            )

    def plot_vc2vc(
        self,
        prune_leaves: bool = False,
        partition: dict[str, str] | None = None,
        partition_colors: dict[str, str] | None = None,
        color_by_partition: bool = False,
        group_by_partition: bool = False,
        include_all_vertices: bool = True,
        group_by_inferred_level: bool = False,
    ):
        """Plot vertex collection relationships.

        Creates a visualization showing the relationships between vertex collections.
        Optionally prunes leaf nodes from the visualization.

        This method discovers edges from both edge_config and resources to ensure
        all relationships are visualized, including those with dynamic relations.

        Args:
            prune_leaves: Whether to remove leaf nodes from the visualization
            partition: Optional vertex->group mapping used for coloring/grouping
            partition_colors: Optional group->color mapping for partition values
            color_by_partition: Whether to color nodes using partition values
            group_by_partition: Whether to add Graphviz clusters for each group
            include_all_vertices: Whether to include vertices with no known edges
            group_by_inferred_level: Group by inferred graph level (SCC-aware)

        Example:
            >>> plotter.plot_vc2vc(prune_leaves=True)
        """
        g = nx.MultiDiGraph()
        nodes: list[tuple[str, dict[str, object]]] = []
        rendered_edges: list[tuple] = []

        (
            discovered_edges,
            relation_source_by_edge_id,
            relation_from_key_by_edge_id,
        ) = self._discover_edges_from_resources()
        # Declared edges from the index; discovered ones supplement it, since
        # an edge that only exists in a resource pipeline is not in the schema.
        configured_edges = {
            edge_id: self.schema_graph.edge(edge_id)
            for edge_id in self.schema_graph.edge_ids
        }
        all_edges = self._merge_edges(configured_edges, discovered_edges)

        known_vertices = set(self.schema_graph.vertex_types)
        valid_edges, invalid_edges = self._filter_edges_with_known_vertices(
            all_edges,
            known_vertices,
        )
        if invalid_edges:
            sampled_invalid = sorted(invalid_edges)[:5]
            logger.error(
                "plot_vc2vc ignored %s edge(s) with unknown vertices not in vertex_config; sample=%s",
                len(invalid_edges),
                sampled_invalid,
            )

        edge_pairs = [(source, target) for (source, target, _relation) in valid_edges]
        for edge_id, edge in valid_edges.items():
            label = self._edge_label(
                edge,
                relation_source_field=relation_source_by_edge_id.get(edge_id),
                relation_from_key=relation_from_key_by_edge_id.get(edge_id, False),
            )
            source, target = edge_id[0], edge_id[1]
            source_id = get_auxnode_id(AuxNodeType.VERTEX, vertex=source)
            target_id = get_auxnode_id(AuxNodeType.VERTEX, vertex=target)
            if label is None:
                rendered_edges.append((source_id, target_id))
            else:
                rendered_edges.append((source_id, target_id, {"label": label}))

        # Create nodes for vertices involved in edges, optionally including all schema vertices
        vertices_in_edges = {
            vertex
            for (source, target, _relation) in valid_edges
            for vertex in (source, target)
        }
        all_vertices = (
            set(self.schema_graph.vertex_types)
            if include_all_vertices
            else vertices_in_edges
        )

        effective_partition = partition.copy() if partition is not None else {}
        if group_by_inferred_level:
            inferred_levels = self._infer_vertex_levels(edge_pairs)
            effective_partition.update(
                {
                    vertex: f"{_LEVEL_PREFIX}{level}"
                    for vertex, level in inferred_levels.items()
                }
            )

        for v in sorted(all_vertices):
            nodes += [
                (
                    get_auxnode_id(AuxNodeType.VERTEX, vertex=v),
                    {
                        "type": AuxNodeType.VERTEX,
                        "label": get_auxnode_id(
                            AuxNodeType.VERTEX, vertex=v, label=True
                        ),
                    },
                )
            ]

        g.add_nodes_from(nodes)
        g.add_edges_from(rendered_edges)

        if prune_leaves:
            out_deg = g.out_degree()
            in_deg = g.in_degree()

            nodes_to_remove = {k for k, v in out_deg if v == 0} & {
                k for k, v in in_deg if v < 2
            }
            g.remove_nodes_from(nodes_to_remove)

        partition_color_map = (
            self._build_partition_color_map(
                partition_values=set(effective_partition.values()),
                partition_colors=partition_colors,
            )
            if effective_partition and color_by_partition
            else {}
        )

        self._style_nodes(
            g,
            partition_by_vertex=effective_partition if partition_color_map else None,
            partition_color_map=partition_color_map if partition_color_map else None,
        )
        self._style_edges(g, source_type_style_map=EDGE_STYLE_BY_SOURCE_TYPE)

        ag = to_dot(g)
        if group_by_partition and effective_partition:
            self._add_partition_subgraphs(ag, g, effective_partition)
        if partition_color_map:
            legend = self._partition_legend(g, effective_partition, partition_color_map)
        else:
            legend = type_legend(g)
        add_legend(ag, legend)

        self._draw(ag, self._versioned_stem(f"{self.prefix}_vc2vc"))

Attributes

config = None instance-attribute

fig_path = fig_path instance-attribute

ingestion_model = manifest.require_ingestion_model() instance-attribute

name = Schema._slug_filename_token(self.schema.metadata.name) instance-attribute

output_dpi = output_dpi instance-attribute

output_format = output_format.lower() instance-attribute

prefix = self.name instance-attribute

schema = manifest.require_schema() instance-attribute

schema_graph = SchemaGraph.from_schema(self.schema) instance-attribute

Methods:

__init__(config_filename=None, fig_path='.', output_format='pdf', output_dpi=None, graph_manifest=None)

Initialize the schema plotter.

Parameters:

Name	Type	Description	Default
config_filename	str | PathLike[str] | None	Optional path to schema configuration file	None
fig_path	str | PathLike[str]	Path to save visualizations	'.'
graph_manifest	GraphManifest | None	Optional pre-built graph manifest	None

Source code in graflo/plot/plotter.py

def __init__(
    self,
    config_filename: str | os.PathLike[str] | None = None,
    fig_path: str | os.PathLike[str] = ".",
    output_format: str = "pdf",
    output_dpi: int | None = None,
    graph_manifest: GraphManifest | None = None,
):
    """Initialize the schema plotter.

    Args:
        config_filename: Optional path to schema configuration file
        fig_path: Path to save visualizations
        graph_manifest: Optional pre-built graph manifest
    """
    self.fig_path = fig_path
    self.output_format = output_format.lower()
    self.output_dpi = output_dpi

    manifest: GraphManifest
    if graph_manifest is not None:
        manifest = graph_manifest
        self.config = None
    elif config_filename is not None:
        self.config = FileHandle.load(config_filename)
        manifest = GraphManifest.from_config(self.config)
    else:
        raise ValueError(
            "ManifestPlotter requires either `config_filename` or `graph_manifest`."
        )

    manifest.finish_init()
    self.schema = manifest.require_schema()
    self.ingestion_model = manifest.require_ingestion_model()

    # A label, not a filename: merged names carry "+" and may carry spaces.
    self.name = Schema._slug_filename_token(self.schema.metadata.name)
    self.prefix = self.name
    # The one index over the schema's declared adjacency. Built here so
    # every plot reads the same traversal rather than each deriving its
    # own from `vertex_config` / `edge_config`.
    self.schema_graph = SchemaGraph.from_schema(self.schema)

plot_resources()

Plot resource relationships.

Creates visualizations for each resource in the schema, showing their internal structure and relationships. Each resource is saved as a separate PDF file.

Source code in graflo/plot/plotter.py

def plot_resources(self):
    """Plot resource relationships.

    Creates visualizations for each resource in the schema, showing their
    internal structure and relationships. Each resource is saved as a
    separate PDF file.
    """
    resource_prefix_dict = shortest_unique_prefix_map(
        [resource.name for resource in self.ingestion_model.resources]
    )
    vertex_prefix_dict = shortest_unique_prefix_map(
        [v for v in self.schema.core_schema.vertex_config.vertex_set]
    )
    kwargs = {"vertex_sh": vertex_prefix_dict, "resource_sh": resource_prefix_dict}

    for resource_config in self.ingestion_model.resources:
        kwargs["resource"] = resource_config.name
        assemble_tree(
            ActorWrapper(*resource_config.pipeline),
            self._figure_path(f"{self.name}.resource-{resource_config.name}"),
            output_format=self.output_format,
            output_dpi=self.output_dpi,
        )

plot_source2vc()

Plot source to vertex collection mappings.

Creates a visualization showing the relationship between source resources and vertex collections. The visualization is saved as a PDF file.

Source code in graflo/plot/plotter.py

def plot_source2vc(self):
    """Plot source to vertex collection mappings.

    Creates a visualization showing the relationship between source resources
    and vertex collections. The visualization is saved as a PDF file.
    """
    g = nx.MultiDiGraph()
    for resource in self.ingestion_model.resources:
        self._add_resource_vertex_links(g, resource)

    self._style_nodes(g, force_labels=True)
    self._style_edges(g)

    ag = to_dot(g)
    ag.graph_attr["rankdir"] = "LR"
    ag.graph_attr["splines"] = "spline"

    resource_nodes = [
        get_auxnode_id(
            AuxNodeType.RESOURCE,
            resource=resource.name,
            resource_type="resource",
        )
        for resource in self.ingestion_model.resources
    ]
    vertex_nodes = [
        get_auxnode_id(AuxNodeType.VERTEX, vertex=v)
        for v in sorted(self.schema.core_schema.vertex_config.vertex_set)
        if get_auxnode_id(AuxNodeType.VERTEX, vertex=v) in g.nodes
    ]
    if resource_nodes:
        ag.add_subgraph(resource_nodes, name="cluster_resources", rank="same")
    if vertex_nodes:
        ag.add_subgraph(vertex_nodes, name="cluster_vertices", rank="same")
    add_legend(ag, type_legend(g))

    self._draw(ag, self._versioned_stem(f"{self.prefix}_source2vc"))

plot_source2vc_detailed()

Plot per-resource source-to-vertex mappings as dedicated detail pages.

Source code in graflo/plot/plotter.py

def plot_source2vc_detailed(self):
    """Plot per-resource source-to-vertex mappings as dedicated detail pages."""
    for resource in self.ingestion_model.resources:
        g = nx.MultiDiGraph()
        self._add_resource_vertex_links(g, resource)

        self._style_nodes(g, force_labels=True)
        self._style_edges(g)

        ag = to_dot(g)
        ag.graph_attr["rankdir"] = "LR"
        add_legend(ag, type_legend(g))
        self._draw(
            ag,
            self._versioned_stem(f"{self.name}.resource2vc-{resource.name}"),
        )

plot_vc2fields()

Plot vertex collections and their fields.

Creates a visualization showing the relationship between vertex collections and their fields, including index fields. The visualization is saved as a PDF file.

Source code in graflo/plot/plotter.py

def plot_vc2fields(self):
    """Plot vertex collections and their fields.

    Creates a visualization showing the relationship between vertex collections
    and their fields, including index fields. The visualization is saved as
    a PDF file.
    """
    g = nx.DiGraph()
    nodes = []
    edges = []
    vconf = self.schema.core_schema.vertex_config
    vertex_prefix_dict = shortest_unique_prefix_map(
        [v for v in self.schema.core_schema.vertex_config.vertex_set]
    )

    kwargs = {"vfield": True, "vertex_sh": vertex_prefix_dict}
    for k in vconf.vertex_set:
        index_fields = vconf.identity_fields(k)
        fields = vconf.property_names(k)
        kwargs["vertex"] = k
        nodes_collection = [
            (
                get_auxnode_id(AuxNodeType.VERTEX, **kwargs),
                {
                    "type": AuxNodeType.VERTEX,
                    "label": get_auxnode_id(
                        AuxNodeType.VERTEX, label=True, **kwargs
                    ),
                },
            )
        ]
        nodes_fields = [
            (
                get_auxnode_id(AuxNodeType.FIELD, field=item, **kwargs),
                {
                    "type": (
                        AuxNodeType.FIELD_DEFINITION
                        if item in index_fields
                        else AuxNodeType.FIELD
                    ),
                    "label": get_auxnode_id(
                        AuxNodeType.FIELD, field=item, label=True, **kwargs
                    ),
                },
            )
            for item in fields
        ]
        nodes += nodes_collection
        nodes += nodes_fields
        edges += [(x[0], y[0]) for x, y in product(nodes_collection, nodes_fields)]

    g.add_nodes_from(nodes)
    g.add_edges_from(edges)

    self._style_nodes(g, force_labels=True)
    self._style_edges(g)

    ag = to_dot(g)

    for k in vconf.vertex_set:
        level_index = [
            get_auxnode_id(
                AuxNodeType.FIELD,
                vertex=k,
                field=item,
                vfield=True,
                vertex_sh=vertex_prefix_dict,
            )
            for item in vconf.identity_fields(k)
        ]
        index_subgraph = ag.add_subgraph(level_index, name=f"cluster_{k}:def")
        index_subgraph.graph_attr["style"] = "filled"
        index_subgraph.graph_attr["label"] = "definition"

    ag = ag.unflatten(levels=5, fans=True, chain=3)
    add_legend(ag, type_legend(g))
    self._draw(ag, self._versioned_stem(f"{self.prefix}_vc2fields"))

plot_vc2vc(prune_leaves=False, partition=None, partition_colors=None, color_by_partition=False, group_by_partition=False, include_all_vertices=True, group_by_inferred_level=False)

Plot vertex collection relationships.

Creates a visualization showing the relationships between vertex collections. Optionally prunes leaf nodes from the visualization.

This method discovers edges from both edge_config and resources to ensure all relationships are visualized, including those with dynamic relations.

Parameters:

Name	Type	Description	Default
prune_leaves	bool	Whether to remove leaf nodes from the visualization	False
partition	dict[str, str] | None	Optional vertex->group mapping used for coloring/grouping	None
partition_colors	dict[str, str] | None	Optional group->color mapping for partition values	None
color_by_partition	bool	Whether to color nodes using partition values	False
group_by_partition	bool	Whether to add Graphviz clusters for each group	False
include_all_vertices	bool	Whether to include vertices with no known edges	True
group_by_inferred_level	bool	Group by inferred graph level (SCC-aware)	False

Example

plotter.plot_vc2vc(prune_leaves=True)

Source code in graflo/plot/plotter.py

def plot_vc2vc(
    self,
    prune_leaves: bool = False,
    partition: dict[str, str] | None = None,
    partition_colors: dict[str, str] | None = None,
    color_by_partition: bool = False,
    group_by_partition: bool = False,
    include_all_vertices: bool = True,
    group_by_inferred_level: bool = False,
):
    """Plot vertex collection relationships.

    Creates a visualization showing the relationships between vertex collections.
    Optionally prunes leaf nodes from the visualization.

    This method discovers edges from both edge_config and resources to ensure
    all relationships are visualized, including those with dynamic relations.

    Args:
        prune_leaves: Whether to remove leaf nodes from the visualization
        partition: Optional vertex->group mapping used for coloring/grouping
        partition_colors: Optional group->color mapping for partition values
        color_by_partition: Whether to color nodes using partition values
        group_by_partition: Whether to add Graphviz clusters for each group
        include_all_vertices: Whether to include vertices with no known edges
        group_by_inferred_level: Group by inferred graph level (SCC-aware)

    Example:
        >>> plotter.plot_vc2vc(prune_leaves=True)
    """
    g = nx.MultiDiGraph()
    nodes: list[tuple[str, dict[str, object]]] = []
    rendered_edges: list[tuple] = []

    (
        discovered_edges,
        relation_source_by_edge_id,
        relation_from_key_by_edge_id,
    ) = self._discover_edges_from_resources()
    # Declared edges from the index; discovered ones supplement it, since
    # an edge that only exists in a resource pipeline is not in the schema.
    configured_edges = {
        edge_id: self.schema_graph.edge(edge_id)
        for edge_id in self.schema_graph.edge_ids
    }
    all_edges = self._merge_edges(configured_edges, discovered_edges)

    known_vertices = set(self.schema_graph.vertex_types)
    valid_edges, invalid_edges = self._filter_edges_with_known_vertices(
        all_edges,
        known_vertices,
    )
    if invalid_edges:
        sampled_invalid = sorted(invalid_edges)[:5]
        logger.error(
            "plot_vc2vc ignored %s edge(s) with unknown vertices not in vertex_config; sample=%s",
            len(invalid_edges),
            sampled_invalid,
        )

    edge_pairs = [(source, target) for (source, target, _relation) in valid_edges]
    for edge_id, edge in valid_edges.items():
        label = self._edge_label(
            edge,
            relation_source_field=relation_source_by_edge_id.get(edge_id),
            relation_from_key=relation_from_key_by_edge_id.get(edge_id, False),
        )
        source, target = edge_id[0], edge_id[1]
        source_id = get_auxnode_id(AuxNodeType.VERTEX, vertex=source)
        target_id = get_auxnode_id(AuxNodeType.VERTEX, vertex=target)
        if label is None:
            rendered_edges.append((source_id, target_id))
        else:
            rendered_edges.append((source_id, target_id, {"label": label}))

    # Create nodes for vertices involved in edges, optionally including all schema vertices
    vertices_in_edges = {
        vertex
        for (source, target, _relation) in valid_edges
        for vertex in (source, target)
    }
    all_vertices = (
        set(self.schema_graph.vertex_types)
        if include_all_vertices
        else vertices_in_edges
    )

    effective_partition = partition.copy() if partition is not None else {}
    if group_by_inferred_level:
        inferred_levels = self._infer_vertex_levels(edge_pairs)
        effective_partition.update(
            {
                vertex: f"{_LEVEL_PREFIX}{level}"
                for vertex, level in inferred_levels.items()
            }
        )

    for v in sorted(all_vertices):
        nodes += [
            (
                get_auxnode_id(AuxNodeType.VERTEX, vertex=v),
                {
                    "type": AuxNodeType.VERTEX,
                    "label": get_auxnode_id(
                        AuxNodeType.VERTEX, vertex=v, label=True
                    ),
                },
            )
        ]

    g.add_nodes_from(nodes)
    g.add_edges_from(rendered_edges)

    if prune_leaves:
        out_deg = g.out_degree()
        in_deg = g.in_degree()

        nodes_to_remove = {k for k, v in out_deg if v == 0} & {
            k for k, v in in_deg if v < 2
        }
        g.remove_nodes_from(nodes_to_remove)

    partition_color_map = (
        self._build_partition_color_map(
            partition_values=set(effective_partition.values()),
            partition_colors=partition_colors,
        )
        if effective_partition and color_by_partition
        else {}
    )

    self._style_nodes(
        g,
        partition_by_vertex=effective_partition if partition_color_map else None,
        partition_color_map=partition_color_map if partition_color_map else None,
    )
    self._style_edges(g, source_type_style_map=EDGE_STYLE_BY_SOURCE_TYPE)

    ag = to_dot(g)
    if group_by_partition and effective_partition:
        self._add_partition_subgraphs(ag, g, effective_partition)
    if partition_color_map:
        legend = self._partition_legend(g, effective_partition, partition_color_map)
    else:
        legend = type_legend(g)
    add_legend(ag, legend)

    self._draw(ag, self._versioned_stem(f"{self.prefix}_vc2vc"))

Functions:

add_legend(dot, entries)

Draw a boxed key in dot: one sample node per (meaning, node attributes).

Add it after any :meth:~graflo.plot.dot.DotGraph.unflatten, which would otherwise treat the samples as leaves of the figure.

Source code in graflo/plot/plotter.py

def add_legend(dot: DotGraph, entries: list[tuple[str, dict[str, str]]]) -> None:
    """Draw a boxed key in *dot*: one sample node per ``(meaning, node attributes)``.

    Add it after any :meth:`~graflo.plot.dot.DotGraph.unflatten`, which would
    otherwise treat the samples as leaves of the figure.
    """
    if not entries:
        return
    ids = []
    for index, (meaning, attrs) in enumerate(entries):
        node = f"legend__{index}"
        dot.add_node(node, **attrs, label=meaning, fontsize="10")
        ids.append(node)
    # Invisible edges stack the samples instead of spreading them along a rank.
    for tail, head in pairwise(ids):
        dot.add_edge(tail, head, style="invis")
    dot.add_subgraph(
        ids,
        name="cluster_legend",
        label="legend",
        style="dashed",
        color="grey50",
        fontsize="10",
    )

assemble_tree(aw, fig_path=None, output_format='pdf', output_dpi=None)

Assemble a tree visualization from an actor wrapper.

Parameters:

Name	Type	Description	Default
aw	ActorWrapper	Actor wrapper containing the tree structure	required
fig_path	Path | str | None	Optional path to save the visualization	None

Returns:

Type	Description
	nx.MultiDiGraph | None: The assembled graph if fig_path is None

Example

graph = assemble_tree(actor_wrapper) assemble_tree(actor_wrapper, "output/tree.pdf")

Source code in graflo/plot/plotter.py

def assemble_tree(
    aw: ActorWrapper,
    fig_path: Path | str | None = None,
    output_format: str = "pdf",
    output_dpi: int | None = None,
):
    """Assemble a tree visualization from an actor wrapper.

    Args:
        aw: Actor wrapper containing the tree structure
        fig_path: Optional path to save the visualization

    Returns:
        nx.MultiDiGraph | None: The assembled graph if fig_path is None

    Example:
        >>> graph = assemble_tree(actor_wrapper)
        >>> assemble_tree(actor_wrapper, "output/tree.pdf")
    """
    _, _, _, edges = aw.fetch_actors(0, [])
    logger.info(f"{len(edges)}")
    nodes = {}
    g = nx.MultiDiGraph()
    for ha, hb, pa, pb in edges:
        nodes[ha] = pa
        nodes[hb] = pb

    for n, props in nodes.items():
        nodes[n]["fillcolor"] = map_class2color[props["class"]]
        nodes[n]["style"] = "filled"
        nodes[n]["color"] = ACTOR_BORDER

    edges = [(ha, hb) for ha, hb, _, _ in edges]
    g.add_edges_from(edges)
    g.add_nodes_from(nodes.items())

    if fig_path is not None:
        ag = to_dot(g)
        present = {props["class"] for props in nodes.values()}
        add_legend(
            ag,
            [
                (
                    meaning,
                    {
                        "style": "filled",
                        "fillcolor": map_class2color[cls],
                        "color": ACTOR_BORDER,
                    },
                )
                for cls, meaning in ACTOR_LEGEND_BY_CLASS.items()
                if cls in present
            ],
        )
        draw(ag, fig_path, output_format=output_format, dpi=output_dpi)
        return None
    else:
        return g

get_auxnode_id(ntype, label=False, vfield=False, **kwargs)

Generate a unique identifier for an auxiliary node.

Parameters:

Name	Type	Description	Default
ntype	AuxNodeType	Type of the auxiliary node	required
label		Whether to generate a label instead of an ID	False
vfield		Whether this is a vertex field	False
**kwargs		Additional parameters for node identification	{}

Returns:

Name	Type	Description
str		Node identifier or label

Example

get_auxnode_id(AuxNodeType.VERTEX, vertex="user", label=True) 'user'

Source code in graflo/plot/plotter.py

def get_auxnode_id(ntype: AuxNodeType, label=False, vfield=False, **kwargs):
    """Generate a unique identifier for an auxiliary node.

    Args:
        ntype: Type of the auxiliary node
        label: Whether to generate a label instead of an ID
        vfield: Whether this is a vertex field
        **kwargs: Additional parameters for node identification

    Returns:
        str: Node identifier or label

    Example:
        >>> get_auxnode_id(AuxNodeType.VERTEX, vertex="user", label=True)
        'user'
    """
    vertex = kwargs.pop("vertex", None)
    resource = kwargs.pop("resource", None)
    vertex_shortcut = kwargs.pop("vertex_sh", None)
    resource_shortcut = kwargs.pop("resource_sh", None)
    s = "***"
    if ntype == AuxNodeType.RESOURCE:
        resource_type = kwargs.pop("resource_type")
        if label:
            s = f"{resource}"
        else:
            s = f"{ntype}:{resource_type}:{resource}"
    elif ntype == AuxNodeType.VERTEX:
        if label:
            s = f"{vertex}"
        else:
            s = f"{ntype}:{vertex}"
    elif ntype == AuxNodeType.FIELD:
        field = kwargs.pop("field", None)
        if vfield:
            if label:
                s = f"({vertex_shortcut[vertex]}){field}"
            else:
                s = f"{ntype}:{vertex}:{field}"
        else:
            if label:
                s = f"<{resource_shortcut[resource]}>{field}"
            else:
                s = f"{ntype}:{resource}:{field}"
    elif ntype == AuxNodeType.TRANSFORM:
        inputs = kwargs.pop("inputs")
        outputs = kwargs.pop("outputs")
        t_spec = inputs + outputs
        t_key = "-".join(t_spec)
        t_label = "-".join([x[0] for x in t_spec])

        if label:
            s = f"[t]{t_label}"
        else:
            s = f"transform:{t_key}"
    return s

lto_dict(strings)

Backward-compatible alias for shortest_unique_prefix_map.

Source code in graflo/plot/plotter.py

def lto_dict(strings):
    """Backward-compatible alias for shortest_unique_prefix_map."""
    return shortest_unique_prefix_map(strings)

shortest_unique_prefix_map(strings)

Create a dictionary of string prefixes for shortening labels.

Parameters:

Name	Type	Description	Default
strings		List of strings to process	required

Returns:

Name	Type	Description
dict		Mapping of shortened prefixes to original prefixes

Example

lto_dict(["user", "user_profile", "user_settings"])

Source code in graflo/plot/plotter.py

def shortest_unique_prefix_map(strings):
    """Create a dictionary of string prefixes for shortening labels.

    Args:
        strings: List of strings to process

    Returns:
        dict: Mapping of shortened prefixes to original prefixes

    Example:
        >>> lto_dict(["user", "user_profile", "user_settings"])
        {'user': 'user', 'user_p': 'user_', 'user_s': 'user_'}
    """
    strings = list(set(strings))
    d = {"": strings}
    while any(len(v) > 1 for v in d.values()):
        keys = list(d.keys())
        for k in keys:
            item = d.pop(k)
            if len(item) < 2:
                d[k] = item
            else:
                for s in item:
                    if s:
                        if k + s[0] in d:
                            d[k + s[0]].append(s[1:])
                        else:
                            d[k + s[0]] = [s[1:]]
                    else:
                        d[k] = [s]
    r = {}
    for k, v in d.items():
        if v:
            r[k + v[0]] = k
        else:
            r[k] = k
    return r

type_legend(graph)

Legend entries for the node types present in graph, in a fixed order.

Source code in graflo/plot/plotter.py

def type_legend(graph: nx.Graph) -> list[tuple[str, dict[str, str]]]:
    """Legend entries for the node types present in *graph*, in a fixed order."""
    present = {props.get("type") for _, props in graph.nodes(data=True)}
    return [
        (
            meaning,
            {
                "shape": NODE_SHAPE_BY_TYPE[node_type],
                "style": "filled",
                "color": NODE_COLOR_BY_TYPE[node_type],
                "fillcolor": NODE_COLOR_BY_TYPE[node_type],
            },
        )
        for node_type, meaning in NODE_LEGEND_BY_TYPE.items()
        if node_type in present
    ]