Skip to content

See also

Minimum Path Cover in General Graphs

1. Definition

The Minimum Path Cover problem on a directed graph, possibly with cycles, is defined as follows:

  • INPUT:

    • A directed graph \(G = (V,E)\).
    • Node subsets \(S \subseteq V\) and \(T \subseteq V\), where the walks are allowed to start and allowed to end, respectively.

  • OUTPUT: A minimum number \(k\) of walks \(W_1,\dots,W_k\), starting in some node in \(S\) and ending in some node in \(T\), such that every edge appears in at least one \(W_i\).

Note

  • This class support also covers of nodes. Set the parameter cover_type = "node". For details on how these are handled internally, see Handling graphs with flows / weights on nodes.
  • The graph may have more than one source or sink nodes, in which case the solution paths are just required to start in any source node, and end in any sink node.

2. Solving the problem

We create the graph as a networkx DiGraph. In real project, you will likely have a method that transforms your graph to a DiGraph.

import flowpaths as fp
import networkx as nx

def test():
  graph = nx.DiGraph()
  graph.add_edge("s", "a")
  graph.add_edge("a", "t")
  graph.add_edge("s", "b")
  graph.add_edge("b", "a")
  graph.add_edge("a", "h")
  graph.add_edge("h", "t")
  graph.add_edge("b", "c")
  graph.add_edge("c", "d")
  graph.add_edge("c", "h")
  graph.add_edge("d", "h")
  graph.add_edge("d", "e")
  graph.add_edge("e", "c")
  graph.add_edge("e", "f")
  graph.add_edge("f", "g")
  graph.add_edge("g", "e")

  mpc_model = fp.MinPathCoverCycles(G=graph)
  mpc_model.solve()

The solution of MinPathCoverCycles is a dictionary, with a key 'walks' containing the solution walks:

if mpc_model.is_solved():
    solution = mpc_model.get_solution()
    print(solution)
    # {'walks': [
    #   ['s', 'a', 't'], 
    #   ['s', 'b', 'a', 'h', 't'], 
    #   ['s', 'b', 'c', 'd', 'e', 'f', 'g', 'e', 'c', 'h', 't'], 
    #   ['s', 'b', 'c', 'd', 'h', 't']]}

We can also support subset constraints:

subset_constraints=[[("b", "a"),("a", "t")]]
mpc_model_sc = fp.MinPathCover(
    graph,
    subset_constraints=subset_constraints,
)
mpc_model_sc.solve()

MinPathCoverCycles 

MinPathCoverCycles(
    G: DiGraph,
    cover_type: str = "edge",
    subset_constraints: list = [],
    subset_constraints_coverage: float = 1.0,
    elements_to_ignore: list = [],
    additional_starts: list = [],
    additional_ends: list = [],
    optimization_options: dict = {},
    solver_options: dict = {},
)

Bases: AbstractWalkModelDiGraph

This class finds a minimum number of walks covering the edges of a directed graph, possibly with cycles – and generalizations of this problem, see the parameters below.

Parameters

  • G: nx.DiGraph

    The input directed graph, as networkx DiGraph.

  • cover_type: str, optional

    The elements of the graph to cover. Default is "edge". Options:

    • "edge": cover the edges of the graph. This is the default.
    • "node": cover the nodes of the graph.

  • subset_constraints: list, optional

    List of subset constraints. Default is an empty list. Each subset constraint is a list of edges that must all appear in some solution walks (in any order), according to the subset_constraints_coverage parameter (see below).

  • subset_constraints_coverage: float, optional

    Coverage fraction of the subset constraints that must be covered by some solution walks.

    Defaults to 1.0, meaning that 100% of the edges (or nodes, if flow_attr_origin is "node") of the constraint need to be covered by some solution walk). See subset constraints documentation

  • elements_to_ignore: list, optional

    List of graph elements to ignore by the solution walks (i.e., these don’t have to be covered, unless they are part of a subset constraint). These elements are either edges or nodes, depending on the cover_type parameter. Default is an empty list. See ignoring edges documentation

  • additional_starts: list, optional

    List of additional start nodes of the walks. Default is an empty list. See additional start/end nodes documentation.

  • additional_ends: list, optional

    List of additional end nodes of the walks. Default is an empty list. See additional start/end nodes documentation.

  • optimization_options: dict, optional

    Dictionary with the optimization options. Default is None. See optimization options documentation.

  • solver_options: dict, optional

    Dictionary with the solver options. Default is None. See solver options documentation.

