使用非连接图或工具的车辆路径概率 Python
Vehicle Routing Prob with non-connected graph or-tools Python
我正在使用这个 python 库或工具：https://developers.google.com/optimization/routing/tsp/vehicle_routing
（代码可以在这里找到）。
问题是，当您 运行 解决方案时，它会为您提供一条覆盖所有节点的路径。但是我的项目需要对节点之间的路径进行约束。例如，如果您在节点 {3} 上，则您可能无法前往节点 {18}。或者换句话说，如果您在节点 {5}，您只能前往节点 {1、12、14}。我不确定如何将此约束添加到当前代码示例中。
请允许我进一步解释...
如果我们看这张图：
您可能会在此处看到此图的表示形式：
https://www.datacamp.com/community/tutorials/networkx-python-graph-tutorial
显然在这个问题中你不能从其他节点前往某些节点。我在 google or-tools 示例中使用此图中的数据来获取车辆路径问题的解决方案。
这是我的代码：
import math
from ortools.constraint_solver import pywrapcp
from ortools.constraint_solver import routing_enums_pb2
import pandas as pd
from sklearn import preprocessing
def distance(x1, y1, x2, y2):
dist = ((x1 - x2)**2 + (y1 - y2)**2)**(1/2)
return dist
class CreateDistanceCallback(object):
"""Create callback to calculate distances between points."""
def __init__(self, locations):
"""Initialize distance array."""
size = len(locations)
self.matrix = {}
for from_node in range(size):
self.matrix[from_node] = {}
for to_node in range(size):
x1 = locations[from_node][0]
y1 = locations[from_node][1]
x2 = locations[to_node][0]
y2 = locations[to_node][1]
self.matrix[from_node][to_node] = distance(x1, y1, x2, y2)
def Distance(self, from_node, to_node):
return int(self.matrix[from_node][to_node])
# Demand callback
class CreateDemandCallback(object):
"""Create callback to get demands at each location."""
def __init__(self, demands):
self.matrix = demands
def Demand(self, from_node, to_node):
return self.matrix[from_node]
def main():
# Create the data.
data = create_data_array()
locations = data[0]
demands = data[1]
num_locations = len(locations)
depot = 0 # The depot is the start and end point of each route.
num_vehicles = 1
# Create routing model.
if num_locations > 0:
routing = pywrapcp.RoutingModel(num_locations, num_vehicles, depot)
search_parameters = pywrapcp.RoutingModel.DefaultSearchParameters()
# Callback to the distance function.
dist_between_locations = CreateDistanceCallback(locations)
dist_callback = dist_between_locations.Distance
routing.SetArcCostEvaluatorOfAllVehicles(dist_callback)
# Put a callback to the demands.
demands_at_locations = CreateDemandCallback(demands)
demands_callback = demands_at_locations.Demand
# Add a dimension for demand.
slack_max = 0
vehicle_capacity = 1500
fix_start_cumul_to_zero = True
demand = "Demand"
routing.AddDimension(demands_callback, slack_max, vehicle_capacity,
fix_start_cumul_to_zero, demand)
# Solve, displays a solution if any.
assignment = routing.SolveWithParameters(search_parameters)
if assignment:
# Display solution.
# Solution cost.
print("Total distance of all routes: " + str(assignment.ObjectiveValue()) + "\n")
for vehicle_nbr in range(num_vehicles):
index = routing.Start(vehicle_nbr)
index_next = assignment.Value(routing.NextVar(index))
route = ''
route_dist = 0
route_demand = 0
while not routing.IsEnd(index_next):
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> "
# Add the distance to the next node.
route_dist += dist_callback(node_index, node_index_next)
# Add demand.
route_demand += demands[node_index_next]
index = index_next
index_next = assignment.Value(routing.NextVar(index))
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> " + str(node_index_next)
route_dist += dist_callback(node_index, node_index_next)
print("Route for vehicle " + str(vehicle_nbr) + ":\n\n" + route + "\n")
print("Distance of route " + str(vehicle_nbr) + ": " + str(route_dist))
print("Demand met by vehicle " + str(vehicle_nbr) + ": " + str(route_demand) + "\n")
else:
print('No solution found.')
else:
print('Specify an instance greater than 0.')
def create_data_array():
nodelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/f989e10af17fb4c85b11409fea47895b/raw/a3a8da0fa5b094f1ca9d82e1642b384889ae16e8/nodelist_sleeping_giant.csv')
locations = [[e.X, e.Y] for e in nodelist.itertuples()]
demands = [1] + [1] + [1] * 75
data = [locations, demands]
return data
if __name__ == '__main__':
main()
这输出一个解决方案：
Total distance of all routes: 12900
Route for vehicle 0:
0 -> 56 -> 40 -> 53 -> 63 -> 55 -> 14 -> 15 -> 12 -> 26 -> 34 -> 69 -> 36 -> 1 -> 64 -> 27 -> 48 -> 70 -> 47 -> 13 -> 10 -> 61 -> 45 -> 42 -> 60 -> 9 -> 8 -> 21 -> 43 -> 44 -> 3 -> 18 -> 58 -> 38 -> 28 -> 49 -> 32 -> 35 -> 50 -> 74 -> 46 -> 54 -> 76 -> 71 -> 65 -> 29 -> 16 -> 17 -> 22 -> 59 -> 7 -> 24 -> 31 -> 37 -> 67 -> 73 -> 41 -> 52 -> 75 -> 72 -> 20 -> 2 -> 39 -> 57 -> 23 -> 66 -> 5 -> 6 -> 30 -> 33 -> 68 -> 19 -> 25 -> 4 -> 11 -> 62 -> 51 -> 0
Distance of route 0: 12900
Demand met by vehicle 0: 76
所以正如你所看到的，我们正在无法穿越的节点之间旅行。
（乍一看）这看起来像是一些已经实现的专用求解器的包装器。也许改变内部结构并不那么容易。
但一个简单的解决方法：只需更改 distance-callback，这样它就会为您的禁止边缘带来非常高的成本。
这种方法需要调整，并不完美。但这是一个解决方法。
如果这有意义，取决于您的任务，它看起来非常非正式地描述。
我认为 Capacitized Vehicular Routing Problem 依赖于一个完整的图表，而您这里没有。没问题——我们可以用一个完整的图替换原始图，其中任意两个节点之间的距离是原始图上它们之间的最短距离。这很容易（O(n^3)) - 不那么容易）由 Floyd-Warshall 计算，我将在 networkx 中计算，因为我对它更熟悉。如果可用，请替换为 ortools 版本。所以 CreateDistanceCallback 现在看起来像
class CreateDistanceCallback(object):
'''Create callback to calculate distances between points.'''
def __init__(self, G):
'''Calculate shortest paths using Floyd-Warshall'''
self.paths = nx.floyd_warshall(G)
def Distance(self, from_node, to_node):
return self.paths[from_node][to_node]
它需要一个 nx.Graph() 对象，因此将 create_data_graph() 替换为
def create_data_graph():
edgelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/e570c38bcc72a8d102422f2af836513b/raw/89c76b2563dbc0e88384719a35cba0dfc04cd522/edgelist_sleeping_giant.csv')
nodelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/f989e10af17fb4c85b11409fea47895b/raw/a3a8da0fa5b094f1ca9d82e1642b384889ae16e8/nodelist_sleeping_giant.csv')
node_dict = dict(zip(nodelist['id'], list(range(nodelist.shape[0]))))
G = nx.Graph()
for i, elrow in edgelist.iterrows():
G.add_edge(node_dict[elrow.node1], node_dict[elrow.node2], weight=elrow.distance)
locations = [[e.X, e.Y] for e in nodelist.itertuples()]
demands = [1] + [1] + [1] * 75
return G, locations, demands
main() 中的第一行现已针对新的输入数据进行了调整
# Create the data.
G, locations, demands = create_data_graph()
并将CreateDistanceCallback对象的实例化替换为
dist_between_locations = CreateDistanceCallback(G)
当我运行时，我得到输出：
Total distance of all routes: 2
Route for vehicle 0:
0 -> 68 -> 75 -> 73 -> 76 -> 74 -> 71 -> 1 -> 8 -> 3 -> 9 -> 10 -> 12 -> 13 -> 14 -> 16 -> 15 -> 18 -> 21 -> 17 -> 22 -> 19 -> 20 -> 2 -> 4 -> 5 -> 6 -> 11 -> 72 -> 67 -> 69 -> 70 -> 65 -> 64 -> 63 -> 61 -> 60 -> 58 -> 50 -> 55 -> 59 -> 62 -> 66 -> 52 -> 39 -> 37 -> 56 -> 51 -> 57 -> 25 -> 33 -> 41 -> 31 -> 36 -> 54 -> 49 -> 48 -> 53 -> 47 -> 46 -> 45 -> 44 -> 43 -> 42 -> 35 -> 38 -> 34 -> 32 -> 29 -> 28 -> 27 -> 26 -> 24 -> 40 -> 7 -> 30 -> 23 -> 0
Distance of route 0: 49.440000000000005
Demand met by vehicle 0: 76
你能检查一下吗？
要按要求打印 "extended path"，重写 main() 的最后一部分：
ext_route = ''
last_node = None
while not routing.IsEnd(index_next):
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> "
if last_node is not None:
last_path = nx.dijkstra_path(G,last_node, node_index)
ext_route += repr(last_path) + " -> "
# Add the distance to the next node.
route_dist += dist_callback(node_index, node_index_next)
# Add demand.
route_demand += demands[node_index_next]
index = index_next
index_next = assignment.Value(routing.NextVar(index))
last_node = node_index
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> " + str(node_index_next)
route_dist += dist_callback(node_index, node_index_next)
print("Route for vehicle " + str(vehicle_nbr) + ":\n\n" + route + "\n")
print("Expanded route for vehicle " + str(vehicle_nbr) + ":\n\n" + ext_route + "\n")
print("Distance of route " + str(vehicle_nbr) + ": " + str(route_dist))
print("Demand met by vehicle " + str(vehicle_nbr) + ": " + str(route_demand) + "\n")
我想向 Charles Pehlivanian 提出一个替代解决方案。要禁止某些节点对之间的转换，您可以尝试使它们过于昂贵而无法出现在分配中。首先计算访问所有站点的最昂贵路线的上限。例如所有边缘之间的最大距离时间停止的次数：
penalty = len(locations) * max(distance(x,y) for x,y in edges)
然后将矩阵中禁止转换的值设置为该值：
if forbidden(from_node, to_node):
self.matrix[from_node][to_node] = penalty
else:
self.matrix[from_node][to_node] = distance(x1, y1, x2, y2)
我正在使用这个 python 库或工具：https://developers.google.com/optimization/routing/tsp/vehicle_routing （代码可以在这里找到）。
问题是，当您 运行 解决方案时，它会为您提供一条覆盖所有节点的路径。但是我的项目需要对节点之间的路径进行约束。例如，如果您在节点 {3} 上，则您可能无法前往节点 {18}。或者换句话说，如果您在节点 {5}，您只能前往节点 {1、12、14}。我不确定如何将此约束添加到当前代码示例中。
请允许我进一步解释...
如果我们看这张图：
您可能会在此处看到此图的表示形式： https://www.datacamp.com/community/tutorials/networkx-python-graph-tutorial
显然在这个问题中你不能从其他节点前往某些节点。我在 google or-tools 示例中使用此图中的数据来获取车辆路径问题的解决方案。
这是我的代码：
import math
from ortools.constraint_solver import pywrapcp
from ortools.constraint_solver import routing_enums_pb2
import pandas as pd
from sklearn import preprocessing
def distance(x1, y1, x2, y2):
dist = ((x1 - x2)**2 + (y1 - y2)**2)**(1/2)
return dist
class CreateDistanceCallback(object):
"""Create callback to calculate distances between points."""
def __init__(self, locations):
"""Initialize distance array."""
size = len(locations)
self.matrix = {}
for from_node in range(size):
self.matrix[from_node] = {}
for to_node in range(size):
x1 = locations[from_node][0]
y1 = locations[from_node][1]
x2 = locations[to_node][0]
y2 = locations[to_node][1]
self.matrix[from_node][to_node] = distance(x1, y1, x2, y2)
def Distance(self, from_node, to_node):
return int(self.matrix[from_node][to_node])
# Demand callback
class CreateDemandCallback(object):
"""Create callback to get demands at each location."""
def __init__(self, demands):
self.matrix = demands
def Demand(self, from_node, to_node):
return self.matrix[from_node]
def main():
# Create the data.
data = create_data_array()
locations = data[0]
demands = data[1]
num_locations = len(locations)
depot = 0 # The depot is the start and end point of each route.
num_vehicles = 1
# Create routing model.
if num_locations > 0:
routing = pywrapcp.RoutingModel(num_locations, num_vehicles, depot)
search_parameters = pywrapcp.RoutingModel.DefaultSearchParameters()
# Callback to the distance function.
dist_between_locations = CreateDistanceCallback(locations)
dist_callback = dist_between_locations.Distance
routing.SetArcCostEvaluatorOfAllVehicles(dist_callback)
# Put a callback to the demands.
demands_at_locations = CreateDemandCallback(demands)
demands_callback = demands_at_locations.Demand
# Add a dimension for demand.
slack_max = 0
vehicle_capacity = 1500
fix_start_cumul_to_zero = True
demand = "Demand"
routing.AddDimension(demands_callback, slack_max, vehicle_capacity,
fix_start_cumul_to_zero, demand)
# Solve, displays a solution if any.
assignment = routing.SolveWithParameters(search_parameters)
if assignment:
# Display solution.
# Solution cost.
print("Total distance of all routes: " + str(assignment.ObjectiveValue()) + "\n")
for vehicle_nbr in range(num_vehicles):
index = routing.Start(vehicle_nbr)
index_next = assignment.Value(routing.NextVar(index))
route = ''
route_dist = 0
route_demand = 0
while not routing.IsEnd(index_next):
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> "
# Add the distance to the next node.
route_dist += dist_callback(node_index, node_index_next)
# Add demand.
route_demand += demands[node_index_next]
index = index_next
index_next = assignment.Value(routing.NextVar(index))
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> " + str(node_index_next)
route_dist += dist_callback(node_index, node_index_next)
print("Route for vehicle " + str(vehicle_nbr) + ":\n\n" + route + "\n")
print("Distance of route " + str(vehicle_nbr) + ": " + str(route_dist))
print("Demand met by vehicle " + str(vehicle_nbr) + ": " + str(route_demand) + "\n")
else:
print('No solution found.')
else:
print('Specify an instance greater than 0.')
def create_data_array():
nodelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/f989e10af17fb4c85b11409fea47895b/raw/a3a8da0fa5b094f1ca9d82e1642b384889ae16e8/nodelist_sleeping_giant.csv')
locations = [[e.X, e.Y] for e in nodelist.itertuples()]
demands = [1] + [1] + [1] * 75
data = [locations, demands]
return data
if __name__ == '__main__':
main()
这输出一个解决方案：
Total distance of all routes: 12900
Route for vehicle 0:
0 -> 56 -> 40 -> 53 -> 63 -> 55 -> 14 -> 15 -> 12 -> 26 -> 34 -> 69 -> 36 -> 1 -> 64 -> 27 -> 48 -> 70 -> 47 -> 13 -> 10 -> 61 -> 45 -> 42 -> 60 -> 9 -> 8 -> 21 -> 43 -> 44 -> 3 -> 18 -> 58 -> 38 -> 28 -> 49 -> 32 -> 35 -> 50 -> 74 -> 46 -> 54 -> 76 -> 71 -> 65 -> 29 -> 16 -> 17 -> 22 -> 59 -> 7 -> 24 -> 31 -> 37 -> 67 -> 73 -> 41 -> 52 -> 75 -> 72 -> 20 -> 2 -> 39 -> 57 -> 23 -> 66 -> 5 -> 6 -> 30 -> 33 -> 68 -> 19 -> 25 -> 4 -> 11 -> 62 -> 51 -> 0
Distance of route 0: 12900
Demand met by vehicle 0: 76
所以正如你所看到的，我们正在无法穿越的节点之间旅行。
（乍一看）这看起来像是一些已经实现的专用求解器的包装器。也许改变内部结构并不那么容易。
但一个简单的解决方法：只需更改 distance-callback，这样它就会为您的禁止边缘带来非常高的成本。
这种方法需要调整，并不完美。但这是一个解决方法。
如果这有意义，取决于您的任务，它看起来非常非正式地描述。
我认为 Capacitized Vehicular Routing Problem 依赖于一个完整的图表，而您这里没有。没问题——我们可以用一个完整的图替换原始图，其中任意两个节点之间的距离是原始图上它们之间的最短距离。这很容易（O(n^3)) - 不那么容易）由 Floyd-Warshall 计算，我将在 networkx 中计算，因为我对它更熟悉。如果可用，请替换为 ortools 版本。所以 CreateDistanceCallback 现在看起来像
class CreateDistanceCallback(object):
'''Create callback to calculate distances between points.'''
def __init__(self, G):
'''Calculate shortest paths using Floyd-Warshall'''
self.paths = nx.floyd_warshall(G)
def Distance(self, from_node, to_node):
return self.paths[from_node][to_node]
它需要一个 nx.Graph() 对象，因此将 create_data_graph() 替换为
def create_data_graph():
edgelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/e570c38bcc72a8d102422f2af836513b/raw/89c76b2563dbc0e88384719a35cba0dfc04cd522/edgelist_sleeping_giant.csv')
nodelist = pd.read_csv('https://gist.githubusercontent.com/brooksandrew/f989e10af17fb4c85b11409fea47895b/raw/a3a8da0fa5b094f1ca9d82e1642b384889ae16e8/nodelist_sleeping_giant.csv')
node_dict = dict(zip(nodelist['id'], list(range(nodelist.shape[0]))))
G = nx.Graph()
for i, elrow in edgelist.iterrows():
G.add_edge(node_dict[elrow.node1], node_dict[elrow.node2], weight=elrow.distance)
locations = [[e.X, e.Y] for e in nodelist.itertuples()]
demands = [1] + [1] + [1] * 75
return G, locations, demands
main() 中的第一行现已针对新的输入数据进行了调整
# Create the data.
G, locations, demands = create_data_graph()
并将CreateDistanceCallback对象的实例化替换为
dist_between_locations = CreateDistanceCallback(G)
当我运行时，我得到输出：
Total distance of all routes: 2
Route for vehicle 0:
0 -> 68 -> 75 -> 73 -> 76 -> 74 -> 71 -> 1 -> 8 -> 3 -> 9 -> 10 -> 12 -> 13 -> 14 -> 16 -> 15 -> 18 -> 21 -> 17 -> 22 -> 19 -> 20 -> 2 -> 4 -> 5 -> 6 -> 11 -> 72 -> 67 -> 69 -> 70 -> 65 -> 64 -> 63 -> 61 -> 60 -> 58 -> 50 -> 55 -> 59 -> 62 -> 66 -> 52 -> 39 -> 37 -> 56 -> 51 -> 57 -> 25 -> 33 -> 41 -> 31 -> 36 -> 54 -> 49 -> 48 -> 53 -> 47 -> 46 -> 45 -> 44 -> 43 -> 42 -> 35 -> 38 -> 34 -> 32 -> 29 -> 28 -> 27 -> 26 -> 24 -> 40 -> 7 -> 30 -> 23 -> 0
Distance of route 0: 49.440000000000005
Demand met by vehicle 0: 76
你能检查一下吗？
要按要求打印 "extended path"，重写 main() 的最后一部分：
ext_route = ''
last_node = None
while not routing.IsEnd(index_next):
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> "
if last_node is not None:
last_path = nx.dijkstra_path(G,last_node, node_index)
ext_route += repr(last_path) + " -> "
# Add the distance to the next node.
route_dist += dist_callback(node_index, node_index_next)
# Add demand.
route_demand += demands[node_index_next]
index = index_next
index_next = assignment.Value(routing.NextVar(index))
last_node = node_index
node_index = routing.IndexToNode(index)
node_index_next = routing.IndexToNode(index_next)
route += str(node_index) + " -> " + str(node_index_next)
route_dist += dist_callback(node_index, node_index_next)
print("Route for vehicle " + str(vehicle_nbr) + ":\n\n" + route + "\n")
print("Expanded route for vehicle " + str(vehicle_nbr) + ":\n\n" + ext_route + "\n")
print("Distance of route " + str(vehicle_nbr) + ": " + str(route_dist))
print("Demand met by vehicle " + str(vehicle_nbr) + ": " + str(route_demand) + "\n")
我想向 Charles Pehlivanian 提出一个替代解决方案。要禁止某些节点对之间的转换，您可以尝试使它们过于昂贵而无法出现在分配中。首先计算访问所有站点的最昂贵路线的上限。例如所有边缘之间的最大距离时间停止的次数：
penalty = len(locations) * max(distance(x,y) for x,y in edges)
然后将矩阵中禁止转换的值设置为该值：
if forbidden(from_node, to_node):
self.matrix[from_node][to_node] = penalty
else:
self.matrix[from_node][to_node] = distance(x1, y1, x2, y2)