Various utils

Flowpaths implements various helper functions on graphs. They can be access with the prefix flowpaths.utils.

For example, you can create drawing as this one

using the following code:

import flowpaths as fp
import networkx as nx

# Create a simple graph
graph = nx.DiGraph()
graph.graph["id"] = "simple_graph"
graph.add_edge("s", "a", flow=6)
graph.add_edge("s", "b", flow=7)
graph.add_edge("a", "b", flow=2)
graph.add_edge("a", "c", flow=5)
graph.add_edge("b", "c", flow=9)
graph.add_edge("c", "d", flow=6)
graph.add_edge("c", "t", flow=7)
graph.add_edge("d", "t", flow=6)

# Solve the minimum path error model
mpe_model = fp.kMinPathError(graph, flow_attr="flow", k=3, weight_type=float)
mpe_model.solve()

# Draw the solution
if mpe_model.is_solved():
    solution = mpe_model.get_solution()
    fp.utils.draw_solution(
        G=graph,
        filename="simple_graph.pdf",
        flow_attr="flow",
        paths=solution["paths"],
        weights=solution["weights"],
        draw_options={
        "show_graph_edges": True,
        "show_edge_weights": False,
        "show_path_weights": False,
        "show_path_weight_on_first_edge": True,
        "pathwidth": 2,
    })

This produces a file with extension .pdf storing the PDF image of the graph.

check_flow_conservation

check_flow_conservation(
    G: DiGraph, flow_attr
) -> bool

Check if the flow conservation property holds for the given graph.

Parameters

• G: nx.DiGraph

  The input directed acyclic graph, as networkx DiGraph.

• flow_attr: str

  The attribute name from where to get the flow values on the edges.

Returns

• bool:

  True if the flow conservation property holds, False otherwise.

Source code in flowpaths/utils/graphutils.py

def check_flow_conservation(G: nx.DiGraph, flow_attr) -> bool:
    """
    Check if the flow conservation property holds for the given graph.

    Parameters
    ----------
    - `G`: nx.DiGraph

        The input directed acyclic graph, as networkx DiGraph.

    - `flow_attr`: str

        The attribute name from where to get the flow values on the edges.

    Returns
    -------

    - bool: 

        True if the flow conservation property holds, False otherwise.
    """

    for v in G.nodes():
        if G.out_degree(v) == 0 or G.in_degree(v) == 0:
            continue

        out_flow = 0
        for x, y, data in G.out_edges(v, data=True):
            if data.get(flow_attr) is None:
                return False
            out_flow += data[flow_attr]

        in_flow = 0
        for x, y, data in G.in_edges(v, data=True):
            if data.get(flow_attr) is None:
                return False
            in_flow += data[flow_attr]

        if out_flow != in_flow:
            return False

    return True

draw_solution

draw_solution(
    G: DiGraph,
    filename: str,
    flow_attr: str = None,
    paths: list = [],
    weights: list = [],
    additional_starts: list = [],
    additional_ends: list = [],
    subpath_constraints: list = [],
    draw_options: dict = {
        "show_graph_edges": True,
        "show_edge_weights": False,
        "show_node_weights": False,
        "show_path_weights": False,
        "show_path_weight_on_first_edge": True,
        "pathwidth": 3.0,
    },
)

Draw the graph with the paths and their weights highlighted.

Parameters

• G: nx.DiGraph

  The input directed acyclic graph, as networkx DiGraph.

• filename: str

  The name of the file to save the drawing. The file type is inferred from the extension. Supported extensions are ‘.bmp’, ‘.canon’, ‘.cgimage’, ‘.cmap’, ‘.cmapx’, ‘.cmapx_np’, ‘.dot’, ‘.dot_json’, ‘.eps’, ‘.exr’, ‘.fig’, ‘.gd’, ‘.gd2’, ‘.gif’, ‘.gtk’, ‘.gv’, ‘.ico’, ‘.imap’, ‘.imap_np’, ‘.ismap’, ‘.jp2’, ‘.jpe’, ‘.jpeg’, ‘.jpg’, ‘.json’, ‘.json0’, ‘.pct’, ‘.pdf’, ‘.pic’, ‘.pict’, ‘.plain’, ‘.plain-ext’, ‘.png’, ‘.pov’, ‘.ps’, ‘.ps2’, ‘.psd’, ‘.sgi’, ‘.svg’, ‘.svgz’, ‘.tga’, ‘.tif’, ‘.tiff’, ‘.tk’, ‘.vml’, ‘.vmlz’, ‘.vrml’, ‘.wbmp’, ‘.webp’, ‘.x11’, ‘.xdot’, ‘.xdot1.2’, ‘.xdot1.4’, ‘.xdot_json’, ‘.xlib’

