Various utils

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

Graph visualization and drawing

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(
        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.

Sankey Diagram Visualization

For acyclic graphs (DAGs), you can create interactive Sankey diagrams using plotly. Sankey diagrams are particularly effective for visualizing flow decompositions, as they show:

• Each node in the graph as a labeled box
• Each path as a colored flow whose width represents the path weight

To create a Sankey diagram, set "style": "sankey" in the draw_options:

import flowpaths as fp
import networkx as nx

# Create a sample DAG
G = nx.DiGraph()
G.add_edge('s', 'a', flow=10)
G.add_edge('s', 'b', flow=5)
G.add_edge('a', 'c', flow=6)
G.add_edge('a', 'd', flow=4)
G.add_edge('b', 'c', flow=3)
G.add_edge('b', 'd', flow=2)
G.add_edge('c', 't', flow=9)
G.add_edge('d', 't', flow=6)

# Compute minimum flow decomposition
solver = fp.MinFlowDecomp(G, flow_attr='flow')
solver.solve()
solution = solver.get_solution()

# Draw as interactive Sankey diagram
fp.utils.draw(
    G=G,
    filename="flow_sankey.html",  # saves as HTML (interactive)
    flow_attr='flow',
    paths=solution['paths'],
    weights=solution['weights'],
    draw_options={
        "style": "sankey"
    }
)

Features:

• Interactive: Hover over nodes and links to see details, zoom and pan the diagram
• Jupyter support: Automatically displays inline when run in Jupyter notebooks
• Dual output: Automatically saves both HTML (interactive) and a static image (PDF by default)
• Automatic coloring: Each path gets a distinct color; shared edges show blended colors
• Graph identification: Uses the graph’s ID as the diagram title if available

Requirements:

• plotly: Installed automatically with flowpaths
• kaleido: Installed automatically with flowpaths for static image export

File formats:

The function automatically saves both formats: - HTML file (interactive): [basename].html - Static image: [basename].pdf (or .png, .svg if specified)

# Saves both output.html and output.pdf
fp.utils.draw(G, "output", paths=paths, weights=weights, 
              draw_options={"style": "sankey"})

# Saves both flow.html and flow.png
fp.utils.draw(G, "flow.png", paths=paths, weights=weights,
              draw_options={"style": "sankey"})

# Saves both diagram.html and diagram.svg
fp.utils.draw(G, "diagram.svg", paths=paths, weights=weights,
              draw_options={"style": "sankey"})

Note: Sankey diagrams require the graph to be acyclic (DAG). If the graph contains cycles, use the traditional graphviz rendering ("style": "default" or "style": "points").

See examples/sankey_demo.py and examples/sankey_demo.ipynb for complete examples.

Logging

flowpaths exposes a simple logging helper via fp.utils.configure_logging. Use it to control verbosity, enable console/file logging, and set file mode.

Basic usage (console logging at INFO level):

import flowpaths as fp

fp.utils.configure_logging(
    level=fp.utils.logging.INFO,
    log_to_console=True,
)

Also log to a file (append mode):

fp.utils.configure_logging(
    level=fp.utils.logging.DEBUG,      # default is DEBUG
    log_to_console=True,               # show logs in terminal
    log_file="flowpaths.log",         # write logs to this file
    file_mode="a",                    # "a" append (or "w" overwrite)
)

Notes: - Levels available: fp.utils.logging.DEBUG, INFO, WARNING, ERROR, CRITICAL. - Default level is DEBUG. If you prefer quieter output, use INFO or WARNING. - Internally, the package logs through its own logger; configure_logging sets handlers/formatters accordingly.

API reference:

Configures logging for the flowpaths package.

Parameters:

• level: int, optional

  Logging level (e.g., fp.utils.logging.DEBUG, fp.utils.logging.INFO). Default is fp.utils.logging.DEBUG.

• log_to_console: bool, optional

  Whether to log to the console. Default is True.

• log_file: str, optional

  File path to log to. If None, logging to a file is disabled. Default is None. If a file path is provided, the log will be written to that file. If the file already exists, it will be overwritten unless file_mode is set to “a”.

• file_mode: str, optional

  Mode for the log file. “a” (append) or “w” (overwrite). Default is “w”.

Source code in flowpaths/utils/logging.py

def configure_logging(
        level=logging.DEBUG, 
        log_to_console=True, 
        log_file=None, 
        file_mode="w"  # "a" for append, "w" for overwrite
    ):
    """
    Configures logging for the flowpaths package.

    Parameters:
    -----------

    - `level: int`, optional

        Logging level (e.g., fp.utils.logging.DEBUG, fp.utils.logging.INFO). 
        Default is fp.utils.logging.DEBUG.

    - `log_to_console: bool`, optional

        Whether to log to the console. Default is True.

    - `log_file: str`, optional

        File path to log to. If None, logging to a file is disabled. Default is None.
        If a file path is provided, the log will be written to that file.
        If the file already exists, it will be overwritten unless `file_mode` is set to "a".

    - `file_mode: str`, optional

        Mode for the log file. "a" (append) or "w" (overwrite). Default is "w".

    """
    # Remove existing handlers to avoid duplicate logs
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)

    # Set the logger level
    logger.setLevel(level)

    # Define a formatter
    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")

    # Add console handler if enabled
    if log_to_console:
        console_handler = logging.StreamHandler()
        console_handler.setLevel(level)
        console_handler.setFormatter(formatter)
        logger.addHandler(console_handler)

    if file_mode not in ["a", "w"]:
        raise ValueError("file_mode must be either 'a' (append) or 'w' (overwrite)")

    # Add file handler if a file path is provided
    if log_file:
        file_handler = logging.FileHandler(log_file, mode=file_mode)  # Use file_mode
        file_handler.setLevel(level)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    logger.info("Logging initialized: level=%s, console=%s, file=%s, mode=%s", 
                level, log_to_console, log_file, file_mode)

