a_star.py 3.8 KB

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