• flow_attr: str

  The attribute name from where to get the flow values on the edges. Default is an empty string, in which case no edge weights are shown.

• paths: list

  The list of paths to highlight, as lists of nodes. Default is an empty list, in which case no path is drawn. Default is an empty list.

• weights: list

  The list of weights corresponding to the paths, of various colors. Default is an empty list, in which case no path is drawn.

• additional_starts: list

  A list of additional nodes to highlight in green as starting nodes. Default is an empty list.
  

• additional_ends: list

  A list of additional nodes to highlight in red as ending nodes. Default is an empty list.
  

• subpath_constraints: list

  A list of subpaths to highlight in the graph as dashed edges, of various colors. Each subpath is a list of edges. Default is an empty list. There is no association between the subpath colors and the path colors.

• draw_options: dict

  A dictionary with the following keys:

  • show_graph_edges: bool

    Whether to show the edges of the graph. Default is True.

  • show_edge_weights: bool

    Whether to show the edge weights in the graph from the flow_attr. Default is False.

  • show_node_weights: bool

    Whether to show the node weights in the graph from the flow_attr. Default is False.

  • show_path_weights: bool

    Whether to show the path weights in the graph on every edge. Default is False.

  • show_path_weight_on_first_edge: bool

    Whether to show the path weight on the first edge of the path. Default is True.

  • pathwidth: float

    The width of the path to be drawn. Default is 3.0.

Source code in flowpaths/utils/graphutils.py