Source code in flowpaths/minpathcovercycles.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __init__(
    self,
    G: nx.DiGraph,
    cover_type: str = "edge",
    subset_constraints: list = [],
    subset_constraints_coverage: float = 1.0,
    elements_to_ignore: list = [],
    additional_starts: list = [],
    additional_ends: list = [],
    optimization_options: dict = {},
    solver_options: dict = {},
):
    """
    This class finds a minimum number of walks covering the edges of a directed graph, possibly with cycles -- and generalizations of this problem, see the parameters below.

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

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

    - `cover_type: str`, optional

        The elements of the graph to cover. Default is `"edge"`. Options:

        - `"edge"`: cover the edges of the graph. This is the default.
        - `"node"`: cover the nodes of the graph.

    - `subset_constraints: list`, optional

        List of subset constraints. Default is an empty list. 
        Each subset constraint is a list of edges that must all appear in some solution walks (in any order), according 
        to the `subset_constraints_coverage` parameter (see below).

    - `subset_constraints_coverage: float`, optional

        Coverage fraction of the subset constraints that must be covered by some solution walks. 

        Defaults to `1.0`, meaning that 100% of the edges (or nodes, if `flow_attr_origin` is `"node"`) of 
        the constraint need to be covered by some solution walk). 
        See [subset constraints documentation](subset-constraints.md#3-relaxing-the-constraint-coverage)

    - `elements_to_ignore: list`, optional

        List of graph elements to ignore by the solution walks (i.e., these don't have to be covered, unless they are part of a subset constraint).
        These elements are either edges or nodes, depending on the `cover_type` parameter.
        Default is an empty list. See [ignoring edges documentation](ignoring-edges.md)

    - `additional_starts: list`, optional

        List of additional start nodes of the walks. Default is an empty list. See [additional start/end nodes documentation](additional-start-end-nodes.md).

    - `additional_ends: list`, optional

        List of additional end nodes of the walks. Default is an empty list. See [additional start/end nodes documentation](additional-start-end-nodes.md).

    - `optimization_options: dict`, optional

        Dictionary with the optimization options. Default is `None`. See [optimization options documentation](solver-options-optimizations.md).

    - `solver_options: dict`, optional

        Dictionary with the solver options. Default is `None`. See [solver options documentation](solver-options-optimizations.md).

    """

    # Handling node-weighted graphs
    self.cover_type = cover_type
    if self.cover_type == "node":
        if G.number_of_nodes() == 0:
            utils.logger.error(f"{__name__}: The input graph G has no nodes. Please provide a graph with at least one node.")
            raise ValueError(f"The input graph G has no nodes. Please provide a graph with at least one node.")
        # NodeExpandedDiGraph needs to have flow_attr on edges, otherwise it will add the edges to edges_to_ignore
        G_with_flow_attr = deepcopy(G)
        node_flow_attr = str(id(G_with_flow_attr)) + "_flow_attr"
        for node in G_with_flow_attr.nodes():
            G_with_flow_attr.nodes[node][node_flow_attr] = 0 # any dummy value
        self.G_internal = nedg.NodeExpandedDiGraph(G_with_flow_attr, node_flow_attr=node_flow_attr)
        subset_constraints_internal = self.G_internal.get_expanded_subpath_constraints(subset_constraints)

        edges_to_ignore_internal = self.G_internal.edges_to_ignore
        if not all(isinstance(node, str) for node in elements_to_ignore):
            utils.logger.error(f"elements_to_ignore must be a list of nodes, i.e. strings, not {elements_to_ignore}")
            raise ValueError(f"elements_to_ignore must be a list of nodes, i.e. strings, not {elements_to_ignore}")
        edges_to_ignore_internal += [self.G_internal.get_expanded_edge(node) for node in elements_to_ignore]

        additional_starts_internal = self.G_internal.get_expanded_additional_starts(additional_starts)
        additional_ends_internal = self.G_internal.get_expanded_additional_ends(additional_ends)
    elif self.cover_type == "edge":
        if G.number_of_edges() == 0:
            utils.logger.error(f"{__name__}: The input graph G has no edges. Please provide a graph with at least one edge.")
            raise ValueError(f"The input graph G has no edges. Please provide a graph with at least one edge.")
        self.G_internal = G
        subset_constraints_internal = subset_constraints

        if not all(isinstance(edge, tuple) and len(edge) == 2 for edge in elements_to_ignore):
            utils.logger.error(f"elements_to_ignore must be a list of edges, i.e. tuples of nodes, not {elements_to_ignore}")
            raise ValueError(f"elements_to_ignore must be a list of edges, i.e. tuples of nodes, not {elements_to_ignore}")
        edges_to_ignore_internal = elements_to_ignore

        additional_starts_internal = additional_starts
        additional_ends_internal = additional_ends
    else:
        utils.logger.error(f"cover_type must be either 'node' or 'edge', not {self.cover_type}")
        raise ValueError(f"cover_type must be either 'node' or 'edge', not {self.cover_type}")

    self.G = self.G_internal
    self.subset_constraints = subset_constraints_internal
    self.edges_to_ignore = edges_to_ignore_internal

    self.subset_constraints_coverage = subset_constraints_coverage
    self.additional_starts = additional_starts_internal
    self.additional_ends = additional_ends_internal

    self._solution = None
    self._lowerbound_k = None
    self._is_solved = None
    self.model = None

    self.solve_statistics = {}
    self.optimization_options = optimization_options
    self.solver_options = solver_options
    self.time_limit = self.solver_options.get("time_limit", sw.SolverWrapper.time_limit)
    self.solve_time_start = None

    utils.logger.info(f"{__name__}: initialized with graph id = {utils.fpid(G)}")

