Skip to content

Abstract Path Model

AbstractPathModelDAG 

AbstractPathModelDAG(G: stDiGraph, num_paths: int, subpath_constraints: list = None, subpath_constraints_coverage: float = 1, subpath_constraints_coverage_length: float = None, encode_edge_position: bool = False, encode_path_length: bool = False, edge_length_attr: str = None, optimization_options: dict = None, solver_options: dict = None, solve_statistics: dict = {})

Bases: ABC

This is an abstract class modelling a path-finding (M)ILP in a DAG. The desing of this package is based on the following principles:

  • The class is designed to be extended by other classes that model specific path-finding problems in DAGs. In this way, they all benefit from the variables it encodes, and the safety optimizations it provides.
  • The class uses our custom SolverWrapper class, which is a wrapper around the solvers HiGHS (open source) and Gurobi (free with academic license). In this way, both solvers can be used interchangeably.

More in detail, this class encodes num_paths s-t paths in the DAG G, where s is the global source of the stDiGraph and t is the global sink. It also allows for subpath constraints that must appear in at least one of the s-t paths.

The class creates the following variables:

  • edge_vars: edge_vars[(u, v, i)] = 1 if path i goes through edge (u, v), 0 otherwise

  • edge_position_vars: edge_position_vars[(u, v, i)] = position of edge (u, v) in path i, starting from position 0

    • These variables are created only if encode_edge_position is set to True
    • Note that positions are relative to the globals source s of the stDiGraph, thus the first edge in a path is the edge from s to the first vertex in the original graph, and this first edge has position 0
    • If you set edge_length_attr, then the positions are relative to the edge lengths, and not the number of edges The first edge still gets position 0, and other edges get positions equal to the sum of the lengths of the edges before them in the path
    • If you set edge_length_attr, and an edge has missing edge length, then it gets length 1

  • path_length_vars: path_length_vars[(i)] = length of path i

    • These variables are created only if encode_path_length is set to True
    • Note that the length of a path is the sum of the lengths of the edges in the path
    • If you set edge_length_attr, then the lengths are the sum of the lengths of the edges in the path
    • If you set edge_length_attr, and an edge has missing edge length, then it gets length 1
    • NOTE: the length also includes the edges from gobal source to the first vertex, and from the last vertex to the global sink. By default, these do not have a length attached, so each gets length 1.

  • solver: a solver object to solve the (M)ILP

Safety optimizations