def draw_solution(
        G: nx.DiGraph, 
        filename: str,
        flow_attr: str = None,
        paths: list = [], 
        weights: list = [], 
        additional_starts: list = [],
        additional_ends: list = [],
        subpath_constraints: list = [],
        draw_options: dict = {
            "show_graph_edges": True,
            "show_edge_weights": False,
            "show_node_weights": False,
            "show_path_weights": False,
            "show_path_weight_on_first_edge": True,
            "pathwidth": 3.0
        },
        ):
        """
        Draw the graph with the paths and their weights highlighted.

        Parameters
        ----------

        - `G`: nx.DiGraph 

            The input directed acyclic graph, as networkx DiGraph. 

        - `filename`: str

            The name of the file to save the drawing. The file type is inferred from the extension. Supported extensions are '.bmp', '.canon', '.cgimage', '.cmap', '.cmapx', '.cmapx_np', '.dot', '.dot_json', '.eps', '.exr', '.fig', '.gd', '.gd2', '.gif', '.gtk', '.gv', '.ico', '.imap', '.imap_np', '.ismap', '.jp2', '.jpe', '.jpeg', '.jpg', '.json', '.json0', '.pct', '.pdf', '.pic', '.pict', '.plain', '.plain-ext', '.png', '.pov', '.ps', '.ps2', '.psd', '.sgi', '.svg', '.svgz', '.tga', '.tif', '.tiff', '.tk', '.vml', '.vmlz', '.vrml', '.wbmp', '.webp', '.x11', '.xdot', '.xdot1.2', '.xdot1.4', '.xdot_json', '.xlib'

        - `flow_attr`: str

            The attribute name from where to get the flow values on the edges. Default is an empty string, in which case no edge weights are shown.

        - `paths`: list

            The list of paths to highlight, as lists of nodes. Default is an empty list, in which case no path is drawn. Default is an empty list.

        - `weights`: list

            The list of weights corresponding to the paths, of various colors. Default is an empty list, in which case no path is drawn.

        - `additional_starts`: list

                A list of additional nodes to highlight in green as starting nodes. Default is an empty list.

        - `additional_ends`: list

                A list of additional nodes to highlight in red as ending nodes. Default is an empty list.

        - `subpath_constraints`: list

            A list of subpaths to highlight in the graph as dashed edges, of various colors. Each subpath is a list of edges. Default is an empty list. There is no association between the subpath colors and the path colors.

        - `draw_options`: dict

            A dictionary with the following keys:

            - `show_graph_edges`: bool

                Whether to show the edges of the graph. Default is `True`.

            - `show_edge_weights`: bool

                Whether to show the edge weights in the graph from the `flow_attr`. Default is `False`.

            - `show_node_weights`: bool

                Whether to show the node weights in the graph from the `flow_attr`. Default is `False`.

            - `show_path_weights`: bool

                Whether to show the path weights in the graph on every edge. Default is `False`.

            - `show_path_weight_on_first_edge`: bool

                Whether to show the path weight on the first edge of the path. Default is `True`.

            - `pathwidth`: float

                The width of the path to be drawn. Default is `3.0`.

        """

        if len(paths) != len(weights):
            raise ValueError(f"{__name__}: Paths and weights must have the same length.")
            raise ValueError("Paths and weights must have the same length, if provided.")

        try:
            import graphviz as gv

            dot = gv.Digraph(format="pdf")
            dot.graph_attr["rankdir"] = "LR"  # Display the graph in landscape mode
            dot.node_attr["shape"] = "rectangle"  # Rectangle nodes
            dot.node_attr["style"] = "rounded"  # Rounded rectangle nodes

            colors = [
                "red",
                "blue",
                "green",
                "purple",
                "brown",
                "cyan",
                "yellow",
                "pink",
                "grey",
                "chocolate",
                "darkblue",
                "darkolivegreen",
                "darkslategray",
                "deepskyblue2",
                "cadetblue3",
                "darkmagenta",
                "goldenrod1"
            ]

            dot.attr('node', fontname='Arial')

            if draw_options.get("show_graph_edges", True):
                # drawing nodes
                for node in G.nodes():
                    color = "black"
                    penwidth = "1.0"
                    if node in additional_starts:
                        color = "green"
                        penwidth = "2.0"
                    elif node in additional_ends:
                        color = "red"
                        penwidth = "2.0"

                    if draw_options.get("show_node_weights", False) and flow_attr is not None and flow_attr in G.nodes[node]:
                        dot.node(
                            name=str(node),
                            #label=f"{{{node} | {G.nodes[node][flow_attr]}}}",
                            label=f"{G.nodes[node][flow_attr]}\\n{node}",
                            #xlabel=str(G.nodes[node][flow_attr]),
                            shape="record",
                            color=color, 
                            penwidth=penwidth)
                    else:
                        dot.node(
                            name=str(node), 
                            label=str(node), 
                            color=color, 
                            penwidth=penwidth)

                # drawing edges
                for u, v, data in G.edges(data=True):
                    if draw_options.get("show_edge_weights", False):
                        dot.edge(
                            tail_name=str(u), 
                            head_name=str(v), 
                            label=str(data.get(flow_attr,"")),
                            fontname="Arial",)
                    else:
                        dot.edge(
                            tail_name=str(u), 
                            head_name=str(v))

            for index, path in enumerate(paths):
                pathColor = colors[index % len(colors)]
                for i in range(len(path) - 1):
                    if i == 0 and draw_options.get("show_path_weight_on_first_edge", True) or \
                        draw_options.get("show_path_weights", True):
                        dot.edge(
                            str(path[i]),
                            str(path[i + 1]),
                            fontcolor=pathColor,
                            color=pathColor,
                            penwidth=str(draw_options.get("pathwidth", 3.0)),
                            label=str(weights[index]),
                            fontname="Arial",
                        )
                    else:
                        dot.edge(
                            str(path[i]),
                            str(path[i + 1]),
                            color=pathColor,
                            penwidth=str(draw_options.get("pathwidth", 3.0)),
                            )

            for index, path in enumerate(subpath_constraints):
                pathColor = colors[index % len(colors)]
                for i in range(len(path)):
                    if len(path[i]) != 2:
                        utils.logger.error(f"{__name__}: Subpaths must be lists of edges.")
                        raise ValueError("Subpaths must be lists of edges.")
                    dot.edge(
                        str(path[i][0]),
                        str(path[i][1]),
                        color=pathColor,
                        style="dashed",
                        penwidth="2.0"
                        )

                if len(path) == 1:
                    dot.node(str(path[0]), color=pathColor, penwidth=str(draw_options.get("pathwidth", 3.0)))

            dot.render(outfile=filename, view=False, cleanup=True)

        except ImportError:
            utils.logger.error(f"{__name__}: graphviz module not found. Please install it via pip (pip install graphviz).")
            raise ImportError("graphviz module not found. Please install it via pip (pip install graphviz).")

fpid

fpid(G) -> str

Returns a unique identifier for the given graph.

Source code in flowpaths/utils/graphutils.py

def fpid(G) -> str:
    """
    Returns a unique identifier for the given graph.
    """
    if isinstance(G, nx.DiGraph):
        if "id" in G.graph:
            return G.graph["id"]

    return str(id(G))

get_subgraph_between_topological_nodes

get_subgraph_between_topological_nodes(
    graph: DiGraph,
    topo_order: list,
    left: int,
    right: int,
) -> nx.DiGraph

Create a subgraph with the nodes between left and right in the topological order, including the edges between them, but also the edges from these nodes that are incident to nodes outside this range.

Source code in flowpaths/utils/graphutils.py