get_solution 

get_solution()

Get the solution of the Min Path Cover model, as dict with unique key "walks".

Source code in flowpaths/minpathcovercycles.py 
193
194
195
196
197
198
def get_solution(self):
    """
    Get the solution of the Min Path Cover model, as dict with unique key `"walks"`.
    """
    self.check_is_solved()
    return self._solution

kPathCoverCycles 

kPathCoverCycles(
    G: DiGraph,
    k: int = None,
    cover_type: str = "edge",
    subset_constraints: list = [],
    subset_constraints_coverage: float = 1.0,
    elements_to_ignore: list = [],
    additional_starts: list = [],
    additional_ends: list = [],
    optimization_options: dict = None,
    solver_options: dict = {},
)

Bases: AbstractWalkModelDiGraph

This class finds, if possible, k walks covering the edges of a directed graph, possibly with cycles – and generalizations of this problem, see the parameters below.

Moreover, among all such walk covers, it finds minimizing the sum of the lengths of the walks (in terms of total number of edges).

Parameters

  • G: nx.DiGraph

    The input directed graph, as networkx DiGraph, which can have cycles.

  • k: int

    The number of walks to decompose in.

  • cover_type : str, optional

    The elements of the graph to cover. Default is "edge". Options:

    • "edge": cover the edges of the graph. This is the default.
    • "node": cover the nodes of the graph.

  • subset_constraints: list, optional

    List of subset constraints. Default is an empty list. Each subset constraint is a list of edges that must be covered by some solution walk, in any order, according to the subset_constraints_coverage parameter (see below).

  • subset_constraints_coverage: float, optional

    Coverage fraction of the subset constraints that must be covered by some solution walk.

    Defaults to 1.0, meaning that 100% of the edges (or nodes, if flow_attr_origin is "node") of the constraint need to be covered by some solution walk). See subset constraints documentation

  • elements_to_ignore: list, optional

    List of edges (or nodes, if flow_attr_origin is "node") to ignore when adding constrains on flow explanation by the weighted paths. Default is an empty list. See ignoring edges documentation

  • additional_starts: list, optional

    List of additional start nodes of the walks. Default is an empty list.

  • additional_ends: list, optional

    List of additional end nodes of the walks. Default is an empty list.

  • optimization_options: dict, optional

    Dictionary with the optimization options. Default is None. See optimization options documentation.

  • solver_options: dict, optional

    Dictionary with the solver options. Default is {}. See solver options documentation.

