discrete_optimization.generic_tools.graph_solver package

Submodules

discrete_optimization.generic_tools.graph_solver.solver_graph module

DAG-based solver workflow (SolverGraph).

class discrete_optimization.generic_tools.graph_solver.solver_graph.BackTransformNode(node_id: str, transformation: ProblemTransformation, source_problem: Problem)

Bases: GraphNode

Node that back-transforms solutions through a transformation.

execute(**kwargs: Any) → NodeData

Back-transform all solutions.

Returns:

NodeData with back-transformed result, solution, and source problem

source_problem: Problem

transformation: ProblemTransformation

class discrete_optimization.generic_tools.graph_solver.solver_graph.GraphNode(node_id: str)

Bases: ABC

Base class for nodes in the solver graph.

A node takes inputs, executes an operation, and produces outputs. Data flows through the graph as NodeData objects.

can_execute() → bool

Check if all required inputs are available.

Default: can execute if we have at least one input. Override in subclasses for specific requirements.

Returns:

True if node can execute

abstractmethod execute(**kwargs: Any) → NodeData

Execute the node’s operation.

Parameters:

**kwargs – Execution parameters (e.g., time_limit)

Returns:

NodeData with problem, result, and/or solution

get_input_problem() → Problem

Extract problem from inputs.

Returns:

Problem instance from first input that has one

get_input_solution() → Solution | None

Extract best solution from inputs for warmstart.

Returns:

Solution if available, None otherwise

inputs: dict[str, NodeData]

node_id: str

output: NodeData | None

class discrete_optimization.generic_tools.graph_solver.solver_graph.MergeNode(node_id: str, strategy: str = 'best')

Bases: GraphNode

Node that merges multiple result storages.

can_execute() → bool

Can execute if we have at least one input.

execute(**kwargs: Any) → NodeData

Merge input result storages.

Returns:

NodeData with merged result and solution

strategy: str

class discrete_optimization.generic_tools.graph_solver.solver_graph.NodeData(problem: Problem | None = None, result: ResultStorage | None = None, solution: Solution | None = None)

Bases: object

Data flowing through graph nodes.

problem

Problem instance (required for most nodes)

Type:

discrete_optimization.generic_tools.do_problem.Problem | None

result

ResultStorage from solver nodes (optional)

Type:

discrete_optimization.generic_tools.result_storage.result_storage.ResultStorage | None

solution

Single best solution for warmstart/kwargs extraction (optional)

Type:

discrete_optimization.generic_tools.do_problem.Solution | None

problem: Problem | None = None

result: ResultStorage | None = None

solution: Solution | None = None

class discrete_optimization.generic_tools.graph_solver.solver_graph.RootNode(node_id: str, problem: Problem)

Bases: GraphNode

Virtual root node that provides the source problem.

can_execute() → bool

Root can always execute.

execute(**kwargs: Any) → NodeData

Return the source problem.

Returns:

NodeData with problem

problem: Problem

class discrete_optimization.generic_tools.graph_solver.solver_graph.SolverGraph(source_problem: Problem)

Bases: object

DAG-based solver workflow.

Supports: - Branching (parallel strategies) - Merging (combine results) - Transformations (problem conversion) - Arbitrary directed acyclic graphs

Example (linear, like SequentialMetasolver): # >>> graph = SolverGraph(problem) # >>> graph.add_solver(“solver1”, SubBrick(cls=Solver1, kwargs={})) # >>> graph.add_solver(“solver2”, SubBrick(cls=Solver2, kwargs={})) # >>> graph.add_edge(“root”, “solver1”) # >>> graph.add_edge(“solver1”, “solver2”) # >>> result = graph.run() # # Example (branching): # >>> graph = SolverGraph(problem) # >>> graph.add_solver(“cpsat”, SubBrick(cls=CPSat, kwargs={})) # >>> graph.add_solver(“lp”, SubBrick(cls=LP, kwargs={})) # >>> graph.add_merge(“merge”, strategy=”best”) # >>> graph.add_edge(“root”, “cpsat”) # >>> graph.add_edge(“root”, “lp”) # >>> graph.add_edge(“cpsat”, “merge”) # >>> graph.add_edge(“lp”, “merge”) # >>> result = graph.run()

add_back_transform(node_id: str, transformation: ProblemTransformation, source_problem: Problem) → str

