LPAstar_backup.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  1. """
  2. LPA_star 2D
  3. @author: huiming zhou
  4. """
  5. import os
  6. import sys
  7. import matplotlib.pyplot as plt
  8. sys.path.append(os.path.dirname(os.path.abspath(__file__)) +
  9. "/../../Search-based Planning/")
  10. from Search_2D import queue
  11. from Search_2D import plotting
  12. from Search_2D import env
  13. class LpaStar:
  14. def __init__(self, x_start, x_goal, heuristic_type):
  15. self.xI, self.xG = x_start, x_goal
  16. self.heuristic_type = heuristic_type
  17. self.Env = env.Env() # class Env
  18. self.u_set = self.Env.motions # feasible input set
  19. self.obs = self.Env.obs # position of obstacles
  20. self.OPEN = queue.QueuePrior() # priority queue / U set
  21. self.g, self.v = {}, {}
  22. for i in range(self.Env.x_range):
  23. for j in range(self.Env.y_range):
  24. self.v[(i, j)] = float("inf")
  25. self.g[(i, j)] = float("inf")
  26. self.v[self.xI] = 0
  27. self.OPEN.put(self.xI, self.Key(self.xI))
  28. self.CLOSED = set()
  29. def searching(self):
  30. self.ComputePath()
  31. path = [self.extract_path()]
  32. # self.print_g()
  33. obs_change = set()
  34. for i in range(25, 30):
  35. self.obs.add((i, 15))
  36. obs_change.add((i, 15))
  37. self.obs.add((30, 14))
  38. obs_change.add((30, 14))
  39. for s in obs_change:
  40. self.v[s] = float("inf")
  41. self.g[s] = float("inf")
  42. for x in self.get_neighbor(s):
  43. self.UpdateMembership(x)
  44. # for x in obs_change:
  45. # self.obs.remove(x)
  46. # for x in obs_change:
  47. # self.UpdateVertex(x)
  48. self.ComputePath()
  49. path.append(self.extract_path_test())
  50. self.print_g()
  51. return path, obs_change
  52. def ComputePath(self):
  53. while self.Key(self.xG) > self.OPEN.top_key() \
  54. or self.v[self.xG] < self.g[self.xG]:
  55. s = self.OPEN.get()
  56. if self.v[s] > self.g[s]:
  57. self.v[s] = self.g[s]
  58. self.CLOSED.add(s)
  59. while self.OPEN.top_key() < self.Key(self.xG) \
  60. or self.v[self.xG] != self.g[self.xG]:
  61. s = self.OPEN.get()
  62. if self.g[s] > self.v[s]:
  63. self.g[s] = self.v[s]
  64. else:
  65. self.g[s] = float("inf")
  66. self.UpdateMembership(s)
  67. for x in self.get_neighbor(s):
  68. self.UpdateMembership(x)
  69. # return self.extract_path()
  70. def UpdateMembership(self, s):
  71. if self.v[s] != self.g[s]:
  72. if s not in self.CLOSED:
  73. self.OPEN.put(s, self.Key(s))
  74. else:
  75. if s in self.OPEN:
  76. self.OPEN.remove(s)
  77. def print_g(self):
  78. print("he")
  79. for k in range(self.Env.y_range):
  80. j = self.Env.y_range - k - 1
  81. string = ""
  82. for i in range(self.Env.x_range):
  83. if self.g[(i, j)] == float("inf"):
  84. string += ("00" + ', ')
  85. else:
  86. if self.g[(i, j)] // 10 == 0:
  87. string += ("0" + str(self.g[(i, j)]) + ', ')
  88. else:
  89. string += (str(self.g[(i, j)]) + ', ')
  90. print(string)
  91. def extract_path(self):
  92. path = []
  93. s = self.xG
  94. while True:
  95. g_list = {}
  96. for x in self.get_neighbor(s):
  97. g_list[x] = self.g[x]
  98. s = min(g_list, key=g_list.get)
  99. if s == self.xI:
  100. return list(reversed(path))
  101. path.append(s)
  102. def extract_path_test(self):
  103. path = []
  104. s = self.xG
  105. for k in range(70):
  106. g_list = {}
  107. for x in self.get_neighbor(s):
  108. g_list[x] = self.g[x]
  109. s = min(g_list, key=g_list.get)
  110. if s == self.xI:
  111. return list(reversed(path))
  112. path.append(s)
  113. return list(reversed(path))
  114. def get_neighbor(self, s):
  115. nei_list = set()
  116. for u in self.u_set:
  117. s_next = tuple([s[i] + u[i] for i in range(2)])
  118. if s_next not in self.obs:
  119. nei_list.add(s_next)
  120. return nei_list
  121. def Key(self, s):
  122. return [min(self.g[s], self.v[s]) + self.h(s),
  123. min(self.g[s], self.v[s])]
  124. def h(self, s):
  125. heuristic_type = self.heuristic_type # heuristic type
  126. goal = self.xG # goal node
  127. if heuristic_type == "manhattan":
  128. return abs(goal[0] - s[0]) + abs(goal[1] - s[1])
  129. elif heuristic_type == "euclidean":
  130. return ((goal[0] - s[0]) ** 2 + (goal[1] - s[1]) ** 2) ** (1 / 2)
  131. else:
  132. print("Please choose right heuristic type!")
  133. def get_cost(self, s_start, s_end):
  134. """
  135. Calculate cost for this motion
  136. :param s_start:
  137. :param s_end:
  138. :return: cost for this motion
  139. :note: cost function could be more complicate!
  140. """
  141. # if s_start not in self.obs:
  142. # if s_end not in self.obs:
  143. # return 1
  144. # else:
  145. # return float("inf")
  146. # return float("inf")
  147. return 1
  148. def main():
  149. x_start = (5, 5)
  150. x_goal = (45, 25)
  151. lpastar = LpaStar(x_start, x_goal, "manhattan")
  152. plot = plotting.Plotting(x_start, x_goal)
  153. path, obs = lpastar.searching()
  154. plot.plot_grid("Lifelong Planning A*")
  155. p = path[0]
  156. px = [x[0] for x in p]
  157. py = [x[1] for x in p]
  158. plt.plot(px, py, marker='o')
  159. plt.pause(0.5)
  160. p = path[1]
  161. px = [x[0] for x in p]
  162. py = [x[1] for x in p]
  163. plt.plot(px, py, marker='o')
  164. plt.pause(0.01)
  165. plt.show()
  166. if __name__ == '__main__':
  167. main()