Source code in flowpaths/kpathcovercycles.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def __init__(
    self,
    G: nx.DiGraph,
    k: int = None,
    cover_type: str = "edge",
    subset_constraints: list = [],
    subset_constraints_coverage: float = 1.0,
    elements_to_ignore: list = [],
    additional_starts: list = [],
    additional_ends: list = [],
    optimization_options: dict = None,
    solver_options: dict = {},
):
    """
    This class finds, if possible, `k` walks covering the edges of a directed graph, possibly with cycles -- and generalizations of this problem, see the parameters below.

    Moreover, among all such walk covers, it finds minimizing the sum of the lengths of the walks (in terms of total number of edges).

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

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

    - `k: int`

        The number of walks to decompose in.

    - `cover_type : str`, optional

        The elements of the graph to cover. Default is `"edge"`. Options:

        - `"edge"`: cover the edges of the graph. This is the default.
        - `"node"`: cover the nodes of the graph.

     - `subset_constraints: list`, optional

        List of subset constraints. Default is an empty list. 
        Each subset constraint is a list of edges that must be covered by some solution walk, in any order, according 
        to the `subset_constraints_coverage` parameter (see below).

    - `subset_constraints_coverage: float`, optional

        Coverage fraction of the subset constraints that must be covered by some solution walk.

        Defaults to `1.0`, meaning that 100% of the edges (or nodes, if `flow_attr_origin` is `"node"`) of
        the constraint need to be covered by some solution walk).
        See [subset constraints documentation](subset-constraints.md#3-relaxing-the-constraint-coverage)

    - `elements_to_ignore: list`, optional

        List of edges (or nodes, if `flow_attr_origin` is `"node"`) to ignore when adding constrains on flow explanation by the weighted paths. 
        Default is an empty list. See [ignoring edges documentation](ignoring-edges.md)

    - `additional_starts: list`, optional

        List of additional start nodes of the walks. Default is an empty list.

    - `additional_ends: list`, optional

        List of additional end nodes of the walks. Default is an empty list.

    - `optimization_options: dict`, optional

        Dictionary with the optimization options. Default is `None`. See [optimization options documentation](solver-options-optimizations.md).

    - `solver_options: dict`, optional

        Dictionary with the solver options. Default is `{}`. See [solver options documentation](solver-options-optimizations.md).

    """

    # Handling node-weighted graphs
    self.cover_type = cover_type
    if self.cover_type == "node":
        if G.number_of_nodes() == 0:
            utils.logger.error(f"{__name__}: The input graph G has no nodes. Please provide a graph with at least one node.")
            raise ValueError(f"The input graph G has no nodes. Please provide a graph with at least one node.")
        # NodeExpandedDiGraph needs to have flow_attr on edges, otherwise it will add the edges to edges_to_ignore
        G_with_flow_attr = deepcopy(G)
        node_flow_attr = str(id(G_with_flow_attr)) + "_flow_attr"
        for node in G_with_flow_attr.nodes():
            G_with_flow_attr.nodes[node][node_flow_attr] = 0 # any dummy value
        self.G_internal = nedg.NodeExpandedDiGraph(G_with_flow_attr, node_flow_attr=node_flow_attr)
        subset_constraints_internal = self.G_internal.get_expanded_subpath_constraints(subset_constraints)
        additional_starts_internal = self.G_internal.get_expanded_additional_starts(additional_starts)
        additional_ends_internal = self.G_internal.get_expanded_additional_ends(additional_ends)

        if not all(isinstance(element_to_ignore, str) for element_to_ignore in elements_to_ignore):
            utils.logger.error(f"elements_to_ignore must be a list of nodes (i.e strings), not {elements_to_ignore}")
            raise ValueError(f"elements_to_ignore must be a list of nodes (i.e strings), not {elements_to_ignore}")
        edges_to_ignore_internal = self.G_internal.edges_to_ignore
        edges_to_ignore_internal += [self.G_internal.get_expanded_edge(node) for node in elements_to_ignore]
        edges_to_ignore_internal = list(set(edges_to_ignore_internal))

    elif self.cover_type == "edge":
        if G.number_of_edges() == 0:
            utils.logger.error(f"{__name__}: The input graph G has no edges. Please provide a graph with at least one edge.")
            raise ValueError(f"The input graph G has no edges. Please provide a graph with at least one edge.")
        self.G_internal = G
        subset_constraints_internal = subset_constraints
        if not all(isinstance(edge, tuple) and len(edge) == 2 for edge in elements_to_ignore):
            utils.logger.error(f"elements_to_ignore must be a list of edges (i.e. tuples of nodes), not {elements_to_ignore}")
            raise ValueError(f"elements_to_ignore must be a list of edges (i.e. tuples of nodes), not {elements_to_ignore}")
        edges_to_ignore_internal = elements_to_ignore
        additional_starts_internal = additional_starts
        additional_ends_internal = additional_ends
    else:
        utils.logger.error(f"flow_attr_origin must be either 'node' or 'edge', not {self.cover_type}")
        raise ValueError(f"flow_attr_origin must be either 'node' or 'edge', not {self.cover_type}")

    self.G = stdigraph.stDiGraph(self.G_internal, additional_starts=additional_starts_internal, additional_ends=additional_ends_internal)
    self.subset_constraints = subset_constraints_internal
    self.edges_to_ignore = self.G.source_sink_edges.union(edges_to_ignore_internal)

    self.k = k
    self.subset_constraints_coverage = subset_constraints_coverage

    self._solution = None
    self._lowerbound_k = None

    self.solve_statistics = {}
    self.optimization_options = optimization_options.copy() if optimization_options else {}
    self.optimization_options["trusted_edges_for_safety"] = set(e for e in self.G.edges() if e not in self.edges_to_ignore)

    # Call the constructor of the parent class AbstractPathModelDAG
    super().__init__(
        G=self.G,
        k=self.k,
        max_edge_repetition=self.G.number_of_edges() * self.G.number_of_nodes(),
        subset_constraints=self.subset_constraints,
        subset_constraints_coverage=self.subset_constraints_coverage,
        optimization_options=self.optimization_options,
        solver_options=solver_options,
        solve_statistics=self.solve_statistics
    )

    # This method is called from the super class AbstractPathModelDiGraph
    self.create_solver_and_walks()

    # This method is called from the current class to encode the path cover
    self._encode_walk_cover()

    # This method is called from the current class to encode the objective function
    self._encode_objective()

    utils.logger.info(f"{__name__}: initialized with graph id = {utils.fpid(G)}, k = {self.k}")

