a_star.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. import queue
  2. import plotting
  3. import env
  4. class Astar:
  5. def __init__(self, x_start, x_goal, heuristic_type):
  6. self.xI, self.xG = x_start, x_goal
  7. self.Env = env.Env() # class Env
  8. self.plotting = plotting.Plotting(self.xI, self.xG) # class Plotting
  9. self.u_set = self.Env.motions # feasible input set
  10. self.obs = self.Env.obs # position of obstacles
  11. [self.path, self.policy, self.visited] = self.searching(self.xI, self.xG, heuristic_type)
  12. self.fig_name = "A* Algorithm"
  13. self.plotting.animation(self.path, self.visited, self.fig_name) # animation generate
  14. def searching(self, xI, xG, heuristic_type):
  15. """
  16. Searching using A_star.
  17. :return: planning path, action in each node, visited nodes in the planning process
  18. """
  19. q_astar = queue.QueuePrior() # priority queue
  20. q_astar.put(xI, 0)
  21. parent = {xI: xI} # record parents of nodes
  22. action = {xI: (0, 0)} # record actions of nodes
  23. visited = []
  24. cost = {xI: 0}
  25. while not q_astar.empty():
  26. x_current = q_astar.get()
  27. if x_current == xG: # stop condition
  28. break
  29. visited.append(x_current)
  30. for u_next in self.u_set: # explore neighborhoods of current node
  31. x_next = tuple([x_current[i] + u_next[i] for i in range(len(x_current))])
  32. if x_next not in self.obs:
  33. new_cost = cost[x_current] + self.get_cost(x_current, u_next)
  34. if x_next not in cost or new_cost < cost[x_next]: # conditions for updating cost
  35. cost[x_next] = new_cost
  36. priority = new_cost + self.Heuristic(x_next, xG, heuristic_type)
  37. q_astar.put(x_next, priority) # put node into queue using priority "f+h"
  38. parent[x_next], action[x_next] = x_current, u_next
  39. [path, policy] = self.extract_path(xI, xG, parent, action)
  40. return path, policy, visited
  41. def extract_path(self, xI, xG, parent, policy):
  42. """
  43. Extract the path based on the relationship of nodes.
  44. :param xI: Starting node
  45. :param xG: Goal node
  46. :param parent: Relationship between nodes
  47. :param policy: Action needed for transfer between two nodes
  48. :return: The planning path
  49. """
  50. path_back = [xG]
  51. acts_back = [policy[xG]]
  52. x_current = xG
  53. while True:
  54. x_current = parent[x_current]
  55. path_back.append(x_current)
  56. acts_back.append(policy[x_current])
  57. if x_current == xI: break
  58. return list(path_back), list(acts_back)
  59. def get_cost(self, x, u):
  60. """
  61. Calculate cost for this motion
  62. :param x: current node
  63. :param u: input
  64. :return: cost for this motion
  65. :note: cost function could be more complicate!
  66. """
  67. return 1
  68. def Heuristic(self, state, goal, heuristic_type):
  69. """
  70. Calculate heuristic.
  71. :param state: current node (state)
  72. :param goal: goal node (state)
  73. :param heuristic_type: choosing different heuristic functions
  74. :return: heuristic
  75. """
  76. if heuristic_type == "manhattan":
  77. return abs(goal[0] - state[0]) + abs(goal[1] - state[1])
  78. elif heuristic_type == "euclidean":
  79. return ((goal[0] - state[0]) ** 2 + (goal[1] - state[1]) ** 2) ** (1 / 2)
  80. else:
  81. print("Please choose right heuristic type!")
  82. if __name__ == '__main__':
  83. x_Start = (5, 5) # Starting node
  84. x_Goal = (49, 5) # Goal node
  85. astar = Astar(x_Start, x_Goal, "manhattan")