Add a back-transformation node.

Parameters:

• node_id – Unique identifier for this node

• transformation – Transformation to reverse

• source_problem – Original problem

Returns:

Node ID (for chaining)

add_edge(from_node: str, to_node: str) → None

Add an edge between nodes.

Parameters:

• from_node – Source node ID

• to_node – Target node ID

Raises:

ValueError – If nodes don’t exist

add_merge(node_id: str, strategy: str = 'best') → str

Add a merge node.

Parameters:

• node_id – Unique identifier for this node

• strategy – Merge strategy (“best” or “all”)

Returns:

Node ID (for chaining)

add_solver(node_id: str, solver_brick: SubBrick) → str

Add a solver node.

Parameters:

• node_id – Unique identifier for this node

• solver_brick – Solver specification

Returns:

Node ID (for chaining)

add_transformation(node_id: str, transformation: ProblemTransformation) → str

Add a transformation node.

Parameters:

• node_id – Unique identifier for this node

• transformation – Transformation to apply

Returns:

Node ID (for chaining)

edges: dict[str, list[str]]

node_outputs: dict[str, NodeData]

nodes: dict[str, GraphNode]

reverse_edges: dict[str, list[str]]

run(**solve_kwargs: Any) → ResultStorage

Execute the graph and return final results.

Parameters:

**solve_kwargs – Keyword arguments passed to all solver nodes

Returns:

ResultStorage with final solutions

Raises:

RuntimeError – If execution fails

source_problem: Problem

topological_sort() → list[str]

Return nodes in topological order using NetworkX.

Returns:

List of node IDs in execution order

Raises:

ValueError – If graph contains a cycle

visualize() → str

Create ASCII art visualization of the graph.

Returns:

String representation of the graph

class discrete_optimization.generic_tools.graph_solver.solver_graph.SolverNode(node_id: str, solver_brick: SubBrick)

Bases: GraphNode

Node that runs a solver.

execute(**kwargs: Any) → NodeData

Solve the input problem.

Returns:

NodeData with result, solution, and problem

problem: Problem | None

solver: SolverDO | None

solver_brick: SubBrick

class discrete_optimization.generic_tools.graph_solver.solver_graph.TransformationNode(node_id: str, transformation: ProblemTransformation)

Bases: GraphNode

Node that applies a problem transformation.

execute(**kwargs: Any) → NodeData

Transform the input problem.

Returns:

NodeData with transformed problem

transformation: ProblemTransformation

Module contents

DAG-based solver workflows (SolverGraph).

class discrete_optimization.generic_tools.graph_solver.BackTransformNode(node_id: str, transformation: ProblemTransformation, source_problem: Problem)

Bases: GraphNode

Node that back-transforms solutions through a transformation.

execute(**kwargs: Any) → NodeData

Back-transform all solutions.

Returns:

NodeData with back-transformed result, solution, and source problem

source_problem: Problem

transformation: ProblemTransformation

class discrete_optimization.generic_tools.graph_solver.GraphNode(node_id: str)

Bases: ABC

Base class for nodes in the solver graph.

A node takes inputs, executes an operation, and produces outputs. Data flows through the graph as NodeData objects.

can_execute() → bool

Check if all required inputs are available.

Default: can execute if we have at least one input. Override in subclasses for specific requirements.

Returns:

True if node can execute

abstractmethod execute(**kwargs: Any) → NodeData

Execute the node’s operation.

Parameters:

**kwargs – Execution parameters (e.g., time_limit)

Returns:

NodeData with problem, result, and/or solution

get_input_problem() → Problem

Extract problem from inputs.

Returns:

Problem instance from first input that has one

get_input_solution() → Solution | None

Extract best solution from inputs for warmstart.

Returns:

Solution if available, None otherwise

inputs: dict[str, NodeData]

node_id: str

output: NodeData | None

class discrete_optimization.generic_tools.graph_solver.MergeNode(node_id: str, strategy: str = 'best')

Bases: GraphNode

Node that merges multiple result storages.

can_execute() → bool

Can execute if we have at least one input.

execute(**kwargs: Any) → NodeData

Merge input result storages.

Returns:

NodeData with merged result and solution

strategy: str

class discrete_optimization.generic_tools.graph_solver.NodeData(problem: Problem | None = None, result: ResultStorage | None = None, solution: Solution | None = None)

