|
@@ -5,6 +5,7 @@ LPA_star 2D
|
|
|
|
|
|
|
|
import os
|
|
import os
|
|
|
import sys
|
|
import sys
|
|
|
|
|
+import math
|
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.pyplot as plt
|
|
|
|
|
|
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)) +
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)) +
|
|
@@ -21,11 +22,14 @@ class LpaStar:
|
|
|
self.heuristic_type = heuristic_type
|
|
self.heuristic_type = heuristic_type
|
|
|
|
|
|
|
|
self.Env = env.Env() # class Env
|
|
self.Env = env.Env() # class Env
|
|
|
|
|
+ self.Plot = plotting.Plotting(x_start, x_goal)
|
|
|
|
|
|
|
|
self.u_set = self.Env.motions # feasible input set
|
|
self.u_set = self.Env.motions # feasible input set
|
|
|
self.obs = self.Env.obs # position of obstacles
|
|
self.obs = self.Env.obs # position of obstacles
|
|
|
|
|
+ self.x = self.Env.x_range
|
|
|
|
|
+ self.y = self.Env.y_range
|
|
|
|
|
|
|
|
- self.U = queue.QueuePrior() # priority queue / OPEN set
|
|
|
|
|
|
|
+ self.U = queue.QueuePrior() # priority queue / U set
|
|
|
self.g, self.rhs = {}, {}
|
|
self.g, self.rhs = {}, {}
|
|
|
|
|
|
|
|
for i in range(self.Env.x_range):
|
|
for i in range(self.Env.x_range):
|
|
@@ -34,51 +38,102 @@ class LpaStar:
|
|
|
self.g[(i, j)] = float("inf")
|
|
self.g[(i, j)] = float("inf")
|
|
|
|
|
|
|
|
self.rhs[self.xI] = 0
|
|
self.rhs[self.xI] = 0
|
|
|
- self.U.put(self.xI, self.CalculateKey(self.xI))
|
|
|
|
|
|
|
+ self.U.put(self.xI, self.Key(self.xI))
|
|
|
|
|
|
|
|
def searching(self):
|
|
def searching(self):
|
|
|
- self.computePath()
|
|
|
|
|
- path = [self.extract_path()]
|
|
|
|
|
-
|
|
|
|
|
- obs_change = set()
|
|
|
|
|
- for j in range(14, 15):
|
|
|
|
|
- self.obs.add((30, j))
|
|
|
|
|
- obs_change.add((30, j))
|
|
|
|
|
- for s in obs_change:
|
|
|
|
|
- self.rhs[s] = float("inf")
|
|
|
|
|
- self.g[s] = float("inf")
|
|
|
|
|
- for x in self.get_neighbor(s):
|
|
|
|
|
- self.UpdateVertex(x)
|
|
|
|
|
- # for x in obs_change:
|
|
|
|
|
- # self.obs.remove(x)
|
|
|
|
|
- # for x in obs_change:
|
|
|
|
|
- # self.UpdateVertex(x)
|
|
|
|
|
- print(self.g[(29, 15)])
|
|
|
|
|
- print(self.g[(29, 14)])
|
|
|
|
|
- print(self.g[(29, 13)])
|
|
|
|
|
- print(self.g[(30, 13)])
|
|
|
|
|
- print(self.g[(31, 13)])
|
|
|
|
|
- print(self.g[(32, 13)])
|
|
|
|
|
- print(self.g[(33, 13)])
|
|
|
|
|
- print(self.g[(34, 13)])
|
|
|
|
|
-
|
|
|
|
|
- self.computePath()
|
|
|
|
|
- path.append(self.extract_path_test())
|
|
|
|
|
-
|
|
|
|
|
- return path, obs_change
|
|
|
|
|
-
|
|
|
|
|
- def computePath(self):
|
|
|
|
|
- while self.U.top_key() < self.CalculateKey(self.xG) \
|
|
|
|
|
- or self.rhs[self.xG] != self.g[self.xG]:
|
|
|
|
|
|
|
+ self.fig = plt.figure()
|
|
|
|
|
+ self.Plot.plot_grid("Lifelong Planning A*")
|
|
|
|
|
+
|
|
|
|
|
+ self.ComputePath()
|
|
|
|
|
+ self.plot_path(self.extract_path_test())
|
|
|
|
|
+
|
|
|
|
|
+ self.fig.canvas.mpl_connect('button_press_event', self.on_press)
|
|
|
|
|
+ print("hahha")
|
|
|
|
|
+
|
|
|
|
|
+ plt.show()
|
|
|
|
|
+
|
|
|
|
|
+ def on_press(self, event):
|
|
|
|
|
+ x, y = event.xdata, event.ydata
|
|
|
|
|
+ if x < 0 or x > self.x - 1 or y < 0 or y > self.y - 1:
|
|
|
|
|
+ print("Please choose right area!")
|
|
|
|
|
+ else:
|
|
|
|
|
+ x, y = int(x), int(y)
|
|
|
|
|
+ print("Change position: x =", x, ",", "y =", y)
|
|
|
|
|
+ if (x, y) not in self.obs:
|
|
|
|
|
+ self.obs.add((x, y))
|
|
|
|
|
+ plt.plot(x, y, 'sk')
|
|
|
|
|
+ self.rhs[(x, y)] = float("inf")
|
|
|
|
|
+ self.g[(x, y)] = float("inf")
|
|
|
|
|
+ for node in self.getSucc((x, y)):
|
|
|
|
|
+ self.UpdateVertex(node)
|
|
|
|
|
+ else:
|
|
|
|
|
+ self.obs.remove((x, y))
|
|
|
|
|
+ plt.plot(x, y, marker='s', color='white')
|
|
|
|
|
+ self.UpdateVertex((x, y))
|
|
|
|
|
+ self.ComputePath()
|
|
|
|
|
+ self.plot_path(self.extract_path_test())
|
|
|
|
|
+ self.fig.canvas.draw_idle()
|
|
|
|
|
+
|
|
|
|
|
+ @staticmethod
|
|
|
|
|
+ def plot_path(path):
|
|
|
|
|
+ px = [x[0] for x in path]
|
|
|
|
|
+ py = [x[1] for x in path]
|
|
|
|
|
+ plt.plot(px, py, marker='o')
|
|
|
|
|
+
|
|
|
|
|
+ def ComputePath(self):
|
|
|
|
|
+ while self.U.top_key() < self.Key(self.xG) or \
|
|
|
|
|
+ self.rhs[self.xG] != self.g[self.xG]:
|
|
|
s = self.U.get()
|
|
s = self.U.get()
|
|
|
if self.g[s] > self.rhs[s]:
|
|
if self.g[s] > self.rhs[s]:
|
|
|
self.g[s] = self.rhs[s]
|
|
self.g[s] = self.rhs[s]
|
|
|
|
|
+ for x in self.getSucc(s):
|
|
|
|
|
+ self.UpdateVertex(x)
|
|
|
else:
|
|
else:
|
|
|
self.g[s] = float("inf")
|
|
self.g[s] = float("inf")
|
|
|
self.UpdateVertex(s)
|
|
self.UpdateVertex(s)
|
|
|
- for x in self.get_neighbor(s):
|
|
|
|
|
- self.UpdateVertex(x)
|
|
|
|
|
- # return self.extract_path()
|
|
|
|
|
|
|
+ for x in self.getSucc(s):
|
|
|
|
|
+ self.UpdateVertex(x)
|
|
|
|
|
+
|
|
|
|
|
+ def getSucc(self, s):
|
|
|
|
|
+ nei_list = set()
|
|
|
|
|
+ for u in self.u_set:
|
|
|
|
|
+ s_next = tuple([s[i] + u[i] for i in range(2)])
|
|
|
|
|
+ if s_next not in self.obs and self.g[s_next] > self.g[s]:
|
|
|
|
|
+ nei_list.add(s_next)
|
|
|
|
|
+ return nei_list
|
|
|
|
|
+
|
|
|
|
|
+ def getPred(self, s):
|
|
|
|
|
+ nei_list = set()
|
|
|
|
|
+ for u in self.u_set:
|
|
|
|
|
+ s_next = tuple([s[i] + u[i] for i in range(2)])
|
|
|
|
|
+ if s_next not in self.obs and self.g[s_next] < self.g[s]:
|
|
|
|
|
+ nei_list.add(s_next)
|
|
|
|
|
+ return nei_list
|
|
|
|
|
+
|
|
|
|
|
+ def UpdateVertex(self, s):
|
|
|
|
|
+ if s != self.xI:
|
|
|
|
|
+ u_min = float("inf")
|
|
|
|
|
+ for x in self.getPred(s):
|
|
|
|
|
+ u_min = min(u_min, self.g[x] + self.get_cost(x, s))
|
|
|
|
|
+ self.rhs[s] = u_min
|
|
|
|
|
+ self.U.remove(s)
|
|
|
|
|
+ if self.g[s] != self.rhs[s]:
|
|
|
|
|
+ self.U.put(s, self.Key(s))
|
|
|
|
|
+
|
|
|
|
|
+ def print_g(self):
|
|
|
|
|
+ print("he")
|
|
|
|
|
+ for k in range(self.Env.y_range):
|
|
|
|
|
+ j = self.Env.y_range - k - 1
|
|
|
|
|
+ string = ""
|
|
|
|
|
+ for i in range(self.Env.x_range):
|
|
|
|
|
+ if self.g[(i, j)] == float("inf"):
|
|
|
|
|
+ string += ("00" + ', ')
|
|
|
|
|
+ else:
|
|
|
|
|
+ if self.g[(i, j)] // 10 == 0:
|
|
|
|
|
+ string += ("0" + str(self.g[(i, j)]) + ', ')
|
|
|
|
|
+ else:
|
|
|
|
|
+ string += (str(self.g[(i, j)]) + ', ')
|
|
|
|
|
+ print(string)
|
|
|
|
|
|
|
|
def extract_path(self):
|
|
def extract_path(self):
|
|
|
path = []
|
|
path = []
|
|
@@ -97,11 +152,13 @@ class LpaStar:
|
|
|
path = []
|
|
path = []
|
|
|
s = self.xG
|
|
s = self.xG
|
|
|
|
|
|
|
|
- for k in range(30):
|
|
|
|
|
|
|
+ for k in range(100):
|
|
|
g_list = {}
|
|
g_list = {}
|
|
|
for x in self.get_neighbor(s):
|
|
for x in self.get_neighbor(s):
|
|
|
g_list[x] = self.g[x]
|
|
g_list[x] = self.g[x]
|
|
|
s = min(g_list, key=g_list.get)
|
|
s = min(g_list, key=g_list.get)
|
|
|
|
|
+ if s == self.xI:
|
|
|
|
|
+ return list(reversed(path))
|
|
|
path.append(s)
|
|
path.append(s)
|
|
|
return list(reversed(path))
|
|
return list(reversed(path))
|
|
|
|
|
|
|
@@ -114,32 +171,21 @@ class LpaStar:
|
|
|
|
|
|
|
|
return nei_list
|
|
return nei_list
|
|
|
|
|
|
|
|
- def CalculateKey(self, s):
|
|
|
|
|
|
|
+ def Key(self, s):
|
|
|
return [min(self.g[s], self.rhs[s]) + self.h(s),
|
|
return [min(self.g[s], self.rhs[s]) + self.h(s),
|
|
|
min(self.g[s], self.rhs[s])]
|
|
min(self.g[s], self.rhs[s])]
|
|
|
|
|
|
|
|
- def UpdateVertex(self, u):
|
|
|
|
|
- if u != self.xI:
|
|
|
|
|
- u_min = float("inf")
|
|
|
|
|
- for x in self.get_neighbor(u):
|
|
|
|
|
- u_min = min(u_min, self.g[x] + self.get_cost(u, x))
|
|
|
|
|
- self.rhs[u] = u_min
|
|
|
|
|
- self.U.check_remove(u)
|
|
|
|
|
- if self.g[u] != self.rhs[u]:
|
|
|
|
|
- self.U.put(u, self.CalculateKey(u))
|
|
|
|
|
-
|
|
|
|
|
def h(self, s):
|
|
def h(self, s):
|
|
|
heuristic_type = self.heuristic_type # heuristic type
|
|
heuristic_type = self.heuristic_type # heuristic type
|
|
|
goal = self.xG # goal node
|
|
goal = self.xG # goal node
|
|
|
|
|
|
|
|
if heuristic_type == "manhattan":
|
|
if heuristic_type == "manhattan":
|
|
|
return abs(goal[0] - s[0]) + abs(goal[1] - s[1])
|
|
return abs(goal[0] - s[0]) + abs(goal[1] - s[1])
|
|
|
- elif heuristic_type == "euclidean":
|
|
|
|
|
- return ((goal[0] - s[0]) ** 2 + (goal[1] - s[1]) ** 2) ** (1 / 2)
|
|
|
|
|
else:
|
|
else:
|
|
|
- print("Please choose right heuristic type!")
|
|
|
|
|
|
|
+ return math.hypot(goal[0] - s[0], goal[1] - s[1])
|
|
|
|
|
|
|
|
- def get_cost(self, s_start, s_end):
|
|
|
|
|
|
|
+ @staticmethod
|
|
|
|
|
+ def get_cost(s_start, s_end):
|
|
|
"""
|
|
"""
|
|
|
Calculate cost for this motion
|
|
Calculate cost for this motion
|
|
|
|
|
|
|
@@ -149,12 +195,7 @@ class LpaStar:
|
|
|
:note: cost function could be more complicate!
|
|
:note: cost function could be more complicate!
|
|
|
"""
|
|
"""
|
|
|
|
|
|
|
|
- if s_start not in self.obs:
|
|
|
|
|
- if s_end not in self.obs:
|
|
|
|
|
- return 1
|
|
|
|
|
- else:
|
|
|
|
|
- return float("inf")
|
|
|
|
|
- return float("inf")
|
|
|
|
|
|
|
+ return 1
|
|
|
|
|
|
|
|
|
|
|
|
|
def main():
|
|
def main():
|
|
@@ -162,23 +203,7 @@ def main():
|
|
|
x_goal = (45, 25)
|
|
x_goal = (45, 25)
|
|
|
|
|
|
|
|
lpastar = LpaStar(x_start, x_goal, "euclidean")
|
|
lpastar = LpaStar(x_start, x_goal, "euclidean")
|
|
|
- plot = plotting.Plotting(x_start, x_goal)
|
|
|
|
|
-
|
|
|
|
|
- path, obs = lpastar.searching()
|
|
|
|
|
-
|
|
|
|
|
- plot.plot_grid("Lifelong Planning A*")
|
|
|
|
|
- p = path[0]
|
|
|
|
|
- px = [x[0] for x in p]
|
|
|
|
|
- py = [x[1] for x in p]
|
|
|
|
|
- plt.plot(px, py, marker='o')
|
|
|
|
|
- plt.pause(0.5)
|
|
|
|
|
-
|
|
|
|
|
- p = path[1]
|
|
|
|
|
- px = [x[0] for x in p]
|
|
|
|
|
- py = [x[1] for x in p]
|
|
|
|
|
- plt.plot(px, py, marker='o')
|
|
|
|
|
- plt.pause(0.01)
|
|
|
|
|
- plt.show()
|
|
|
|
|
|
|
+ lpastar.searching()
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if __name__ == '__main__':
|