---

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](https://networkx.org/documentation/stable/reference/classes/digraph.html).

    - `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

draw(
    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_graph_title": False,
        "show_path_weights": False,
        "show_path_weight_on_first_edge": True,
        "pathwidth": 3.0,
        "style": "default",
        "color_nodes": False,
        "sankey_arrowlen": 0,
        "sankey_color_toggle": False,
        "sankey_arrow_toggle": False,
    },
)

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_graph_title: bool

    Whether to show the graph title (from graph id) in the figure. 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.

  • style: str

    The style of the drawing. Available options: default, points, sankey.

    • default: Standard graphviz rendering with nodes as rounded rectangles
    • points: Graphviz rendering with nodes as points
    • sankey: Interactive Sankey diagram using plotly (requires acyclic graph). Saves as HTML by default (interactive) or static image formats (png, pdf, svg) if kaleido is installed. Automatically displays in Jupyter notebooks.

  • color_nodes: bool

    Whether to use the existing node coloring behavior.
    If `False` (default), all nodes use a neutral color.
    If `True`, nodes are colored as before (including `additional_starts`
    in green and `additional_ends` in red for graphviz styles).
    

  • sankey_arrowlen: float

    Length of arrowheads for Sankey links (Plotly arrowlen). Default is 0 (no arrowheads).

  • sankey_color_toggle: bool

    Whether to add an interactive toggle (buttons) to switch Sankey links between colored and monochrome gray. Default is False.

  • sankey_arrow_toggle: bool

    Whether to add an interactive toggle (buttons) to switch Sankey link arrowheads on/off. Default is False.

Source code in flowpaths/utils/graphutils.py

def draw(
        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_graph_title": False,
            "show_path_weights": False,
            "show_path_weight_on_first_edge": True,
            "pathwidth": 3.0,
            "style": "default",
            "color_nodes": False,
            "sankey_arrowlen": 0,
            "sankey_color_toggle": False,
            "sankey_arrow_toggle": False,
        },
        ):
        """
        Draw the graph with the paths and their weights highlighted.

        Parameters
        ----------

        - `G`: nx.DiGraph 

            The input directed acyclic graph, as [networkx DiGraph](https://networkx.org/documentation/stable/reference/classes/digraph.html). 

        - `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_graph_title`: bool

                Whether to show the graph title (from graph id) in the figure.
                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`.

            - `style`: str

                The style of the drawing. Available options: `default`, `points`, `sankey`.

                - `default`: Standard graphviz rendering with nodes as rounded rectangles
                - `points`: Graphviz rendering with nodes as points
                - `sankey`: Interactive Sankey diagram using plotly (requires acyclic graph). 
                  Saves as HTML by default (interactive) or static image formats (png, pdf, svg) if kaleido is installed.
                  Automatically displays in Jupyter notebooks.

            - `color_nodes`: bool

                    Whether to use the existing node coloring behavior.
                    If `False` (default), all nodes use a neutral color.
                    If `True`, nodes are colored as before (including `additional_starts`
                    in green and `additional_ends` in red for graphviz styles).

            - `sankey_arrowlen`: float

                Length of arrowheads for Sankey links (Plotly `arrowlen`).
                Default is `0` (no arrowheads).

            - `sankey_color_toggle`: bool

                Whether to add an interactive toggle (buttons) to switch Sankey
                links between colored and monochrome gray.
                Default is `False`.

            - `sankey_arrow_toggle`: bool

                Whether to add an interactive toggle (buttons) to switch Sankey
                link arrowheads on/off.
                Default is `False`.

        """

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

        style = draw_options.get("style", "default")

        # Handle Sankey diagram separately
        if style == "sankey":
            # Check if graph is acyclic
            if not nx.is_directed_acyclic_graph(G):
                utils.logger.error(f"{__name__}: Sankey diagram requires an acyclic graph.")
                raise ValueError("Sankey diagram requires an acyclic graph.")

            try:
                sankey_arrowlen = float(draw_options.get("sankey_arrowlen", 0))
            except (TypeError, ValueError):
                utils.logger.error(f"{__name__}: draw_options['sankey_arrowlen'] must be numeric.")
                raise ValueError("draw_options['sankey_arrowlen'] must be numeric.")

            if sankey_arrowlen < 0:
                utils.logger.error(f"{__name__}: draw_options['sankey_arrowlen'] must be >= 0.")
                raise ValueError("draw_options['sankey_arrowlen'] must be >= 0.")

            sankey_color_toggle = bool(draw_options.get("sankey_color_toggle", False))
            sankey_arrow_toggle = bool(draw_options.get("sankey_arrow_toggle", False))
            color_nodes = bool(draw_options.get("color_nodes", False))
            show_graph_title = bool(draw_options.get("show_graph_title", False))
            default_arrowlen_for_toggle = sankey_arrowlen if sankey_arrowlen > 0 else 15.0

            try:
                import plotly.graph_objects as go
            except ImportError:
                utils.logger.error(f"{__name__}: plotly module not found. It should be installed with flowpaths. Try reinstalling: pip install --force-reinstall flowpaths")
                raise ImportError("plotly module not found. It should be installed with flowpaths. Try reinstalling: pip install --force-reinstall flowpaths")

            # Create node list in topological order, with sources and sinks at the end
            # This ordering can help preserve link ordering in the Sankey layout
            topo_order = list(nx.topological_sort(G))
            longest_path_len = nx.algorithms.dag.dag_longest_path_length(G)
            sankey_width = max(900, 500 + 50 * max(1, longest_path_len))

            # Identify sources (in-degree 0) and sinks (out-degree 0)
            sources = [node for node in topo_order if G.in_degree(node) == 0]
            sinks = [node for node in topo_order if G.out_degree(node) == 0]

            # Middle nodes (neither pure source nor pure sink)
            middle_nodes = [node for node in topo_order if node not in sources and node not in sinks]

            # Build node list: middle nodes in topo order, then sources, then sinks
            node_list = middle_nodes + sources + sinks
            node_dict = {node: idx for idx, node in enumerate(node_list)}

            # Define colors for paths (with transparency for blending)
            colors = [
                "rgba(255, 0, 0, 0.4)",      # red
                "rgba(0, 0, 255, 0.4)",      # blue
                "rgba(0, 128, 0, 0.4)",      # green
                "rgba(128, 0, 128, 0.4)",    # purple
                "rgba(165, 42, 42, 0.4)",    # brown
                "rgba(0, 255, 255, 0.4)",    # cyan
                "rgba(255, 255, 0, 0.4)",    # yellow
                "rgba(255, 192, 203, 0.4)",  # pink
                "rgba(128, 128, 128, 0.4)",  # grey
                "rgba(210, 105, 30, 0.4)",   # chocolate
                "rgba(0, 0, 139, 0.4)",      # darkblue
                "rgba(85, 107, 47, 0.4)",    # darkolivegreen
                "rgba(47, 79, 79, 0.4)",     # darkslategray
                "rgba(0, 191, 255, 0.4)",    # deepskyblue
                "rgba(95, 158, 160, 0.4)",   # cadetblue
                "rgba(139, 0, 139, 0.4)",    # darkmagenta
                "rgba(255, 193, 37, 0.4)",   # goldenrod 
            ]

            # Build links with path information to maintain consistent ordering at nodes
            # Structure: list of (source, target, weight, color, path_idx)
            links_with_metadata = []

            for path_idx, path in enumerate(paths):
                path_weight = weights[path_idx] if path_idx < len(weights) else 1
                path_color = colors[path_idx % len(colors)]

                # Add each edge in the path
                for i in range(len(path) - 1):
                    source = node_dict[path[i]]
                    target = node_dict[path[i + 1]]
                    links_with_metadata.append((source, target, path_weight, path_color, path_idx))

            # Sort links by path index to maintain consistent ordering throughout the diagram
            # This ensures edges from the same path appear in the same relative order at all nodes
            links_with_metadata.sort(key=lambda x: x[4])

            # Extract sorted components
            link_sources = [link[0] for link in links_with_metadata]
            link_targets = [link[1] for link in links_with_metadata]
            link_values = [link[2] for link in links_with_metadata]
            link_colors = [link[3] for link in links_with_metadata]

            # Create Sankey diagram
            link_dict = dict(
                source=link_sources,
                target=link_targets,
                value=link_values,
                color=link_colors,
            )
            if sankey_arrowlen > 0:
                link_dict["arrowlen"] = sankey_arrowlen

            node_color = "rgba(99, 110, 120, 0.85)" if not color_nodes else "rgba(31, 119, 180, 0.8)"

            base_fig = go.Figure(data=[go.Sankey(
                node=dict(
                    pad=15,
                    thickness=20,
                    line=dict(color="black", width=0.5),
                    label=[str(node) for node in node_list],
                    color=[node_color] * len(node_list),
                ),
                link=link_dict
            )])

            # Use graph ID as title if available
            graph_id = G.graph.get("id", fpid(G))
            title_text = f"{graph_id}" if show_graph_title and graph_id and graph_id != str(id(G)) else ""

            base_fig.update_layout(
                title_text=title_text,
                font_size=10,
                width=sankey_width,
                height=600,
            )

            fig = go.Figure(base_fig)

            updatemenus = []

            if sankey_color_toggle and len(link_colors) > 0:
                monochrome_colors = ["rgba(150, 150, 150, 0.6)"] * len(link_colors)
                updatemenus.append(
                    dict(
                        type="buttons",
                        direction="left",
                        x=0.0,
                        y=1.12,
                        showactive=True,
                        buttons=[
                            dict(
                                label="Colored links",
                                method="restyle",
                                args=[{"link.color": [link_colors]}],
                            ),
                            dict(
                                label="Monochrome links",
                                method="restyle",
                                args=[{"link.color": [monochrome_colors]}],
                            ),
                        ],
                    )
                )

            if sankey_arrow_toggle:
                updatemenus.append(
                    dict(
                        type="buttons",
                        direction="left",
                        x=0.0,
                        y=1.05,
                        showactive=True,
                        buttons=[
                            dict(
                                label="Arrowheads on",
                                method="restyle",
                                args=[{"link.arrowlen": [default_arrowlen_for_toggle]}],
                            ),
                            dict(
                                label="Arrowheads off",
                                method="restyle",
                                args=[{"link.arrowlen": [0]}],
                            ),
                        ],
                    )
                )

            if len(updatemenus) > 0:
                fig.update_layout(updatemenus=updatemenus)

            # Determine base filename and extension
            file_ext = filename.split('.')[-1].lower() if '.' in filename else ''
            base_filename = filename.rsplit('.', 1)[0] if '.' in filename else filename

            # Always save HTML (interactive version)
            html_filename = base_filename + '.html'
            fig.write_html(html_filename)
            utils.logger.info(f"{__name__}: Sankey diagram (HTML) saved as {html_filename}")

            # Also save static image (PDF by default, or specified format)
            static_format = file_ext if file_ext in ['png', 'pdf', 'svg', 'jpg', 'jpeg'] else 'pdf'
            static_filename = base_filename + '.' + static_format

            try:
                static_fig = go.Figure(base_fig)
                if static_format == 'pdf':
                    static_fig.update_layout(
                        width=sankey_width,
                        height=900,
                    )

                static_fig.write_image(static_filename, format=static_format)
                utils.logger.info(f"{__name__}: Sankey diagram (static) saved as {static_filename}")
            except Exception as e:
                utils.logger.warning(f"{__name__}: Could not save static image. Error: {e}")
                utils.logger.warning(f"{__name__}: Static image export may require additional system dependencies.")

            # Check if we're in a Jupyter notebook and show the figure
            if "get_ipython" in globals():
                try:
                    if globals()["get_ipython"]() is not None:
                        fig.show()
                except Exception:
                    pass  # Not in a notebook, just save

            return

        try:
            import graphviz as gv

            color_nodes = bool(draw_options.get("color_nodes", False))

            dot = gv.Digraph(format="pdf")
            dot.graph_attr["rankdir"] = "LR"  # Display the graph in landscape mode

            # style already extracted above
            if style == "default":
                dot.node_attr["shape"] = "rectangle"  # Rectangle nodes
                dot.node_attr["style"] = "rounded"  # Rounded rectangle nodes
            elif style == "points":
                dot.node_attr["shape"] = "point"  # Point nodes
                dot.node_attr["style"] = "filled"  # Filled point nodes
                # dot.node_attr['label'] = '' 
                dot.node_attr['width'] = '0.1' 

            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():
                    neutral_node_color = "gray40"
                    color = neutral_node_color
                    penwidth = "1.0"
                    if color_nodes:
                        color = "black"
                        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]:
                        label = f"{G.nodes[node][flow_attr]}\\n{node}" if style != "points" else ""
                        dot.node(
                            name=str(node),
                            label=label,
                            shape="record",
                            color=color, 
                            penwidth=penwidth)
                    else:
                        label = str(node) if style != "points" 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]) if len(weights) > 0 else "",
                            fontname="Arial",
                        )
                    else:
                        dot.edge(
                            str(path[i]),
                            str(path[i + 1]),
                            color=pathColor,
                            penwidth=str(draw_options.get("pathwidth", 3.0)),
                            )
                if len(path) == 1:
                    dot.node(str(path[0]), 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"
                        )

            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

read_graph

read_graph(
    graph_raw,
) -> nx.DiGraph

Parse a single graph block from a list of lines.

Accepts one or more header lines at the beginning (each prefixed by ‘#’), followed by a line containing the number of vertices (n), then any number of edge lines of the form: “u v w” (whitespace-separated).

Subpath constraint lines

Lines starting with “#S” define a (directed) subpath constraint as a sequence of nodes: “#S n1 n2 n3 …”. For each such line we build the list of consecutive edge tuples [(n1,n2), (n2,n3), …] and append this edge-list (the subpath) to G.graph[“constraints”]. Duplicate filtering is applied on the whole node sequence: if an identical sequence of nodes has already appeared in a previous “#S” line, the entire subpath line is ignored (its edges are not added again). Different subpaths may

share edges; they are kept as separate entries. After all graph edges are parsed, every constraint edge is validated to ensure it exists in the graph; a missing edge raises ValueError.

Example block

graph number = 1 name = foo

any other header line

S a b c d (adds subpath [(a,b),(b,c),(c,d)])

S b c e (adds subpath [(b,c),(c,e)])

S a b c d (ignored: exact node sequence already seen)

5 a b 1.0 b c 2.5 c d 3.0 c e 4.0

Source code in flowpaths/utils/graphutils.py

def read_graph(graph_raw) -> nx.DiGraph:
    """
    Parse a single graph block from a list of lines.

    Accepts one or more header lines at the beginning (each prefixed by '#'),
    followed by a line containing the number of vertices (n), then any number
    of edge lines of the form: "u v w" (whitespace-separated).

    Subpath constraint lines:
        Lines starting with "#S" define a (directed) subpath constraint as a
        sequence of nodes: "#S n1 n2 n3 ...". For each such line we build the
        list of consecutive edge tuples [(n1,n2), (n2,n3), ...] and append this
        edge-list (the subpath) to G.graph["constraints"]. Duplicate filtering
        is applied on the whole node sequence: if an identical sequence of
        nodes has already appeared in a previous "#S" line, the entire subpath
        line is ignored (its edges are not added again). Different subpaths may
    share edges; they are kept as separate entries. After all graph edges
    are parsed, every constraint edge is validated to ensure it exists in
    the graph; a missing edge raises ValueError.

    Example block:
        # graph number = 1 name = foo
        # any other header line
        #S a b c d          (adds subpath [(a,b),(b,c),(c,d)])
        #S b c e            (adds subpath [(b,c),(c,e)])
        #S a b c d          (ignored: exact node sequence already seen)
        5
        a b 1.0
        b c 2.5
        c d 3.0
        c e 4.0
    """

    # Collect leading header lines (prefixed by '#') and parse constraint lines prefixed by '#S'
    idx = 0
    header_lines = []
    constraint_subpaths = []       # list of subpaths, each a list of (u,v) edge tuples
    subpaths_seen = set()          # set of full node sequences (tuples) to filter duplicate subpaths
    while idx < len(graph_raw) and graph_raw[idx].lstrip().startswith("#"):
        stripped = graph_raw[idx].lstrip()
        # Subpath constraint line: starts with '#S'
        if stripped.startswith("#S"):
            # Remove leading '#S' and split remaining node sequence
            nodes_part = stripped[2:].strip()  # drop '#S'
            if nodes_part:
                nodes_seq = nodes_part.split()
                seq_key = tuple(nodes_seq)
                # Skip if this exact subpath sequence already processed
                if seq_key not in subpaths_seen:
                    subpaths_seen.add(seq_key)
                    edges_list = [(u, v) for u, v in zip(nodes_seq, nodes_seq[1:])]
                    # Only append if there is at least one edge (>=2 nodes)
                    if edges_list:
                        constraint_subpaths.append(edges_list)
        else:
            # Regular header line (remove leading '#') for metadata / id extraction
            header_lines.append(stripped.lstrip("#").strip())
        idx += 1

    # Determine graph id from the first (non-#S) header line if present
    graph_id = header_lines[0] if header_lines else str(id(graph_raw))

    # Skip blank lines before the vertex-count line
    while idx < len(graph_raw) and graph_raw[idx].strip() == "":
        idx += 1

    if idx >= len(graph_raw):
        error_msg = "Graph block missing vertex-count line."
        utils.logger.error(f"{__name__}: {error_msg}")
        raise ValueError(error_msg)
    # Parse number of vertices (kept for information; not used to count edges here)
    try:
        n = int(graph_raw[idx].strip())
    except ValueError:
        utils.logger.error(f"{__name__}: Invalid vertex-count line: {graph_raw[idx].rstrip()}.")
        raise

    idx += 1

    G = nx.DiGraph()
    G.graph["id"] = graph_id
    # Store (possibly empty) list of subpaths (each a list of edge tuples)
    G.graph["constraints"] = constraint_subpaths

    if n == 0:
        utils.logger.info(f"Graph {graph_id} has 0 vertices.")
        return G

    # Parse edges: skip blanks and comment/header lines defensively
    for line in graph_raw[idx:]:
        if not line.strip() or line.lstrip().startswith('#'):
            continue
        elements = line.split()
        if len(elements) != 3:
            utils.logger.error(f"{__name__}: Invalid edge format: {line.rstrip()}")
            raise ValueError(f"Invalid edge format: {line.rstrip()}")
        u, v, w_str = elements
        try:
            w = float(w_str)
        except ValueError:
            utils.logger.error(f"{__name__}: Invalid weight value in edge: {line.rstrip()}")
            raise
        G.add_edge(u.strip(), v.strip(), flow=w)

    # Validate that every constraint edge exists in the graph
    for subpath in constraint_subpaths:
        for (u, v) in subpath:
            if not G.has_edge(u, v):
                utils.logger.error(f"{__name__}: Constraint edge ({u}, {v}) not found in graph {graph_id} edges.")
                raise ValueError(f"Constraint edge ({u}, {v}) not found in graph edges.")

    G.graph["n"] = G.number_of_nodes()
    G.graph["m"] = G.number_of_edges()
    # Lazy import here to avoid circular import at module load time
    from flowpaths import stdigraph as _stdigraph  # type: ignore
    G.graph["w"] = _stdigraph.stDiGraph(G).get_width()

    return G

read_graphs

read_graphs(filename)

Read one or more graphs from a file.

Supports graphs whose header consists of one or multiple consecutive lines prefixed by ‘#’. Each graph block is: - one or more header lines starting with ‘#’ - one line with the number of vertices (n) - zero or more edge lines “u v w”

Graphs are delimited by the start of the next header (a line starting with ‘#’) or the end of file.

Source code in flowpaths/utils/graphutils.py

def read_graphs(filename):
    """
    Read one or more graphs from a file.

    Supports graphs whose header consists of one or multiple consecutive lines
    prefixed by '#'. Each graph block is:
        - one or more header lines starting with '#'
        - one line with the number of vertices (n)
        - zero or more edge lines "u v w"

    Graphs are delimited by the start of the next header (a line starting with '#')
    or the end of file.
    """
    with open(filename, "r") as f:
        lines = f.readlines()

    graphs = []
    n_lines = len(lines)
    i = 0

    # Iterate through the file, capturing blocks that start with one or more '#' lines
    while i < n_lines:
        # Move to the start of the next graph header
        while i < n_lines and not lines[i].lstrip().startswith('#'):
            i += 1
        if i >= n_lines:
            break

        start = i

        # Consume all consecutive header lines for this graph
        while i < n_lines and lines[i].lstrip().startswith('#'):
            i += 1

        # Advance until the next header line (start of next graph) or EOF
        j = i
        while j < n_lines and not lines[j].lstrip().startswith('#'):
            j += 1

        graphs.append(read_graph(lines[start:j]))
        i = j

    return graphs