Bases: object

Data flowing through graph nodes.

problem

Problem instance (required for most nodes)

Type:

discrete_optimization.generic_tools.do_problem.Problem | None

result

ResultStorage from solver nodes (optional)

Type:

discrete_optimization.generic_tools.result_storage.result_storage.ResultStorage | None

solution

Single best solution for warmstart/kwargs extraction (optional)

Type:

discrete_optimization.generic_tools.do_problem.Solution | None

problem: Problem | None = None

result: ResultStorage | None = None

solution: Solution | None = None

class discrete_optimization.generic_tools.graph_solver.SolverGraph(source_problem: Problem)

Bases: object

DAG-based solver workflow.

Supports: - Branching (parallel strategies) - Merging (combine results) - Transformations (problem conversion) - Arbitrary directed acyclic graphs

Example (linear, like SequentialMetasolver): # >>> graph = SolverGraph(problem) # >>> graph.add_solver(“solver1”, SubBrick(cls=Solver1, kwargs={})) # >>> graph.add_solver(“solver2”, SubBrick(cls=Solver2, kwargs={})) # >>> graph.add_edge(“root”, “solver1”) # >>> graph.add_edge(“solver1”, “solver2”) # >>> result = graph.run() # # Example (branching): # >>> graph = SolverGraph(problem) # >>> graph.add_solver(“cpsat”, SubBrick(cls=CPSat, kwargs={})) # >>> graph.add_solver(“lp”, SubBrick(cls=LP, kwargs={})) # >>> graph.add_merge(“merge”, strategy=”best”) # >>> graph.add_edge(“root”, “cpsat”) # >>> graph.add_edge(“root”, “lp”) # >>> graph.add_edge(“cpsat”, “merge”) # >>> graph.add_edge(“lp”, “merge”) # >>> result = graph.run()

add_back_transform(node_id: str, transformation: ProblemTransformation, source_problem: Problem) → str

Add a back-transformation node.

Parameters:

• node_id – Unique identifier for this node

• transformation – Transformation to reverse

• source_problem – Original problem

Returns:

Node ID (for chaining)

add_edge(from_node: str, to_node: str) → None

Add an edge between nodes.

Parameters:

• from_node – Source node ID

• to_node – Target node ID

Raises:

ValueError – If nodes don’t exist

add_merge(node_id: str, strategy: str = 'best') → str

Add a merge node.

Parameters:

• node_id – Unique identifier for this node

• strategy – Merge strategy (“best” or “all”)

Returns:

Node ID (for chaining)

add_solver(node_id: str, solver_brick: SubBrick) → str

Add a solver node.

Parameters:

• node_id – Unique identifier for this node

• solver_brick – Solver specification

Returns:

Node ID (for chaining)

add_transformation(node_id: str, transformation: ProblemTransformation) → str

Add a transformation node.

Parameters:

• node_id – Unique identifier for this node

• transformation – Transformation to apply

Returns:

Node ID (for chaining)

edges: dict[str, list[str]]

node_outputs: dict[str, NodeData]

nodes: dict[str, GraphNode]

reverse_edges: dict[str, list[str]]

run(**solve_kwargs: Any) → ResultStorage

Execute the graph and return final results.

Parameters:

**solve_kwargs – Keyword arguments passed to all solver nodes

Returns:

ResultStorage with final solutions

Raises:

RuntimeError – If execution fails

source_problem: Problem

topological_sort() → list[str]

Return nodes in topological order using NetworkX.

Returns:

List of node IDs in execution order

Raises:

ValueError – If graph contains a cycle

visualize() → str

Create ASCII art visualization of the graph.

Returns:

String representation of the graph

class discrete_optimization.generic_tools.graph_solver.SolverNode(node_id: str, solver_brick: SubBrick)

Bases: GraphNode

Node that runs a solver.

execute(**kwargs: Any) → NodeData

Solve the input problem.

Returns:

NodeData with result, solution, and problem

problem: Problem | None

solver: SolverDO | None

solver_brick: SubBrick

class discrete_optimization.generic_tools.graph_solver.TransformationNode(node_id: str, transformation: ProblemTransformation)

Bases: GraphNode

Node that applies a problem transformation.

execute(**kwargs: Any) → NodeData

Transform the input problem.

Returns:

NodeData with transformed problem

transformation: ProblemTransformation