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.

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_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(
        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](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_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) and len(weights) > 0:
            raise ValueError(f"{__name__}: 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]) 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.")

    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