""" Bidirectional_a_star 2D @author: huiming zhou """ import os import sys import math import heapq sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../../Search_based_Planning/") from Search_2D import plotting, env class BidirectionalAStar: def __init__(self, s_start, s_goal, heuristic_type): self.s_start = s_start self.s_goal = s_goal self.heuristic_type = heuristic_type self.Env = env.Env() # class Env self.u_set = self.Env.motions # feasible input set self.obs = self.Env.obs # position of obstacles self.OPEN_fore = [] # OPEN set for forward searching self.OPEN_back = [] # OPEN set for backward searching self.CLOSED_fore = [] # CLOSED set for forward self.CLOSED_back = [] # CLOSED set for backward self.PARENT_fore = dict() # recorded parent for forward self.PARENT_back = dict() # recorded parent for backward self.g_fore = dict() # cost to come for forward self.g_back = dict() # cost to come for backward def init(self): """ initialize parameters """ self.g_fore[self.s_start] = 0.0 self.g_fore[self.s_goal] = math.inf self.g_back[self.s_goal] = 0.0 self.g_back[self.s_start] = math.inf self.PARENT_fore[self.s_start] = self.s_start self.PARENT_back[self.s_goal] = self.s_goal heapq.heappush(self.OPEN_fore, (self.f_value_fore(self.s_start), self.s_start)) heapq.heappush(self.OPEN_back, (self.f_value_back(self.s_goal), self.s_goal)) def searching(self): """ Bidirectional A* :return: connected path, visited order of forward, visited order of backward """ self.init() s_meet = self.s_start while self.OPEN_fore and self.OPEN_back: # solve foreward-search _, s_fore = heapq.heappop(self.OPEN_fore) if s_fore in self.PARENT_back: s_meet = s_fore break self.CLOSED_fore.append(s_fore) for s_n in self.get_neighbor(s_fore): new_cost = self.g_fore[s_fore] + self.cost(s_fore, s_n) if s_n not in self.g_fore: self.g_fore[s_n] = math.inf if new_cost < self.g_fore[s_n]: self.g_fore[s_n] = new_cost self.PARENT_fore[s_n] = s_fore heapq.heappush(self.OPEN_fore, (self.f_value_fore(s_n), s_n)) # solve backward-search _, s_back = heapq.heappop(self.OPEN_back) if s_back in self.PARENT_fore: s_meet = s_back break self.CLOSED_back.append(s_back) for s_n in self.get_neighbor(s_back): new_cost = self.g_back[s_back] + self.cost(s_back, s_n) if s_n not in self.g_back: self.g_back[s_n] = math.inf if new_cost < self.g_back[s_n]: self.g_back[s_n] = new_cost self.PARENT_back[s_n] = s_back heapq.heappush(self.OPEN_back, (self.f_value_back(s_n), s_n)) return self.extract_path(s_meet), self.CLOSED_fore, self.CLOSED_back def get_neighbor(self, s): """ find neighbors of state s that not in obstacles. :param s: state :return: neighbors """ return [(s[0] + u[0], s[1] + u[1]) for u in self.u_set] def extract_path(self, s_meet): """ extract path from start and goal :param s_meet: meet point of bi-direction a* :return: path """ # extract path for foreward part path_fore = [s_meet] s = s_meet while True: s = self.PARENT_fore[s] path_fore.append(s) if s == self.s_start: break # extract path for backward part path_back = [] s = s_meet while True: s = self.PARENT_back[s] path_back.append(s) if s == self.s_goal: break return list(reversed(path_fore)) + list(path_back) def f_value_fore(self, s): """ forward searching: f = g + h. (g: Cost to come, h: heuristic value) :param s: current state :return: f """ return self.g_fore[s] + self.h(s, self.s_goal) def f_value_back(self, s): """ backward searching: f = g + h. (g: Cost to come, h: heuristic value) :param s: current state :return: f """ return self.g_back[s] + self.h(s, self.s_start) def h(self, s, goal): """ Calculate heuristic value. :param s: current node (state) :param goal: goal node (state) :return: heuristic value """ heuristic_type = self.heuristic_type if heuristic_type == "manhattan": return abs(goal[0] - s[0]) + abs(goal[1] - s[1]) else: return math.hypot(goal[0] - s[0], goal[1] - s[1]) def cost(self, s_start, s_goal): """ Calculate Cost for this motion :param s_start: starting node :param s_goal: end node :return: Cost for this motion :note: Cost function could be more complicate! """ if self.is_collision(s_start, s_goal): return math.inf return math.hypot(s_goal[0] - s_start[0], s_goal[1] - s_start[1]) def is_collision(self, s_start, s_end): """ check if the line segment (s_start, s_end) is collision. :param s_start: start node :param s_end: end node :return: True: is collision / False: not collision """ if s_start in self.obs or s_end in self.obs: return True if s_start[0] != s_end[0] and s_start[1] != s_end[1]: if s_end[0] - s_start[0] == s_start[1] - s_end[1]: s1 = (min(s_start[0], s_end[0]), min(s_start[1], s_end[1])) s2 = (max(s_start[0], s_end[0]), max(s_start[1], s_end[1])) else: s1 = (min(s_start[0], s_end[0]), max(s_start[1], s_end[1])) s2 = (max(s_start[0], s_end[0]), min(s_start[1], s_end[1])) if s1 in self.obs or s2 in self.obs: return True return False def main(): x_start = (5, 5) x_goal = (45, 25) bastar = BidirectionalAStar(x_start, x_goal, "euclidean") plot = plotting.Plotting(x_start, x_goal) path, visited_fore, visited_back = bastar.searching() plot.animation_bi_astar(path, visited_fore, visited_back, "Bidirectional-A*") # animation if __name__ == '__main__': main()