a_star.py 4.2 KB

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