def get_subgraph_between_topological_nodes(graph: nx.DiGraph, topo_order: list, left: int, right: int) -> nx.DiGraph:
    """
    Create a subgraph with the nodes between left and right in the topological order, 
    including the edges between them, but also the edges from these nodes that are incident to nodes outside this range.
    """

    if left < 0 or right >= len(topo_order):
        utils.logger.error(f"{__name__}: Invalid range for topological order: {left}, {right}.")
        raise ValueError("Invalid range for topological order")
    if left > right:
        utils.logger.error(f"{__name__}: Invalid range for topological order: {left}, {right}.")
        raise ValueError("Invalid range for topological order")

    # Create a subgraph with the nodes between left and right in the topological order
    subgraph = nx.DiGraph()
    if "id" in graph.graph:
        subgraph.graph["id"] = graph.graph["id"]
    for i in range(left, right):
        subgraph.add_node(topo_order[i], **graph.nodes[topo_order[i]])

    fixed_nodes = set(subgraph.nodes())

    # Add the edges between the nodes in the subgraph
    for u, v in graph.edges():
        if u in fixed_nodes or v in fixed_nodes:
            subgraph.add_edge(u, v, **graph[u][v])
            if u not in fixed_nodes:
                subgraph.add_node(u, **graph.nodes[u])
            if v not in fixed_nodes:
                subgraph.add_node(v, **graph.nodes[v])

    return subgraph

max_bottleneck_path

max_bottleneck_path(
    G: DiGraph, flow_attr
) -> tuple

Computes the maximum bottleneck path in a directed graph.

Parameters

• G: nx.DiGraph

  A directed graph where each edge has a flow attribute.

• flow_attr: str

  The flow attribute from where to get the flow values.

Returns

• tuple: A tuple containing:

  • The value of the maximum bottleneck.
  • The path corresponding to the maximum bottleneck (list of nodes). If no s-t flow exists in the network, returns (None, None).

Source code in flowpaths/utils/graphutils.py

def max_bottleneck_path(G: nx.DiGraph, flow_attr) -> tuple:
    """
    Computes the maximum bottleneck path in a directed graph.

    Parameters
    ----------
    - `G`: nx.DiGraph

        A directed graph where each edge has a flow attribute.

    - `flow_attr`: str

        The flow attribute from where to get the flow values.

    Returns
    --------

    - tuple: A tuple containing:

        - The value of the maximum bottleneck.
        - The path corresponding to the maximum bottleneck (list of nodes).
            If no s-t flow exists in the network, returns (None, None).
    """
    B = dict()
    maxInNeighbor = dict()
    maxBottleneckSink = None

    # Computing the B values with DP
    for v in nx.topological_sort(G):
        if G.in_degree(v) == 0:
            B[v] = float("inf")
        else:
            B[v] = float("-inf")
            for u in G.predecessors(v):
                uBottleneck = min(B[u], G.edges[u, v][flow_attr])
                if uBottleneck > B[v]:
                    B[v] = uBottleneck
                    maxInNeighbor[v] = u
            if G.out_degree(v) == 0:
                if maxBottleneckSink is None or B[v] > B[maxBottleneckSink]:
                    maxBottleneckSink = v

    # If no s-t flow exists in the network
    if B[maxBottleneckSink] == 0:
        return None, None

    # Recovering the path of maximum bottleneck
    reverse_path = [maxBottleneckSink]
    while G.in_degree(reverse_path[-1]) > 0:
        reverse_path.append(maxInNeighbor[reverse_path[-1]])

    return B[maxBottleneckSink], list(reversed(reverse_path))

max_occurrence

max_occurrence(
    seq,
    paths_in_DAG,
    edge_lengths: dict = {},
) -> int

Check what is the maximum number of edges of seq that appear in some path in the list paths_in_DAG.

This assumes paths_in_DAG are paths in a directed acyclic graph.

Parameters

• seq (list): The sequence of edges to check.
• paths (list): The list of paths to check against, as lists of nodes.

Returns

• int: the largest number of seq edges that appear in some path in paths_in_DAG

Source code in flowpaths/utils/graphutils.py

def max_occurrence(seq, paths_in_DAG, edge_lengths: dict = {}) -> int:
    """
    Check what is the maximum number of edges of seq that appear in some path in the list paths_in_DAG. 

    This assumes paths_in_DAG are paths in a directed acyclic graph. 

    Parameters
    ----------
    - seq (list): The sequence of edges to check.
    - paths (list): The list of paths to check against, as lists of nodes.

    Returns
    -------
    - int: the largest number of seq edges that appear in some path in paths_in_DAG
    """
    max_occurence = 0
    for path in paths_in_DAG:
        path_edges = set([(path[i], path[i + 1]) for i in range(len(path) - 1)])
        # Check how many seq edges are in path_edges
        occurence = 0
        for edge in seq:
            if edge in path_edges:
                occurence += edge_lengths.get(edge, 1)
        if occurence > max_occurence:
            max_occurence = occurence

    return max_occurence