Minimal value stream optimization minimizes the price of shifting stream by way of a community of nodes and edges. Nodes embody sources (provide) and sinks (demand), with totally different prices and capability limits. The goal is to search out the least expensive technique to transfer quantity from sources to sinks whereas adhering to all capability limitations.
Purposes
Purposes of minimal value stream optimization are huge and different, spanning a number of industries and sectors. This strategy is essential in logistics and provide chain administration, the place it’s used to reduce transportation prices whereas guaranteeing well timed supply of products. In telecommunications, it helps in optimizing the routing of information by way of networks to scale back latency and enhance bandwidth utilization. The vitality sector leverages minimal value stream optimization to effectively distribute electrical energy by way of energy grids, decreasing losses and operational prices. City planning and infrastructure growth additionally profit from this optimization approach, because it assists in designing environment friendly public transportation programs and water distribution networks.
Instance
Beneath is an easy stream optimization instance:
The picture above illustrates a minimal value stream optimization downside with six nodes and eight edges. Nodes A and B function sources, every with a provide of fifty items, whereas nodes E and F act as sinks, every with a requirement of 40 items. Each edge has a most capability of 25 items, with variable prices indicated within the picture. The target of the optimization is to allocate stream on every edge to maneuver the required items from nodes A and B to nodes E and F, respecting the sting capacities on the lowest doable value.
Node F can solely obtain provide from node B. There are two paths: immediately or by way of node D. The direct path has a price of two, whereas the oblique path by way of D has a mixed value of three. Thus, 25 items (the utmost edge capability) are moved immediately from B to F. The remaining 15 items are routed by way of B -D-F to fulfill the demand.
At the moment, 40 out of fifty items have been transferred from node B, leaving a remaining provide of 10 items that may be moved to node E. The out there pathways for supplying node E embody: A-E and B-E with a price of three, A-C-E with a price of 4, and B-C-E with a price of 5. Consequently, 25 items are transported from A-E (restricted by the sting capability) and 10 items from B-E (restricted by the remaining provide at node B). To fulfill the demand of 40 items at node E, an extra 5 items are moved by way of A-C-E, leading to no stream being allotted to the B-C pathway.
Mathematical formulation
I introduce two mathematical formulations of minimal value stream optimization:
1. LP (linear program) with steady variables solely
2. MILP (combined integer linear program) with steady and discrete variables
I’m utilizing following definitions:
LP formulation
This formulation solely incorporates determination variables which might be steady, that means they’ll have any worth so long as all constraints are fulfilled. Resolution variables are on this case the stream variables x(u, v) of all edges.
The target operate describes how the prices which might be presupposed to be minimized are calculated. On this case it’s outlined because the stream multiplied with the variable value summed up over all edges:
Constraints are circumstances that should be glad for the answer to be legitimate, guaranteeing that the stream doesn’t exceed capability limitations.
First, all flows should be non-negative and never exceed to edge capacities:
Move conservation constraints be certain that the identical quantity of stream that goes right into a node has to return out of the node. These constraints are utilized to all nodes which might be neither sources nor sinks:
For supply and sink nodes the distinction of out stream and in stream is smaller or equal the availability of the node:
If v is a supply the distinction of outflow minus influx should not exceed the availability s(v). In case v is a sink node we don’t permit that greater than -s(v) can stream into the node than out of the node (for sinks s(v) is unfavorable).
MILP
Moreover, to the continual variables of the LP formulation, the MILP formulation additionally incorporates discreate variables that may solely have particular values. Discrete variables permit to limit the variety of used nodes or edges to sure values. It can be used to introduce mounted prices for utilizing nodes or edges. On this article I present easy methods to add mounted prices. You will need to be aware that including discrete determination variables makes it rather more troublesome to search out an optimum answer, therefore this formulation ought to solely be used if a LP formulation isn’t doable.
The target operate is outlined as:
With three phrases: variable value of all edges, mounted value of all edges, and stuck value of all nodes.
The utmost stream that may be allotted to an edge depends upon the sting’s capability, the sting choice variable, and the origin node choice variable:
This equation ensures that stream can solely be assigned to edges if the sting choice variable and the origin node choice variable are 1.
The stream conservation constraints are equal to the LP downside.
Implementation
On this part I clarify easy methods to implement a MILP optimization in Python. Yow will discover the code on this repo.
Libraries
To construct the stream community, I used NetworkX which is a superb library (https://networkx.org/) for working with graphs. There are lots of fascinating articles that show how highly effective and simple to make use of NetworkX is to work with graphs, i.a. customizing NetworkX Graphs, NetworkX: Code Demo for Manipulating Subgraphs, Social Network Analysis with NetworkX: A Gentle Introduction.
One necessary facet when constructing an optimization is to be sure that the enter is accurately outlined. Even one small error could make the issue infeasible or can result in an surprising answer. To keep away from this, I used Pydantic to validate the consumer enter and lift any points on the earliest doable stage. This article provides a straightforward to grasp introduction to Pydantic.
To remodel the outlined community right into a mathematical optimization downside I used PuLP. Which permits to outline all variables and constraint in an intuitive method. This library additionally has the benefit that it could possibly use many alternative solvers in a easy pug-and-play trend. This article supplies good introduction to this library.
Defining nodes and edges
The code beneath exhibits how nodes are outlined:
from pydantic import BaseModel, model_validator
from typing import Elective
# node and edge definitions
class Node(BaseModel, frozen=True):
"""
class of community node with attributes:
title: str - title of node
demand: float - demand of node (if node is sink)
provide: float - provide of node (if node is supply)
capability: float - most stream out of node
kind: str - kind of node
x: float - x-coordinate of node
y: float - y-coordinate of node
fixed_cost: float - value of choosing node
"""
title: str
demand: Elective[float] = 0.0
provide: Elective[float] = 0.0
capability: Elective[float] = float('inf')
kind: Elective[str] = None
x: Elective[float] = 0.0
y: Elective[float] = 0.0
fixed_cost: Elective[float] = 0.0
@model_validator(mode="after")
def validate(self):
"""
validate if node definition are appropriate
"""
# verify that demand is non-negative
if self.demand < 0 or self.demand == float('inf'): elevate ValueError('demand should be non-negative and finite')
# verify that offer is non-negative
if self.provide < 0: elevate ValueError('provide should be non-negative')
# verify that capability is non-negative
if self.capability < 0: elevate ValueError('capability should be non-negative')
# verify that fixed_cost is non-negative
if self.fixed_cost < 0: elevate ValueError('fixed_cost should be non-negative')
return selfNodes are outlined by way of the Node class which is inherited from Pydantic’s BaseModel. This permits an computerized validation that ensures that every one properties are outlined with the right datatype every time a brand new object is created. On this case solely the title is a required enter, all different properties are elective, if they aren’t offered the required default worth is assigned to them. By setting the “frozen” parameter to True I made all properties immutable, that means they can’t be modified after the item has been initialized.
The validate methodology is executed after the item has been initialized and applies extra checks to make sure the offered values are as anticipated. Particularly it checks that demand, provide, capability, variable value and stuck value will not be unfavorable. Moreover, it additionally doesn’t permit infinite demand as this could result in an infeasible optimization downside.
These checks look trivial, nonetheless their foremost profit is that they’ll set off an error on the earliest doable stage when an enter is inaccurate. Thus, they forestall making a optimization mannequin that’s incorrect. Exploring why a mannequin can’t be solved can be rather more time consuming as there are various elements that may have to be analyzed, whereas such “trivial” enter error will not be the primary facet to analyze.
Edges are applied as follows:
class Edge(BaseModel, frozen=True):
"""
class of edge between two nodes with attributes:
origin: 'Node' - origin node of edge
vacation spot: 'Node' - vacation spot node of edge
capability: float - most stream by way of edge
variable_cost: float - value per unit stream by way of edge
fixed_cost: float - value of choosing edge
"""
origin: Node
vacation spot: Node
capability: Elective[float] = float('inf')
variable_cost: Elective[float] = 0.0
fixed_cost: Elective[float] = 0.0@model_validator(mode="after")
def validate(self):
"""
validate of edge definition is appropriate
"""
# verify that node names are totally different
if self.origin.title == self.vacation spot.title: elevate ValueError('origin and vacation spot names should be totally different')
# verify that capability is non-negative
if self.capability < 0: elevate ValueError('capability should be non-negative')
# verify that variable_cost is non-negative
if self.variable_cost < 0: elevate ValueError('variable_cost should be non-negative')
# verify that fixed_cost is non-negative
if self.fixed_cost < 0: elevate ValueError('fixed_cost should be non-negative')
return self
The required inputs are an origin node and a vacation spot node object. Moreover, capability, variable value and stuck value may be offered. The default worth for capability is infinity which implies if no capability worth is offered it’s assumed the sting doesn’t have a capability limitation. The validation ensures that the offered values are non-negative and that origin node title and the vacation spot node title are totally different.
Initialization of flowgraph object
To outline the flowgraph and optimize the stream I created a brand new class known as FlowGraph that’s inherited from NetworkX’s DiGraph class. By doing this I can add my very own strategies which might be particular to the stream optimization and on the identical time use all strategies DiGraph supplies:
from networkx import DiGraph
from pulp import LpProblem, LpVariable, LpMinimize, LpStatus
class FlowGraph(DiGraph):
"""
class to outline and clear up minimal value stream issues
"""
def __init__(self, nodes=[], edges=[]):
"""
initialize FlowGraph object
:param nodes: record of nodes
:param edges: record of edges
"""
# initialialize digraph
tremendous().__init__(None)
# add nodes and edges
for node in nodes: self.add_node(node)
for edge in edges: self.add_edge(edge)
def add_node(self, node):
"""
add node to graph
:param node: Node object
"""
# verify if node is a Node object
if not isinstance(node, Node): elevate ValueError('node should be a Node object')
# add node to graph
tremendous().add_node(node.title, demand=node.demand, provide=node.provide, capability=node.capability, kind=node.kind,
fixed_cost=node.fixed_cost, x=node.x, y=node.y)
def add_edge(self, edge):
"""
add edge to graph
@param edge: Edge object
"""
# verify if edge is an Edge object
if not isinstance(edge, Edge): elevate ValueError('edge should be an Edge object')
# verify if nodes exist
if not edge.origin.title in tremendous().nodes: self.add_node(edge.origin)
if not edge.vacation spot.title in tremendous().nodes: self.add_node(edge.vacation spot)
# add edge to graph
tremendous().add_edge(edge.origin.title, edge.vacation spot.title, capability=edge.capability,
variable_cost=edge.variable_cost, fixed_cost=edge.fixed_cost)
FlowGraph is initialized by offering a listing of nodes and edges. Step one is to initialize the mother or father class as an empty graph. Subsequent, nodes and edges are added by way of the strategies add_node and add_edge. These strategies first verify if the offered ingredient is a Node or Edge object. If this isn’t the case an error can be raised. This ensures that every one components added to the graph have handed the validation of the earlier part. Subsequent, the values of those objects are added to the Digraph object. Notice that the Digraph class additionally makes use of add_node and add_edge strategies to take action. Through the use of the identical methodology title I’m overwriting these strategies to make sure that every time a brand new ingredient is added to the graph it should be added by way of the FlowGraph strategies which validate the item kind. Thus, it isn’t doable to construct a graph with any ingredient that has not handed the validation checks.
Initializing the optimization downside
The tactic beneath converts the community into an optimization mannequin, solves it, and retrieves the optimized values.
def min_cost_flow(self, verbose=True):
"""
run minimal value stream optimization
@param verbose: bool - print optimization standing (default: True)
@return: standing of optimization
"""
self.verbose = verbose
# get most stream
self.max_flow = sum(node['demand'] for _, node in tremendous().nodes.information() if node['demand'] > 0)
start_time = time.time()
# create LP downside
self.prob = LpProblem("FlowGraph.min_cost_flow", LpMinimize)
# assign determination variables
self._assign_decision_variables()
# assign goal operate
self._assign_objective_function()
# assign constraints
self._assign_constraints()
if self.verbose: print(f"Mannequin creation time: {time.time() - start_time:.2f} s")
start_time = time.time()
# clear up LP downside
self.prob.clear up()
solve_time = time.time() - start_time
# get standing
standing = LpStatus[self.prob.status]
if verbose:
# print optimization standing
if standing == 'Optimum':
# get goal worth
goal = self.prob.goal.worth()
print(f"Optimum answer discovered: {goal:.2f} in {solve_time:.2f} s")
else:
print(f"Optimization standing: {standing} in {solve_time:.2f} s")
# assign variable values
self._assign_variable_values(standing=='Optimum')
return standingPulp’s LpProblem is initialized, the fixed LpMinimize defines it as a minimization downside — that means it’s supposed to reduce the worth of the target operate. Within the following traces all determination variables are initialized, the target operate in addition to all constraints are outlined. These strategies can be defined within the following sections.
Subsequent, the issue is solved, on this step the optimum worth of all determination variables is decided. Following the standing of the optimization is retrieved. When the standing is “Optimum” an optimum answer may very well be discovered different statuses are “Infeasible” (it isn’t doable to meet all constraints), “Unbounded” (the target operate can have an arbitrary low values), and “Undefined” that means the issue definition isn’t full. In case no optimum answer was discovered the issue definition must be reviewed.
Lastly, the optimized values of all variables are retrieved and assigned to the respective nodes and edges.
Defining determination variables
All determination variables are initialized within the methodology beneath:
def _assign_variable_values(self, opt_found):
"""
assign determination variable values if optimum answer discovered, in any other case set to None
@param opt_found: bool - if optimum answer was discovered
"""
# assign edge values
for _, _, edge in tremendous().edges.information():
# initialize values
edge['flow'] = None
edge['selected'] = None
# verify if optimum answer discovered
if opt_found and edge['flow_var'] isn't None:
edge['flow'] = edge['flow_var'].varValue
if edge['selection_var'] isn't None:
edge['selected'] = edge['selection_var'].varValue
# assign node values
for _, node in tremendous().nodes.information():
# initialize values
node['selected'] = None
if opt_found:
# verify if node has choice variable
if node['selection_var'] isn't None:
node['selected'] = node['selection_var'].varValue
First it iterates by way of all edges and assigns steady determination variables if the sting capability is bigger than 0. Moreover, if mounted prices of the sting are better than 0 a binary determination variable is outlined as nicely. Subsequent, it iterates by way of all nodes and assigns binary determination variables to nodes with mounted prices. The whole variety of steady and binary determination variables is counted and printed on the finish of the strategy.
Defining goal
In any case determination variables have been initialized the target operate may be outlined:
def _assign_objective_function(self):
"""
outline goal operate
"""
goal = 0
# add edge prices
for _, _, edge in tremendous().edges.information():
if edge['selection_var'] isn't None: goal += edge['selection_var'] * edge['fixed_cost']
if edge['flow_var'] isn't None: goal += edge['flow_var'] * edge['variable_cost']
# add node prices
for _, node in tremendous().nodes.information():
# add node choice prices
if node['selection_var'] isn't None: goal += node['selection_var'] * node['fixed_cost']
self.prob += goal, 'Goal',
The target is initialized as 0. Then for every edge mounted prices are added if the sting has a range variable, and variable prices are added if the sting has a stream variable. For all nodes with choice variables mounted prices are added to the target as nicely. On the finish of the strategy the target is added to the LP object.
Defining constraints
All constraints are outlined within the methodology beneath:
def _assign_constraints(self):
"""
outline constraints
"""
# depend of contraints
constr_count = 0
# add capability constraints for edges with mounted prices
for origin_name, destination_name, edge in tremendous().edges.information():
# get capability
capability = edge['capacity'] if edge['capacity'] < float('inf') else self.max_flow
rhs = capability
if edge['selection_var'] isn't None: rhs *= edge['selection_var']
self.prob += edge['flow_var'] <= rhs, f"capacity_{origin_name}-{destination_name}",
constr_count += 1
# get origin node
origin_node = tremendous().nodes[origin_name]
# verify if origin node has a range variable
if origin_node['selection_var'] isn't None:
rhs = capability * origin_node['selection_var']
self.prob += (edge['flow_var'] <= rhs, f"node_selection_{origin_name}-{destination_name}",)
constr_count += 1
total_demand = total_supply = 0
# add stream conservation constraints
for node_name, node in tremendous().nodes.information():
# combination out and in flows
in_flow = 0
for _, _, edge in tremendous().in_edges(node_name, information=True):
if edge['flow_var'] isn't None: in_flow += edge['flow_var']
out_flow = 0
for _, _, edge in tremendous().out_edges(node_name, information=True):
if edge['flow_var'] isn't None: out_flow += edge['flow_var']
# add node capability contraint
if node['capacity'] < float('inf'):
self.prob += out_flow <= node['capacity'], f"node_capacity_{node_name}",
constr_count += 1
# verify what kind of node it's
if node['demand'] == node['supply']:
# transshipment node: in_flow = out_flow
self.prob += in_flow == out_flow, f"flow_balance_{node_name}",
else:
# in_flow - out_flow >= demand - provide
rhs = node['demand'] - node['supply']
self.prob += in_flow - out_flow >= rhs, f"flow_balance_{node_name}",
constr_count += 1
# replace complete demand and provide
total_demand += node['demand']
total_supply += node['supply']
if self.verbose:
print(f"Constraints: {constr_count}")
print(f"Complete provide: {total_supply}, Complete demand: {total_demand}")First, capability constraints are outlined for every edge. If the sting has a range variable the capability is multiplied with this variable. In case there is no such thing as a capability limitation (capability is ready to infinity) however there’s a choice variable, the choice variable is multiplied with the utmost stream that has been calculated by aggregating the demand of all nodes. A further constraint is added in case the sting’s origin node has a range variable. This constraint signifies that stream can solely come out of this node if the choice variable is ready to 1.
Following, the stream conservation constraints for all nodes are outlined. To take action the whole in and outflow of the node is calculated. Getting all in and outgoing edges can simply be finished by utilizing the in_edges and out_edges strategies of the DiGraph class. If the node has a capability limitation the utmost outflow can be constraint by that worth. For the stream conservation it’s essential to verify if the node is both a supply or sink node or a transshipment node (demand equals provide). Within the first case the distinction between influx and outflow should be better or equal the distinction between demand and provide whereas within the latter case in and outflow should be equal.
The whole variety of constraints is counted and printed on the finish of the strategy.
Retrieving optimized values
After working the optimization, the optimized variable values may be retrieved with the next methodology:
def _assign_variable_values(self, opt_found):
"""
assign determination variable values if optimum answer discovered, in any other case set to None
@param opt_found: bool - if optimum answer was discovered
"""
# assign edge values
for _, _, edge in tremendous().edges.information():
# initialize values
edge['flow'] = None
edge['selected'] = None
# verify if optimum answer discovered
if opt_found and edge['flow_var'] isn't None:
edge['flow'] = edge['flow_var'].varValue
if edge['selection_var'] isn't None:
edge['selected'] = edge['selection_var'].varValue
# assign node values
for _, node in tremendous().nodes.information():
# initialize values
node['selected'] = None
if opt_found:
# verify if node has choice variable
if node['selection_var'] isn't None:
node['selected'] = node['selection_var'].varValue
This methodology iterates by way of all edges and nodes, checks if determination variables have been assigned and provides the choice variable worth by way of varValue to the respective edge or node.
Demo
To show easy methods to apply the stream optimization I created a provide chain community consisting of two factories, 4 distribution facilities (DC), and 15 markets. All items produced by the factories need to stream by way of one distribution middle till they are often delivered to the markets.
Node properties have been outlined:
Ranges imply that uniformly distributed random numbers have been generated to assign these properties. Since Factories and DCs have mounted prices the optimization additionally must resolve which of those entities ought to be chosen.
Edges are generated between all Factories and DCs, in addition to all DCs and Markets. The variable value of edges is calculated because the Euclidian distance between origin and vacation spot node. Capacities of edges from Factories to DCs are set to 350 whereas from DCs to Markets are set to 100.
The code beneath exhibits how the community is outlined and the way the optimization is run:
# Outline nodes
factories = [Node(name=f'Factory {i}', supply=700, type="Factory", fixed_cost=100, x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(2)]
dcs = [Node(name=f'DC {i}', fixed_cost=25, capacity=500, type="DC", x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(4)]
markets = [Node(name=f'Market {i}', demand=random.randint(1, 100), type="Market", x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(15)]
# Outline edges
edges = []
# Factories to DCs
for manufacturing facility in factories:
for dc in dcs:
distance = ((manufacturing facility.x - dc.x)**2 + (manufacturing facility.y - dc.y)**2)**0.5
edges.append(Edge(origin=manufacturing facility, vacation spot=dc, capability=350, variable_cost=distance))
# DCs to Markets
for dc in dcs:
for market in markets:
distance = ((dc.x - market.x)**2 + (dc.y - market.y)**2)**0.5
edges.append(Edge(origin=dc, vacation spot=market, capability=100, variable_cost=distance))
# Create FlowGraph
G = FlowGraph(edges=edges)
G.min_cost_flow()The output of stream optimization is as follows:
Variable varieties: 68 steady, 6 binary
Constraints: 161
Complete provide: 1400.0, Complete demand: 909.0
Mannequin creation time: 0.00 s
Optimum answer discovered: 1334.88 in 0.23 sThe issue consists of 68 steady variables that are the sides’ stream variables and 6 binary determination variables that are the choice variables of the Factories and DCs. There are 161 constraints in complete which encompass edge and node capability constraints, node choice constraints (edges can solely have stream if the origin node is chosen), and stream conservation constraints. The following line exhibits that the whole provide is 1400 which is greater than the whole demand of 909 (if the demand was greater than the availability the issue can be infeasible). Since it is a small optimization downside, the time to outline the optimization mannequin was lower than 0.01 seconds. The final line exhibits that an optimum answer with an goal worth of 1335 may very well be present in 0.23 seconds.
Moreover, to the code I described on this put up I additionally added two strategies that visualize the optimized answer. The code of those strategies can be discovered within the repo.
All nodes are positioned by their respective x and y coordinates. The node and edge measurement is relative to the whole quantity that’s flowing by way of. The sting coloration refers to its utilization (stream over capability). Dashed traces present edges with out stream allocation.
Within the optimum answer each Factories have been chosen which is inevitable as the utmost provide of 1 Manufacturing unit is 700 and the whole demand is 909. Nevertheless, solely 3 of the 4 DCs are used (DC 0 has not been chosen).
Typically the plot exhibits the Factories are supplying the closest DCs and DCs the closest Markets. Nevertheless, there are just a few exceptions to this commentary: Manufacturing unit 0 additionally provides DC 3 though Manufacturing unit 1 is nearer. That is as a result of capability constraints of the sides which solely permit to maneuver at most 350 items per edge. Nevertheless, the closest Markets to DC 3 have a barely greater demand, therefore Manufacturing unit 0 is shifting further items to DC 3 to fulfill that demand. Though Market 9 is closest to DC 3 it’s equipped by DC 2. It’s because DC 3 would require an extra provide from Manufacturing unit 0 to provide this market and for the reason that complete distance from Manufacturing unit 0 over DC 3 is longer than the gap from Manufacturing unit 0 by way of DC 2, Market 9 is equipped by way of the latter route.
One other technique to visualize the outcomes is by way of a Sankey diagram which focuses on visualizing the flows of the sides:
The colours signify the sides’ utilizations with lowest utilizations in inexperienced altering to yellow and pink for the best utilizations. This diagram exhibits very nicely how a lot stream goes by way of every node and edge. It highlights the stream from Manufacturing unit 0 to DC 3 and in addition that Market 13 is equipped by DC 2 and DC 1.
Abstract
Minimal value stream optimizations could be a very useful software in lots of domains like logistics, transportation, telecommunication, vitality sector and lots of extra. To use this optimization you will need to translate a bodily system right into a mathematical graph consisting of nodes and edges. This ought to be finished in a technique to have as few discrete (e.g. binary) determination variables as crucial as these make it considerably tougher to search out an optimum answer. By combining Python’s NetworkX, Pulp and Pydantic libraries I constructed an stream optimization class that’s intuitive to initialize and on the identical time follows a generalized formulation which permits to use it in many alternative use instances. Graph and stream diagrams are very useful to grasp the answer discovered by the optimizer.
If not in any other case acknowledged all photographs have been created by the creator.
Source link
