a_star.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  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. @staticmethod
  42. def extract_path(xI, xG, parent, policy):
  43. """
  44. Extract the path based on the relationship of nodes.
  45. :param xI: Starting node
  46. :param xG: Goal node
  47. :param parent: Relationship between nodes
  48. :param policy: Action needed for transfer between two nodes
  49. :return: The planning path
  50. """
  51. path_back = [xG]
  52. acts_back = [policy[xG]]
  53. x_current = xG
  54. while True:
  55. x_current = parent[x_current]
  56. path_back.append(x_current)
  57. acts_back.append(policy[x_current])
  58. if x_current == xI:
  59. break
  60. return list(path_back), list(acts_back)
  61. @staticmethod
  62. def get_cost(x, u):
  63. """
  64. Calculate cost for this motion
  65. :param x: current node
  66. :param u: input
  67. :return: cost for this motion
  68. :note: cost function could be more complicate!
  69. """
  70. return 1
  71. @staticmethod
  72. def Heuristic(state, goal, heuristic_type):
  73. """
  74. Calculate heuristic.
  75. :param state: current node (state)
  76. :param goal: goal node (state)
  77. :param heuristic_type: choosing different heuristic functions
  78. :return: heuristic
  79. """
  80. if heuristic_type == "manhattan":
  81. return abs(goal[0] - state[0]) + abs(goal[1] - state[1])
  82. elif heuristic_type == "euclidean":
  83. return ((goal[0] - state[0]) ** 2 + (goal[1] - state[1]) ** 2) ** (1 / 2)
  84. else:
  85. print("Please choose right heuristic type!")
  86. if __name__ == '__main__':
  87. x_Start = (5, 5) # Starting node
  88. x_Goal = (49, 5) # Goal node
  89. astar = Astar(x_Start, x_Goal, "manhattan")