This class uses the “safety information” (see https://doi.org/10.48550/arXiv.2411.03871) in the graph to fix some edge_vars to 1 or 0. The safety information consists of safe paths, or safe sequences, that are guaranteed to appear in at least one cover (made up of any number of s-t paths) of the edges in trusted_edges_for_safety. That is, when implementing a new path-finding (M)ILP, you can guarantee that

  1. The solution is made up of s-t paths
  2. Any solution covers all edges in trusted_edges_for_safety, then safety optimizations can be used to fix some edge_vars to 1, which can speed up the solving process, while guaranteeing global optimality.

Initializes the class with the given parameters.

Args

  • G: stDiGraph.stDiGraph

    The directed acyclic graph (DAG) to be used.

  • num_paths: int

    The number of paths to be computed. - subpath_constraints: list, optional

    A list of lists, where each list is a sequence of edges (not necessarily contiguous, i.e. path). Defaults to None.

    Each sequence of edges must appear in at least one solution path; if you also pass subpath_constraints_coverage, then each sequence of edges must appear in at least subpath_constraints_coverage fraction of some solution path, see below.

  • subpath_constraints_coverage: float, optional

    Coverage fraction of the subpath constraints that must be covered by some solution paths, in terms of number of edges. - Defaults to 1 (meaning that 100% of the edges of the constraint need to be covered by some solution path).

  • subpath_constraints_coverage_length: float, optional

    Coverage fraction of the subpath constraints that must be covered by some solution paths, in terms of length of the subpath. Defaults to None, meaning that this is not imposed. - If you set this constraint, you cannot set subpath_constraints_coverage (and its default value of 1 will be ignored). - If you set this constraint, you also need to set edge_length_attr. If an edge has missing edge length, it gets length 1.

  • encode_edge_position: bool, optional

    Whether to encode the position of the edges in the paths. Defaults to False.

  • encode_path_length: bool, optional

    Whether to encode the length of the paths (in terms of number of edges, or sum of lengths of edges, if set via edge_length_attr). Defaults to False.

  • edge_length_attr: str, optional

    The attribute name from where to get the edge lengths. Defaults to None.

    • If set, then the edge positions, or path lengths (above) are in terms of the edge lengths specified in the edge_length_attr field of each edge
    • If set, and an edge has a missing edge length, then it gets length 1.

  • optimization_options: dict, optional

    Dictionary of optimization options. Defaults to None, in which case the default values are used. See the available optimizations. If you pass any safety optimizations, you must also pass trusted_edges_for_safety (see below). If a child class has already solved the problem and has the solution paths, it can pass them via external_solution_paths to skip the solver creation and encoding of paths (see below).

    • "trusted_edges_for_safety" (set): Set of trusted edges for safety. Defaults to None. If you use any safety optimization, this cannot be None.

    Global optimality

    In order for the optimizations to still guarantee a global optimium, you must guarantee that:

    1. The solution is made up of source-to-sink paths, and
    2. Every edge in trusted_edges_for_safety appears in some solution path, for all solutions. This naturally holds for several problems, for example Minimum Flow Decompositon or [k-Minimum Path Error] where in fact, under default settings, all edges appear in all solutions.
    • "external_solution_paths" (list): External solution paths, as a list of paths, where every path is a list of nodes. Defaults to None. If you provide this, this class skip the solver creation and encoding of paths, and just return these paths. This is useful when the child class managed to solver the problem in a different way, and needs to let this class know them to have a consistent API.

  • solver_options: dict, optional

    Dictionary of solver options. Defaults to None, in which case the default values are used. See the available solver options.

  • solve_statistics: dict, optional

    Dictionary to store solve statistics. Defaults to {}.

Raises

  • ValueError: If trusted_edges_for_safety is not provided when optimizing with optimize_with_safe_paths or optimize_with_safe_sequences.
  • ValueError: If both optimize_with_safe_paths and optimize_with_safe_sequences are set to True.
Source code in flowpaths/abstractpathmodeldag.py 
 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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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
def __init__(
    self,
    G: stdigraph.stDiGraph,
    num_paths: int,
    subpath_constraints: list = None,
    subpath_constraints_coverage: float = 1,
    subpath_constraints_coverage_length: float = None,
    encode_edge_position: bool = False,
    encode_path_length: bool = False,
    edge_length_attr: str = None,
    optimization_options: dict = None,
    solver_options: dict = None,
    solve_statistics: dict = {},
):
    """
    Initializes the class with the given parameters.

    Args
    ----
    - `G: stDiGraph.stDiGraph`  

        The directed acyclic graph (DAG) to be used.

    - `num_paths: int`

        The number of paths to be computed.
    - `subpath_constraints: list`, optional

        A list of lists, where each list is a sequence of edges (not necessarily contiguous, i.e. path). Defaults to None.

        Each sequence of edges must appear in at least one solution path; if you also pass subpath_constraints_coverage, 
        then each sequence of edges must appear in at least subpath_constraints_coverage fraction of some solution path, see below.

    - `subpath_constraints_coverage: float`, optional

        Coverage fraction of the subpath constraints that must be covered by some solution paths, in terms of number of edges. 
            - Defaults to 1 (meaning that 100% of the edges of the constraint need to be covered by some solution path).

    - `subpath_constraints_coverage_length: float`, optional 

        Coverage fraction of the subpath constraints that must be covered by some solution paths, in terms of length of the subpath. Defaults to `None`, meaning that this is not imposed. 
        - If you set this constraint, you cannot set `subpath_constraints_coverage` (and its default value of 1 will be ignored).
        - If you set this constraint, you also need to set `edge_length_attr`. If an edge has missing edge length, it gets length 1.

    - `encode_edge_position: bool`, optional

        Whether to encode the position of the edges in the paths. Defaults to `False`.

    - `encode_path_length: bool`, optional

        Whether to encode the length of the paths (in terms of number of edges, or sum of lengths of edges, if set via `edge_length_attr`). Defaults to `False`.

    - `edge_length_attr: str`, optional

        The attribute name from where to get the edge lengths. Defaults to `None`.

        - If set, then the edge positions, or path lengths (above) are in terms of the edge lengths specified in the `edge_length_attr` field of each edge
        - If set, and an edge has a missing edge length, then it gets length 1.

    - `optimization_options: dict`, optional 

        Dictionary of optimization options. Defaults to `None`, in which case the default values are used. See the [available optimizations](solver-options-optimizations.md). If you pass any safety optimizations, you must also pass `trusted_edges_for_safety` (see below). If a child class has already solved the problem and has the solution paths, it can pass them via `external_solution_paths` to skip the solver creation and encoding of paths (see below).

        - `"trusted_edges_for_safety"` (set): Set of trusted edges for safety. Defaults to `None`. If you use any safety optimization, this cannot be `None`.

        !!! warning "Global optimality"
            In order for the optimizations to still guarantee a global optimium, you must guarantee that:

            1. The solution is made up of source-to-sink paths, and
            2. Every edge in `trusted_edges_for_safety` appears in some solution path, for all solutions. This naturally holds for several problems, for example [Minimum Flow Decompositon](minimum-flow-decomposition.md) or [k-Minimum Path Error] where in fact, under default settings, **all** edges appear in all solutions.

        - `"external_solution_paths"` (list): External solution paths, as a list of paths, where every path is a list of nodes. Defaults to `None`. If you provide this, this class skip the solver creation and encoding of paths, and just return these paths. This is useful when the child class managed to solver the problem in a different way, and needs to let this class know them to have a consistent API.



    - `solver_options: dict`, optional

        Dictionary of solver options. Defaults to `None`, in which case the default values are used. See the [available solver options](solver-options-optimizations.md).

    - `solve_statistics: dict`, optional

        Dictionary to store solve statistics. Defaults to `{}`.


    Raises
    ----------
    - ValueError: If `trusted_edges_for_safety` is not provided when optimizing with `optimize_with_safe_paths` or `optimize_with_safe_sequences`.
    - ValueError: If both `optimize_with_safe_paths` and `optimize_with_safe_sequences` are set to `True`.
    """

    self.G = G
    self.id = self.G.id
    self.k = num_paths
    self.edge_length_attr = edge_length_attr

    self.subpath_constraints = subpath_constraints
    if self.subpath_constraints is not None:
        self.__check_valid_subpath_constraints()

    self.subpath_constraints_coverage = subpath_constraints_coverage
    self.subpath_constraints_coverage_length = subpath_constraints_coverage_length
    if subpath_constraints is not None:
        if self.subpath_constraints_coverage <= 0 or self.subpath_constraints_coverage > 1:
            raise ValueError("subpath_constraints_coverage must be in the range (0, 1]")
        if self.subpath_constraints_coverage_length is not None:
            if self.subpath_constraints_coverage_length <= 0 or self.subpath_constraints_coverage_length > 1:
                raise ValueError("If set, subpath_constraints_coverage_length must be in the range (0, 1]")
            if self.edge_length_attr is None:
                raise ValueError("If subpath_constraints_coverage_length is set, edge_length_attr must be provided.")
            if self.subpath_constraints_coverage < 1:
                raise ValueError("If subpath_constraints_coverage_length is set, you cannot set also subpath_constraints_coverage.")

    self.solve_statistics = solve_statistics
    self.edge_vars = {}
    self.edge_vars_sol = {}
    self.subpaths_vars = {}
    self.encode_edge_position = encode_edge_position
    self.encode_path_length = encode_path_length
    self.edge_position_vars = {}

    if solver_options is None:
        solver_options = {}
    self.threads = solver_options.get("threads", sw.SolverWrapper.threads)
    self.time_limit = solver_options.get("time_limit", sw.SolverWrapper.time_limit)
    self.presolve = solver_options.get("presolve", sw.SolverWrapper.presolve)
    self.log_to_console = solver_options.get("log_to_console", sw.SolverWrapper.log_to_console)
    self.external_solver = solver_options.get("external_solver", sw.SolverWrapper.external_solver)


    # optimizations
    if optimization_options is None:
        optimization_options = {}
    self.optimize_with_safe_paths = optimization_options.get("optimize_with_safe_paths", AbstractPathModelDAG.optimize_with_safe_paths)
    self.optimize_with_safe_sequences = optimization_options.get("optimize_with_safe_sequences", AbstractPathModelDAG.optimize_with_safe_sequences)
    self.trusted_edges_for_safety = optimization_options.get("trusted_edges_for_safety", None)
    self.optimize_with_safe_zero_edges = optimization_options.get("optimize_with_safe_zero_edges", AbstractPathModelDAG.optimize_with_safe_zero_edges)
    self.external_solution_paths = optimization_options.get("external_solution_paths", None)
    if self.external_solution_paths is None:
        self.__is_solved = None
    else:
        self.__is_solved = True


    self.safe_lists = None
    if self.optimize_with_safe_paths and not self.is_solved() and self.trusted_edges_for_safety is not None:
        start_time = time.time()
        self.safe_lists = safetypathcovers.safe_paths(
            self.G,
            self.trusted_edges_for_safety,
            no_duplicates=False,
            threads=self.threads,
        )
        self.solve_statistics["safe_paths_time"] = time.time() - start_time

    if self.optimize_with_safe_sequences and not self.is_solved():
        start_time = time.time()
        self.safe_lists = safetypathcovers.safe_sequences(
            self.G,
            self.trusted_edges_for_safety,
            no_duplicates=False,
            threads=self.threads,
        )
        self.solve_statistics["safe_sequences_time"] = time.time() - start_time

    # some checks
    if self.safe_lists is not None and self.trusted_edges_for_safety is None:
        raise ValueError(
            "trusted_edges_for_safety must be provided when optimizing with safe lists"
        )
    if self.optimize_with_safe_paths and self.optimize_with_safe_sequences:
        raise ValueError("Cannot optimize with both safe paths and safe sequences")

__check_valid_subpath_constraints 

__check_valid_subpath_constraints()

Checks if the subpath constraints are valid.

Parameters
  • subpath_constraints (list): The subpath constraints to be checked.
Returns
  • True if the subpath constraints are valid, False otherwise.

The method checks if the subpath constraints are valid by ensuring that each subpath is a valid path in the graph.

Source code in flowpaths/abstractpathmodeldag.py 
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
def __check_valid_subpath_constraints(self):
    """
    Checks if the subpath constraints are valid.

    Parameters
    ----------
    - subpath_constraints (list): The subpath constraints to be checked.

    Returns
    ----------
    - True if the subpath constraints are valid, False otherwise.

    The method checks if the subpath constraints are valid by ensuring that each subpath
    is a valid path in the graph.
    """

    # Check that self.subpath_constraints is a list of lists of edges
    if not all(isinstance(subpath, list) for subpath in self.subpath_constraints):
        raise ValueError("subpath_constraints must be a list of lists of edges.")

    for subpath in self.subpath_constraints:
        for e in subpath:
            if not self.G.has_edge(e[0], e[1]):
                raise ValueError(f"Subpath {subpath} contains the edge {e} which is not in the graph.")

create_solver_and_paths 

create_solver_and_paths()

Creates a solver instance and encodes the paths in the graph.

This method initializes the solver with the specified parameters and encodes the paths by creating variables for edges and subpaths.

If external solution paths are provided, it skips the solver creation.

Source code in flowpaths/abstractpathmodeldag.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def create_solver_and_paths(self):
    """
    Creates a solver instance and encodes the paths in the graph.

    This method initializes the solver with the specified parameters and encodes the paths
    by creating variables for edges and subpaths.

    If external solution paths are provided, it skips the solver creation.
    """
    if self.external_solution_paths is not None:
        return

    self.solver = sw.SolverWrapper(
        external_solver=self.external_solver,
        threads=self.threads,
        time_limit=self.time_limit,
        presolve=self.presolve,
        log_to_console=self.log_to_console,
    )

    self.__encode_paths()

get_lowerbound_k abstractmethod 

get_lowerbound_k()

Implement this class in the child class to return a lower bound on the number of solution paths to the model. If you have no lower bound, you should implement this method to return 1.

Source code in flowpaths/abstractpathmodeldag.py 
565
566
567
568
569
570
571
@abstractmethod
def get_lowerbound_k(self):
    """
    Implement this class in the child class to return a lower bound on the number of solution paths to the model.
    If you have no lower bound, you should implement this method to return 1.
    """
    pass

get_objective_value abstractmethod 

get_objective_value()

Implement this class in the child class to return the objective value of the model. This is needed to be able to compute the safe paths (i.e. those appearing any optimum solution) for any child class.

A basic objective value is the num_paths (when we’re trying to minimize the number of paths). If your model has a different objective, you should implement this method to return the objective value of the model. If your model has no objective value, you should implement this method to return None.

Source code in flowpaths/abstractpathmodeldag.py 
674
675
676
677
678
679
680
681
682
683
684
@abstractmethod
def get_objective_value(self):
    """
    Implement this class in the child class to return the objective value of the model. This is needed to be able to
    compute the safe paths (i.e. those appearing any optimum solution) for any child class.

    A basic objective value is the num_paths (when we're trying to minimize the number of paths). If your model has a different
    objective, you should implement this method to return the objective value of the model. If your model has no objective value,
    you should implement this method to return None.
    """
    pass

get_solution abstractmethod 

get_solution()

Implement this class in the child class to return the full solution of the model. The solution paths are obtained with the get_solution_paths method.

Source code in flowpaths/abstractpathmodeldag.py 
557
558
559
560
561
562
563
@abstractmethod
def get_solution(self):
    """
    Implement this class in the child class to return the full solution of the model.
    The solution paths are obtained with the get_solution_paths method.
    """
    pass

get_solution_paths 

get_solution_paths() -> list

Retrieves the solution paths from the graph.

This method returns the solution paths either from the external solution paths if they are provided at initialization time, or by calculating them based on the edge variable solutions.

Returns
  • A list of paths, where each path is represented as a list of vertices.
Source code in flowpaths/abstractpathmodeldag.py 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
def get_solution_paths(self) -> list:
    """
    Retrieves the solution paths from the graph.

    This method returns the solution paths either from the external solution paths
    if they are provided at initialization time, or by calculating them based on the
    edge variable solutions.

    Returns
    ----------
    - A list of paths, where each path is represented as a list of vertices.
    """
    if self.external_solution_paths is not None:
        return self.external_solution_paths

    if self.edge_vars_sol == {}:
        self.edge_vars_sol = self.solver.get_variable_values(
            "edge", [str, str, int], 
            binary_values=True,
        )

    paths = []
    for i in range(self.k):
        vertex = self.G.source
        # checking if there is a path from source to sink
        found_path = False
        for out_neighbor in self.G.successors(vertex):
            if self.edge_vars_sol[(str(vertex), str(out_neighbor), i)] == 1:
                found_path = True
                break
        if not found_path:
            path = []
            paths.append(path)
            print("Warning: No path found for path index", i)
        else:
            path = [vertex]
            while vertex != self.G.sink:
                for out_neighbor in self.G.successors(vertex):
                    if self.edge_vars_sol[(str(vertex), str(out_neighbor), i)] == 1:
                        vertex = out_neighbor
                        break
                path.append(vertex)
            if len(path) < 2:
                raise Exception(f"Something went wrong, solution path {path} has less than 2 vertices. this should not happen. Make sure the stDiGraph has no edge from global source {self.G.source} to global sink {self.G.sink}.")

            paths.append(path[1:-1])

    return paths

is_valid_solution abstractmethod 

is_valid_solution() -> bool

Implement this class in the child class to perform a basic check whether the solution is valid.

If you cannot perform such a check, provide an implementation that always returns True.

Source code in flowpaths/abstractpathmodeldag.py 
622
623
624
625
626
627
628
629
@abstractmethod
def is_valid_solution(self) -> bool:
    """
    Implement this class in the child class to perform a basic check whether the solution is valid.

    If you cannot perform such a check, provide an implementation that always returns True.
    """
    pass

solve 

solve() -> bool

Solves the optimization model for the current instance.

Returns
  • True if the model is solved successfully, False otherwise.

The method first checks if an external solution is already provided. If so, it sets the solved attribute to True and returns True.

If not, it optimizes the model using the solver, and records the solve time and solver status in the solve_statistics dictionary. If the solver status indicates an optimal solution (either ‘kOptimal’ (highs) or status code 2 (gurobi)), it sets the solved attribute to True and returns True. Otherwise, it sets the solved attribute to False and returns False.

Source code in flowpaths/abstractpathmodeldag.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
def solve(self) -> bool:
    """
    Solves the optimization model for the current instance.

    Returns
    ----------
    - True if the model is solved successfully, False otherwise.

    The method first checks if an external solution is already provided. If so, it sets the
    solved attribute to True and returns True.

    If not, it optimizes the model using the solver, and records the solve time and solver status
    in the solve_statistics dictionary. If the solver status indicates an optimal solution
    (either 'kOptimal' (highs) or status code 2 (gurobi)), it sets the solved attribute to True and returns True.
    Otherwise, it sets the solved attribute to False and returns False.
    """
    # If we already received an external solution, we don't need to solve the model
    if self.external_solution_paths is not None:
        self.__is_solved = True
        return True

    # self.write_model(f"model-{self.id}.lp")
    start_time = time.time()
    self.solver.optimize()
    self.solve_statistics[f"milp_solve_time_for_num_paths_{self.k}"] = (
        time.time() - start_time
    )

    self.solve_statistics[f"milp_solver_status_for_num_paths_{self.k}"] = (
        self.solver.get_model_status()
    )

    if (
        self.solver.get_model_status() == "kOptimal"
        or self.solver.get_model_status() == 2
    ):
        self.__is_solved = True
        return True

    self.__is_solved = False
    return False