Source code for superblockify.partitioning.checks

"""Checks for the partitioning module."""

from itertools import chain

from networkx import is_weakly_connected
from numpy import argwhere, fill_diagonal
from osmnx import plot_graph

from ..config import logger
from ..plot import plot_by_attribute
from ..utils import has_pairwise_overlap


[docs] def is_valid_partitioning(partitioning): """Check if a partitioning is valid. The components of a partitioning are the subgraphs, a special subgraph is the sparsified graph. A partitioning is valid if the following conditions are met: 1. The sparsified graph is connected 2. Each subgraph is connected 3. Each node is contained in exactly one subgraph and not the sparsified graph 4. Each edge is contained in exactly one subgraph and not the sparsified graph 5. No node or edge of `graph` is not contained in any subgraph or the sparsified graph 6. Each subgraph is connected to the sparsified graph 7. Representative nodes are contained in their respective subgraph Parameters ---------- partitioning : partitioning.base.BasePartitioner Partitioning to check. Returns ------- bool Whether the partitioning is valid """ is_valid = True # 1. Check if the sparsified graph is connected logger.debug( "Checking if the sparsified graph of %s is connected.", partitioning.name ) if not is_weakly_connected(partitioning.sparsified): logger.warning( "The sparsified graph of %s is not connected.", partitioning.name ) is_valid = False # 2. Check if each subgraph is connected logger.debug("Checking if each subgraph of %s is connected.", partitioning.name) if not components_are_connected(partitioning): is_valid = False # 3. - 5. For every node and edge in the graph, check if it is contained in exactly # one subgraph and not the sparsified graph logger.debug( "Checking if each node and edge of %s is contained in exactly one subgraph " "and not the sparsified graph.", partitioning.name, ) if not nodes_and_edges_are_contained_in_exactly_one_subgraph(partitioning): is_valid = False # 6. Check if each subgraph is connected to the sparsified graph logger.debug( "Checking if each subgraph of %s is connected to the sparsified graph.", partitioning.name, ) if not components_are_connect_sparsified(partitioning): is_valid = False # 7. Check if representative nodes are contained in their subgraph logger.debug( "Checking if representative nodes of %s are contained in their subgraph.", partitioning.name, ) if not representative_nodes_are_contained_in_subgraph(partitioning): is_valid = False if is_valid: logger.info("The partitioning %s is valid.", partitioning.name) else: logger.error( "The partitioning %s is not valid. See warnings for more information.", partitioning.name, ) return is_valid
[docs] def components_are_connected(partitioning): """Check if each component is connected to the sparsified graph. Parameters ---------- partitioning : partitioning.base.BasePartitioner Partitioning to check. Returns ------- bool Whether each component is connected to the sparsified graph. """ found = True for component in partitioning.get_ltns(): if not is_weakly_connected(component["subgraph"]): logger.warning( "The subgraph %s of %s is not connected.", component["name"], partitioning.name, ) # Plot graph with highlighted subgraph # Write to attribute 'highlight' 1 if node is in subgraph, 0 otherwise for edge in component["subgraph"].edges: partitioning.graph.edges[edge]["highlight"] = 1 plot_by_attribute( partitioning.graph, edge_attr="highlight", edge_attr_types="numerical", edge_cmap="hsv", edge_minmax_val=(0, 1), ) # Reset edge attribute 'highlight' for edge in component["subgraph"].edges: partitioning.graph.edges[edge]["highlight"] = None found = False return found
[docs] def nodes_and_edges_are_contained_in_exactly_one_subgraph(partitioning): """Check if each node and edge is contained in exactly one subgraph. Edges can also be contained in the sparsified graph. Use :func:`superblockify.utils.has_pairwise_overlap` to check subgraphs overlap. Parameters ---------- partitioning : partitioning.base.BasePartitioner Partitioning to check. Returns ------- bool Whether each node and edge is contained in exactly one subgraph. """ is_valid = True partition_nodes_subgraphs = partitioning.get_partition_nodes() # Check overlap of nodes nodes_overlap_matrix = has_pairwise_overlap( [list(part["nodes"]) for part in partition_nodes_subgraphs] ) fill_diagonal(nodes_overlap_matrix, False) if nodes_overlap_matrix.any(): logger.warning( "The nodes of %s overlap in the following subgraphs: %s", partitioning.name, [ f"{partition_nodes_subgraphs[i]['name']} and " f"{partition_nodes_subgraphs[j]['name']}" for i, j in argwhere(nodes_overlap_matrix) # only lower triangle, because matrix is symmetric if i < j ], ) is_valid = False # Nodes that are in no subgraph, neither the sparsified graph missing_nodes = ( set(partitioning.graph.nodes) - set( # flattened set of all nodes in all subgraphs chain.from_iterable(part["nodes"] for part in partition_nodes_subgraphs) ) - set(partitioning.sparsified.nodes) ) if len(missing_nodes) > 0: logger.warning( "%s has %d nodes that are not contained in any subgraph or the " "sparsified graph: %s", partitioning.name, len(missing_nodes), missing_nodes, ) # Write to attribute 'missing_nodes' 1 if node is in missing_nodes, 0 otherwise for node in partitioning.graph.nodes: partitioning.graph.nodes[node]["missing_nodes"] = node in missing_nodes plot_graph( partitioning.graph, node_color=[ "red" if partitioning.graph.nodes[node]["missing_nodes"] else "none" for node in partitioning.graph.nodes ], bgcolor="none", ) is_valid = False # Check overlap of edges, including sparsified graph edges_overlap_matrix = has_pairwise_overlap( [list(part["subgraph"].edges) for part in partition_nodes_subgraphs] + [list(partitioning.sparsified.edges)] ) fill_diagonal(edges_overlap_matrix, False) if edges_overlap_matrix.any(): namelist = [part["name"] for part in partition_nodes_subgraphs] + ["sparse"] logger.warning( "The edges of %s overlap in the following subgraphs: %s", partitioning.name, [ f"{namelist[i]} and {namelist[j]}" for i, j in argwhere(edges_overlap_matrix) # only lower triangle, because matrix is symmetric if i < j ], ) is_valid = False # Edges that are in no subgraph, neither the sparsified graph missing_edges = ( set(partitioning.graph.edges) - set( # flattened set of all edges in all subgraphs chain.from_iterable( part["subgraph"].edges for part in partition_nodes_subgraphs ) ) - set(partitioning.sparsified.edges) ) if len(missing_edges) > 0: logger.warning( "%s has %d edges that are not contained in any subgraph or the " "sparsified graph: %s", partitioning.name, len(missing_edges), missing_edges, ) is_valid = False return is_valid
[docs] def components_are_connect_sparsified(partitioning): """Check if each subgraph is connected to the sparsified graph. Parameters ---------- partitioning : partitioning.base.BasePartitioner Partitioning to check. Returns ------- bool Whether each subgraph is connected to the sparsified graph """ for component in partitioning.get_ltns(): # subgraph and sparsified graph are connected if there is at least one node # that is contained in both if not any( node in component["subgraph"].nodes and node in partitioning.sparsified.nodes for node in partitioning.graph.nodes ): logger.warning( "The subgraph %s of %s is not connected to the sparsified graph.", component["name"], partitioning.name, ) return False return True
[docs] def representative_nodes_are_contained_in_subgraph(partitioning): """Check if representative nodes are contained in their subgraph. Parameters ---------- partitioning : partitioning.base.BasePartitioner Partitioning to check. Returns ------- bool Whether representative nodes are contained in their subgraph. """ partition_nodes_subgraphs = partitioning.get_partition_nodes() for component in partition_nodes_subgraphs: if not component["rep_node"] in component["subgraph"].nodes: logger.warning( "The representative node %s of %s is not contained in the " "subgraph %s.", component["rep_node"], partitioning.name, component["name"], ) return False return True