get_solution 

get_solution()

Retrieves the solution for problem.

If the solution has already been computed and cached as self._solution, it returns the cached solution. Otherwise, it checks if the problem has been solved, computes the solution walks and caches the solution.

Warning

Make sure you called .solve() before calling this method.

Returns

  • solution: dict

    A dictionary containing the solution walks (key "walks").

Raises
  • exception If model is not solved.
Source code in flowpaths/kpathcovercycles.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def get_solution(self):
    """
    Retrieves the solution for problem.

    If the solution has already been computed and cached as `self._solution`, it returns the cached solution.
    Otherwise, it checks if the problem has been solved, computes the solution walks
    and caches the solution.

    !!! warning "Warning"
        Make sure you called `.solve()` before calling this method.

    Returns
    -------
    - `solution: dict`

        A dictionary containing the solution walks (key `"walks"`).

    Raises
    -------
    - `exception` If model is not solved.
    """

    if self._solution is None:
        self.check_is_solved()

        if self.cover_type == "edge":
            self._solution = {
                "walks": self.get_solution_walks(),
                }
        elif self.cover_type == "node":
            self._solution = {
                "_walks_internal": self.get_solution_walks(),
                "walks": self.G_internal.get_condensed_paths(self.get_solution_walks()),
                }

    return self._solution

is_valid_solution 

is_valid_solution()

Checks if the solution is valid, meaning it covers all required edges.

Raises
  • ValueError: If the solution is not available (i.e., self.solution is None).
Returns
  • bool: True if the solution is valid, False otherwise.
Notes
  • get_solution() must be called before this method.
Source code in flowpaths/kpathcovercycles.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def is_valid_solution(self):
    """
    Checks if the solution is valid, meaning it covers all required edges.

    Raises
    ------
    - ValueError: If the solution is not available (i.e., self.solution is None).

    Returns
    -------
    - bool: True if the solution is valid, False otherwise.

    Notes
    -------
    - get_solution() must be called before this method.
    """

    if self._solution is None:
        utils.logger.error(f"{__name__}: Solution is not available. Call get_solution() first.")
        raise ValueError("Solution is not available. Call get_solution() first.")

    solution_walks = self._solution.get("_walks_internal", self._solution["walks"])
    solution_walks_of_edges = [
        [(walk[i], walk[i + 1]) for i in range(len(walk) - 1)]
        for walk in solution_walks
    ]

    covered_by_walks = {(u, v): 0 for (u, v) in self.G.edges()}
    for walk in solution_walks_of_edges:
        for e in walk:
            if e in covered_by_walks:
                covered_by_walks[e] += 1

    for u, v in self.G.edges():
        if (u,v) not in self.edges_to_ignore:
            if covered_by_walks[(u, v)] == 0: 
                return False

    return True