diff options
Diffstat (limited to 'ant.py')
| -rw-r--r-- | ant.py | 262 |
1 files changed, 164 insertions, 98 deletions
@@ -5,6 +5,26 @@ from threading import Event class Ant: + """Instancia de agente en Ant Colony System. Paralelizable. + + Attributes + ---------- + graph : VrptwGraph + VRPTW problem instance (parameters + graph + pheromones) + current_index : int + Current route's vehicle's location + vehicle_load : float + Current route's vehicle's spare capacity + vehicle_travel_time : float + Current route's vehicle's travel time + travel_path : list + Representation of the solution. A sequence of node indeces. + arrival_time : list + ? + index_to_visit : list + Remaining node indices not yet included in travel_path + total_travel_distance : float + """ def __init__(self, graph: VrptwGraph, start_index=0): super() self.graph = graph @@ -20,28 +40,31 @@ class Ant: self.total_travel_distance = 0 def clear(self): + """Restart Ant solution (Kill)""" self.travel_path.clear() self.index_to_visit.clear() def move_to_next_index(self, next_index): - # 更新蚂蚁路径 + """Update ant path + "current simulated vehicle route parameters".""" self.travel_path.append(next_index) - self.total_travel_distance += self.graph.node_dist_mat[self.current_index][next_index] - dist = self.graph.node_dist_mat[self.current_index][next_index] - self.arrival_time.append(self.vehicle_travel_time + dist) + self.total_travel_distance += dist + + self.arrival_time.append(self.vehicle_travel_time + dist * 1) if self.graph.nodes[next_index].is_depot: - # 如果一下个位置为服务器点，则要将车辆负载等清空 + # If the next location is a depot, the vehicle load, etc., must be cleared + # (start next vehicle route) self.vehicle_load = 0 self.vehicle_travel_time = 0 else: - # 更新车辆负载、行驶距离、时间 + # Update vehicle load, distance traveled, time self.vehicle_load += self.graph.nodes[next_index].demand - # 如果早于客户要求的时间窗(ready_time)，则需要等待 - - self.vehicle_travel_time += dist + max(self.graph.nodes[next_index].ready_time - self.vehicle_travel_time - dist, 0) + self.graph.nodes[next_index].service_time + # If it is earlier than the time window (ready_time) requested by the client, you need to wait + self.vehicle_travel_time += dist * 1 \ + + max(self.graph.nodes[next_index].ready_time - self.vehicle_travel_time - dist * 1, 0) \ + + self.graph.nodes[next_index].service_time self.index_to_visit.remove(next_index) self.current_index = next_index @@ -53,75 +76,102 @@ class Ant: return self.travel_path.count(0)-1 def check_condition(self, next_index) -> bool: - """ - 检查移动到下一个点是否满足约束条件 - :param next_index: - :return: + """Check if moving to next point satisfies constraints + + Returns + ------- + bool + Conditions satisfied. """ if self.vehicle_load + self.graph.nodes[next_index].demand > self.graph.vehicle_capacity: + # Capacity is not enough to serve whole customer demand return False - dist = self.graph.node_dist_mat[self.current_index][next_index] - wait_time = max(self.graph.nodes[next_index].ready_time - self.vehicle_travel_time - dist, 0) + # Time that takes traversing the arc ("tramo") (vehicle mean speed may vary) + dist_time = self.graph.node_dist_mat[self.current_index][next_index] * 1 + wait_time = max(self.graph.nodes[next_index].ready_time - self.vehicle_travel_time - dist_time, 0) service_time = self.graph.nodes[next_index].service_time - # 检查访问某一个旅客之后，能否回到服务店 - if self.vehicle_travel_time + dist + wait_time + service_time + self.graph.node_dist_mat[next_index][0] > self.graph.nodes[0].due_time: + # Check whether a traveler can return to the service shop after visiting a certain traveler + if self.vehicle_travel_time + dist_time + wait_time + service_time \ + + self.graph.node_dist_mat[next_index][0] > self.graph.nodes[0].due_time: return False - # 不可以服务due time之外的旅客 - if self.vehicle_travel_time + dist > self.graph.nodes[next_index].due_time: + # Passengers outside the due time cannot be served + if self.vehicle_travel_time + dist_time > self.graph.nodes[next_index].due_time: return False return True def cal_next_index_meet_constrains(self): - """ - 找出所有从当前位置（ant.current_index）可达的customer - :return: + """Find all customers reachable from the current location (ant.current_index) + + Returns + ------- + list + Neighbor nodes' indices that satisfy constraints """ next_index_meet_constrains = [] - for next_ind in self.index_to_visit: - if self.check_condition(next_ind): - next_index_meet_constrains.append(next_ind) + for next_index in self.index_to_visit: + if self.check_condition(next_index): + next_index_meet_constrains.append(next_index) return next_index_meet_constrains def cal_nearest_next_index(self, next_index_list): """ - 从待选的customers中选择，离当前位置（ant.current_index）最近的customer + Select from the available customers, the customer closest to the current + location (ant.current_index) - :param next_index_list: - :return: + Parameters + ---------- + next_index_list : list + ??? + + Returns + ------- + int """ - current_ind = self.current_index + current_index = self.current_index - nearest_ind = next_index_list[0] - min_dist = self.graph.node_dist_mat[current_ind][next_index_list[0]] + nearest_index = next_index_list[0] + min_dist = self.graph.node_dist_mat[current_index][nearest_index] - for next_ind in next_index_list[1:]: - dist = self.graph.node_dist_mat[current_ind][next_ind] + for next_index in next_index_list[1:]: + dist = self.graph.node_dist_mat[current_index][next_index] if dist < min_dist: min_dist = dist - nearest_ind = next_ind + nearest_index = next_index - return nearest_ind + return nearest_index @staticmethod def cal_total_travel_distance(graph: VrptwGraph, travel_path): + """ + (Update to ignore depot-to-depot distances later) + """ distance = 0 - current_ind = travel_path[0] - for next_ind in travel_path[1:]: - distance += graph.node_dist_mat[current_ind][next_ind] - current_ind = next_ind + current_index = travel_path[0] + for next_index in travel_path[1:]: + distance += graph.node_dist_mat[current_index][next_index] + current_index = next_index return distance def try_insert_on_path(self, node_id, stop_event: Event): """ - 尝试性地将node_id插入当前的travel_path中 - 插入的位置不能违反载重，时间，行驶距离的限制 - 如果有多个位置，则找出最优的位置 - :param node_id: - :return: + Attempt to insert node_id into current travel_path. Inserted locations + cannot violate load, time, travel distance restrictions. If there are + multiple positions, find the best position. + + Parameters + ---------- + + Returns + ------- + + Note + ---- + Seems like at least one vehicle route (depot-to-depot substring) in + travel_path is required. This is OK since """ best_insert_index = None best_distance = None @@ -135,85 +185,92 @@ class Ant: if self.graph.nodes[self.travel_path[insert_index]].is_depot: continue - # 找出insert_index的前面的最近的depot - front_depot_index = insert_index - while front_depot_index >= 0 and not self.graph.nodes[self.travel_path[front_depot_index]].is_depot: - front_depot_index -= 1 - front_depot_index = max(front_depot_index, 0) + # Find the nearest depot before insert_index + last_depot_index = insert_index + while last_depot_index > 0 and not self.graph.nodes[self.travel_path[last_depot_index]].is_depot: + last_depot_index -= 1 - # check_ant从front_depot_index出发 - check_ant = Ant(self.graph, self.travel_path[front_depot_index]) + # check_ant starts from last_depot_index + check_ant = Ant(self.graph, self.travel_path[last_depot_index]) - # 让check_ant 走过 path中下标从front_depot_index开始到insert_index-1的点 - for i in range(front_depot_index+1, insert_index): + # Let check_ant walk through the point in the path from + # last_depot_index to insert_index-1 + for i in range(last_depot_index+1, insert_index): + # (only one depot, so no start point needs to be passed) check_ant.move_to_next_index(self.travel_path[i]) - # 开始尝试性地对排序后的index_to_visit中的结点进行访问 + # Begin to tentatively visit the nodes in the sorted index_to_visit if check_ant.check_condition(node_id): check_ant.move_to_next_index(node_id) else: continue - # 如果可以到node_id，则要保证vehicle可以行驶回到depot - for next_ind in self.travel_path[insert_index:]: - + # If you can get to the node_id, make sure that the vehicle can drive back to the depot + for next_index in self.travel_path[insert_index:]: if stop_event.is_set(): # print('[try_insert_on_path]: receive stop event') return - if check_ant.check_condition(next_ind): - check_ant.move_to_next_index(next_ind) + if check_ant.check_condition(next_index): + check_ant.move_to_next_index(next_index) - # 如果回到了depot - if self.graph.nodes[next_ind].is_depot: - temp_front_index = self.travel_path[insert_index-1] - temp_back_index = self.travel_path[insert_index] + # If back to the depot, then it means the vehicle whose + # route would be modified is over *and valid*. So now, + # compute the new total_travel_distance if node_id where + # added just before travel_path[insert_index] + if self.graph.nodes[next_index].is_depot: + insert_node_id_prev = self.travel_path[insert_index-1] + insert_node_id = self.travel_path[insert_index] - check_ant_distance = self.total_travel_distance - self.graph.node_dist_mat[temp_front_index][temp_back_index] + \ - self.graph.node_dist_mat[temp_front_index][node_id] + self.graph.node_dist_mat[node_id][temp_back_index] + check_ant_distance = self.total_travel_distance \ + - self.graph.node_dist_mat[insert_node_id_prev][insert_node_id] \ + + self.graph.node_dist_mat[insert_node_id_prev][node_id] \ + + self.graph.node_dist_mat[node_id][insert_node_id] if best_distance is None or check_ant_distance < best_distance: best_distance = check_ant_distance best_insert_index = insert_index break - # 如果不可以回到depot，则返回上一层 + # If you can't go back to the depot, simmulated route was invalid else: break return best_insert_index - def insertion_procedure(self, stop_even: Event): + def insertion_procedure(self, stop_event: Event): """ - 为每个未访问的结点尝试性地找到一个合适的位置，插入到当前的travel_path - 插入的位置不能违反载重，时间，行驶距离的限制 + Try to find a suitable location for each unvisited node, insert into the + current travel_path and insert the location without violating load, + time, travel distance constraints :return: """ if self.index_to_visit_empty(): return success_to_insert = True - # 直到未访问的结点中没有一个结点可以插入成功 + # Until none of the unvisited nodes can be inserted successfully while success_to_insert: success_to_insert = False - # 获取未访问的结点 + # Get unvisited nodes ind_to_visit = np.array(copy.deepcopy(self.index_to_visit)) - # 获取为访问客户点的demand，降序排序 + # Sort ind_to_visit by customer demand in descending order demand = np.zeros(len(ind_to_visit)) for i, ind in zip(range(len(ind_to_visit)), ind_to_visit): demand[i] = self.graph.nodes[ind].demand - arg_ind = np.argsort(demand)[::-1] - ind_to_visit = ind_to_visit[arg_ind] + arg_index = np.argsort(demand)[::-1] + ind_to_visit = ind_to_visit[arg_index] + # Try to insert at least one from the ordered list for node_id in ind_to_visit: - if stop_even.is_set(): + if stop_event.is_set(): # print('[insertion_procedure]: receive stop event') return - best_insert_index = self.try_insert_on_path(node_id, stop_even) + best_insert_index = self.try_insert_on_path(node_id, stop_event) if best_insert_index is not None: self.travel_path.insert(best_insert_index, node_id) self.index_to_visit.remove(node_id) @@ -229,26 +286,36 @@ class Ant: @staticmethod def local_search_once(graph: VrptwGraph, travel_path: list, travel_distance: float, i_start, stop_event: Event): + """ - # 找出path中所有的depot的位置 - depot_ind = [] + Returns + ------- + list + None, None, None + """ + + # Find the locations of all depots in the path + depot_index = [] for ind in range(len(travel_path)): if graph.nodes[travel_path[ind]].is_depot: - depot_ind.append(ind) + depot_index.append(ind) - # 将self.travel_path分成多段，每段以depot开始，以depot结束，称为route - for i in range(i_start, len(depot_ind)): - for j in range(i + 1, len(depot_ind)): + # Divide self.travel_path into multiple segments, each segment starts + # with depot and ends with depot, called route + for i in range(i_start, len(depot_index)): + for j in range(i + 1, len(depot_index)): if stop_event.is_set(): return None, None, None - for start_a in range(depot_ind[i - 1] + 1, depot_ind[i]): - for end_a in range(start_a, min(depot_ind[i], start_a + 6)): - for start_b in range(depot_ind[j - 1] + 1, depot_ind[j]): - for end_b in range(start_b, min(depot_ind[j], start_b + 6)): + for start_a in range(depot_index[i - 1] + 1, depot_index[i]): + for end_a in range(start_a, min(depot_index[i], start_a + 6)): + for start_b in range(depot_index[j - 1] + 1, depot_index[j]): + for end_b in range(start_b, min(depot_index[j], start_b + 6)): if start_a == end_a and start_b == end_b: continue + # Create new travel_path changing swaping routes + # a and b new_path = [] new_path.extend(travel_path[:start_a]) new_path.extend(travel_path[start_b:end_b + 1]) @@ -256,13 +323,13 @@ class Ant: new_path.extend(travel_path[start_a:end_a]) new_path.extend(travel_path[end_b + 1:]) - depot_before_start_a = depot_ind[i - 1] + depot_before_start_a = depot_index[i - 1] - depot_before_start_b = depot_ind[j - 1] + (end_b - start_b) - (end_a - start_a) + 1 + depot_before_start_b = depot_index[j - 1] + (end_b - start_b) - (end_a - start_a) + 1 if not graph.nodes[new_path[depot_before_start_b]].is_depot: raise RuntimeError('error') - # 判断发生改变的route a是否是feasible的 + # Determine whether the changed route a is feasible success_route_a = False check_ant = Ant(graph, new_path[depot_before_start_a]) for ind in new_path[depot_before_start_a + 1:]: @@ -277,7 +344,7 @@ class Ant: check_ant.clear() del check_ant - # 判断发生改变的route b是否是feasible的 + # Determine whether the changed route b is feasible success_route_b = False check_ant = Ant(graph, new_path[depot_before_start_b]) for ind in new_path[depot_before_start_b + 1:]: @@ -296,10 +363,10 @@ class Ant: if new_path_distance < travel_distance: # print('success to search') - # 删除路径中连在一起的depot中的一个 - for temp_ind in range(1, len(new_path)): - if graph.nodes[new_path[temp_ind]].is_depot and graph.nodes[new_path[temp_ind - 1]].is_depot: - new_path.pop(temp_ind) + # Delete one of the depots linked together in the path + for temp_index in range(1, len(new_path)): + if graph.nodes[new_path[temp_index]].is_depot and graph.nodes[new_path[temp_index - 1]].is_depot: + new_path.pop(temp_index) break return new_path, new_path_distance, i else: @@ -309,8 +376,7 @@ class Ant: def local_search_procedure(self, stop_event: Event): """ - 对当前的已经访问完graph中所有节点的travel_path使用cross进行局部搜索 - :return: + Use cross to perform a local search on the current travel_path that has visited all nodes in the graph """ new_path = copy.deepcopy(self.travel_path) new_path_distance = self.total_travel_distance @@ -326,7 +392,7 @@ class Ant: new_path = temp_path new_path_distance = temp_distance - # 设置i_start + # set i_start i_start = (i_start + 1) % (new_path.count(0)-1) i_start = max(i_start, 1) else: @@ -334,4 +400,4 @@ class Ant: self.travel_path = new_path self.total_travel_distance = new_path_distance - print('[local_search_procedure]: local search finished')
\ No newline at end of file + print('[local_search_procedure]: local search finished') |