microsoft--graphrag/tests/unit/graphs/nx_stable_lcc.py

# Copyright (c) 2024 Microsoft Corporation.
# Licensed under the MIT License

"""NetworkX-based stable LCC utility, kept for side-by-side test comparisons.

This was originally at graphrag.index.utils.stable_lcc and has been moved here
because production code no longer uses it (superseded by the DataFrame-based
graphrag.graphs.stable_lcc).
"""

import html
from typing import Any, cast

import networkx as nx


def _largest_connected_component(graph: nx.Graph) -> nx.Graph:
    """Return the largest connected component of the graph (NX-based)."""
    graph = graph.copy()
    lcc_nodes = max(nx.connected_components(graph), key=len)
    return graph.subgraph(lcc_nodes).copy()


def stable_largest_connected_component(graph: nx.Graph) -> nx.Graph:
    """Return the largest connected component of the graph, with nodes and edges sorted in a stable way."""
    graph = graph.copy()
    graph = cast("nx.Graph", _largest_connected_component(graph))
    graph = normalize_node_names(graph)
    return _stabilize_graph(graph)


def _stabilize_graph(graph: nx.Graph) -> nx.Graph:
    """Ensure an undirected graph with the same relationships will always be read the same way."""
    fixed_graph = nx.DiGraph() if graph.is_directed() else nx.Graph()

    sorted_nodes = graph.nodes(data=True)
    sorted_nodes = sorted(sorted_nodes, key=lambda x: x[0])

    fixed_graph.add_nodes_from(sorted_nodes)
    edges = list(graph.edges(data=True))

    if not graph.is_directed():

        def _sort_source_target(edge):
            source, target, edge_data = edge
            if source > target:
                temp = source
                source = target
                target = temp
            return source, target, edge_data

        edges = [_sort_source_target(edge) for edge in edges]

    def _get_edge_key(source: Any, target: Any) -> str:
        return f"{source} -> {target}"

    edges = sorted(edges, key=lambda x: _get_edge_key(x[0], x[1]))

    fixed_graph.add_edges_from(edges)
    return fixed_graph


def normalize_node_names(graph: nx.Graph | nx.DiGraph) -> nx.Graph | nx.DiGraph:
    """Normalize node names."""
    node_mapping = {node: html.unescape(node.upper().strip()) for node in graph.nodes()}  # type: ignore
    return nx.relabel_nodes(graph, node_mapping)