|
| 1 | +""" |
| 2 | +
|
| 3 | +A* grid planning |
| 4 | +
|
| 5 | +author: Atsushi Sakai(@Atsushi_twi) |
| 6 | + Nikos Kanargias ([email protected]) |
| 7 | +
|
| 8 | +adapted by: Reinis Cimurs |
| 9 | +
|
| 10 | +See Wikipedia article (https://en.wikipedia.org/wiki/A*_search_algorithm) |
| 11 | +
|
| 12 | +""" |
| 13 | + |
| 14 | +import math |
| 15 | + |
| 16 | +import matplotlib.pyplot as plt |
| 17 | +import numpy as np |
| 18 | +import shapely |
| 19 | +from irsim.lib.handler.geometry_handler import GeometryFactory |
| 20 | + |
| 21 | + |
| 22 | +class AStarPlanner: |
| 23 | + |
| 24 | + def __init__(self, env, resolution): |
| 25 | + """ |
| 26 | + Initialize A* planner |
| 27 | +
|
| 28 | + env (EnvBase): environment where the planning will take place |
| 29 | + resolution: grid resolution [m] |
| 30 | + """ |
| 31 | + |
| 32 | + self.resolution = resolution |
| 33 | + self.env = env.env |
| 34 | + self.obstacle_list = self.env.obstacle_list[:] |
| 35 | + self.min_x, self.min_y = 0, 0 |
| 36 | + self.max_x, self.max_y = ( |
| 37 | + self.env.env_config.parse["world"]["width"], |
| 38 | + self.env.env_config.parse["world"]["height"], |
| 39 | + ) |
| 40 | + self.x_width = round((self.max_x - self.min_x) / self.resolution) |
| 41 | + self.y_width = round((self.max_y - self.min_y) / self.resolution) |
| 42 | + self.motion = self.get_motion_model() |
| 43 | + |
| 44 | + class Node: |
| 45 | + def __init__(self, x, y, cost, parent_index): |
| 46 | + self.x = x # index of grid |
| 47 | + self.y = y # index of grid |
| 48 | + self.cost = cost |
| 49 | + self.parent_index = parent_index |
| 50 | + |
| 51 | + def __str__(self): |
| 52 | + return ( |
| 53 | + str(self.x) |
| 54 | + + "," |
| 55 | + + str(self.y) |
| 56 | + + "," |
| 57 | + + str(self.cost) |
| 58 | + + "," |
| 59 | + + str(self.parent_index) |
| 60 | + ) |
| 61 | + |
| 62 | + def planning(self, sx, sy, gx, gy, show_animation=True): |
| 63 | + """ |
| 64 | + A star path search |
| 65 | +
|
| 66 | + input: |
| 67 | + s_x: start x position [m] |
| 68 | + s_y: start y position [m] |
| 69 | + gx: goal x position [m] |
| 70 | + gy: goal y position [m] |
| 71 | +
|
| 72 | + output: |
| 73 | + rx: x position list of the final path |
| 74 | + ry: y position list of the final path |
| 75 | + """ |
| 76 | + |
| 77 | + start_node = self.Node( |
| 78 | + self.calc_xy_index(sx, self.min_x), |
| 79 | + self.calc_xy_index(sy, self.min_y), |
| 80 | + 0.0, |
| 81 | + -1, |
| 82 | + ) |
| 83 | + goal_node = self.Node( |
| 84 | + self.calc_xy_index(gx, self.min_x), |
| 85 | + self.calc_xy_index(gy, self.min_y), |
| 86 | + 0.0, |
| 87 | + -1, |
| 88 | + ) |
| 89 | + |
| 90 | + open_set, closed_set = dict(), dict() |
| 91 | + open_set[self.calc_grid_index(start_node)] = start_node |
| 92 | + |
| 93 | + while True: |
| 94 | + if len(open_set) == 0: |
| 95 | + print("Open set is empty..") |
| 96 | + break |
| 97 | + |
| 98 | + c_id = min( |
| 99 | + open_set, |
| 100 | + key=lambda o: open_set[o].cost |
| 101 | + + self.calc_heuristic(goal_node, open_set[o]), |
| 102 | + ) |
| 103 | + current = open_set[c_id] |
| 104 | + |
| 105 | + # show graph |
| 106 | + if show_animation: # pragma: no cover |
| 107 | + plt.plot( |
| 108 | + self.calc_grid_position(current.x, self.min_x), |
| 109 | + self.calc_grid_position(current.y, self.min_y), |
| 110 | + "xc", |
| 111 | + ) |
| 112 | + # for stopping simulation with the esc key. |
| 113 | + plt.gcf().canvas.mpl_connect( |
| 114 | + "key_release_event", |
| 115 | + lambda event: [exit(0) if event.key == "escape" else None], |
| 116 | + ) |
| 117 | + if len(closed_set.keys()) % 10 == 0: |
| 118 | + plt.pause(0.001) |
| 119 | + |
| 120 | + if current.x == goal_node.x and current.y == goal_node.y: |
| 121 | + print("Find goal") |
| 122 | + goal_node.parent_index = current.parent_index |
| 123 | + goal_node.cost = current.cost |
| 124 | + break |
| 125 | + |
| 126 | + # Remove the item from the open set |
| 127 | + del open_set[c_id] |
| 128 | + |
| 129 | + # Add it to the closed set |
| 130 | + closed_set[c_id] = current |
| 131 | + |
| 132 | + # expand_grid search grid based on motion model |
| 133 | + for i, _ in enumerate(self.motion): |
| 134 | + node = self.Node( |
| 135 | + current.x + self.motion[i][0], |
| 136 | + current.y + self.motion[i][1], |
| 137 | + current.cost + self.motion[i][2], |
| 138 | + c_id, |
| 139 | + ) |
| 140 | + n_id = self.calc_grid_index(node) |
| 141 | + |
| 142 | + # If the node is not safe, do nothing |
| 143 | + if not self.verify_node(node): |
| 144 | + continue |
| 145 | + |
| 146 | + if n_id in closed_set: |
| 147 | + continue |
| 148 | + |
| 149 | + if n_id not in open_set: |
| 150 | + open_set[n_id] = node # discovered a new node |
| 151 | + else: |
| 152 | + if open_set[n_id].cost > node.cost: |
| 153 | + # This path is the best until now. record it |
| 154 | + open_set[n_id] = node |
| 155 | + |
| 156 | + rx, ry = self.calc_final_path(goal_node, closed_set) |
| 157 | + |
| 158 | + return rx, ry |
| 159 | + |
| 160 | + def calc_final_path(self, goal_node, closed_set): |
| 161 | + # generate final course |
| 162 | + rx, ry = [self.calc_grid_position(goal_node.x, self.min_x)], [ |
| 163 | + self.calc_grid_position(goal_node.y, self.min_y) |
| 164 | + ] |
| 165 | + parent_index = goal_node.parent_index |
| 166 | + while parent_index != -1: |
| 167 | + n = closed_set[parent_index] |
| 168 | + rx.append(self.calc_grid_position(n.x, self.min_x)) |
| 169 | + ry.append(self.calc_grid_position(n.y, self.min_y)) |
| 170 | + parent_index = n.parent_index |
| 171 | + |
| 172 | + return rx, ry |
| 173 | + |
| 174 | + @staticmethod |
| 175 | + def calc_heuristic(n1, n2): |
| 176 | + w = 1.0 # weight of heuristic |
| 177 | + d = w * math.hypot(n1.x - n2.x, n1.y - n2.y) |
| 178 | + return d |
| 179 | + |
| 180 | + def calc_grid_position(self, index, min_position): |
| 181 | + """ |
| 182 | + calc grid position |
| 183 | +
|
| 184 | + :param index: |
| 185 | + :param min_position: |
| 186 | + :return: |
| 187 | + """ |
| 188 | + pos = index * self.resolution + min_position |
| 189 | + return pos |
| 190 | + |
| 191 | + def calc_xy_index(self, position, min_pos): |
| 192 | + return round((position - min_pos) / self.resolution) |
| 193 | + |
| 194 | + def calc_grid_index(self, node): |
| 195 | + return (node.y - self.min_y) * self.x_width + (node.x - self.min_x) |
| 196 | + |
| 197 | + def verify_node(self, node): |
| 198 | + px = self.calc_grid_position(node.x, self.min_x) |
| 199 | + py = self.calc_grid_position(node.y, self.min_y) |
| 200 | + |
| 201 | + if px < self.min_x: |
| 202 | + return False |
| 203 | + elif py < self.min_y: |
| 204 | + return False |
| 205 | + elif px >= self.max_x: |
| 206 | + return False |
| 207 | + elif py >= self.max_y: |
| 208 | + return False |
| 209 | + |
| 210 | + # collision check |
| 211 | + if self.check_node(px, py): |
| 212 | + return False |
| 213 | + |
| 214 | + return True |
| 215 | + |
| 216 | + def check_node(self, x, y): |
| 217 | + node_position = [x, y] |
| 218 | + shape = { |
| 219 | + "name": "rectangle", |
| 220 | + "length": self.resolution, |
| 221 | + "width": self.resolution, |
| 222 | + } |
| 223 | + gf = GeometryFactory.create_geometry(**shape) |
| 224 | + geometry = gf.step(np.c_[node_position]) |
| 225 | + covered_node = any( |
| 226 | + [shapely.intersects(geometry, obj._geometry) for obj in self.obstacle_list] |
| 227 | + ) |
| 228 | + return covered_node |
| 229 | + |
| 230 | + @staticmethod |
| 231 | + def get_motion_model(): |
| 232 | + # dx, dy, cost |
| 233 | + motion = [ |
| 234 | + [1, 0, 1], |
| 235 | + [0, 1, 1], |
| 236 | + [-1, 0, 1], |
| 237 | + [0, -1, 1], |
| 238 | + [-1, -1, math.sqrt(2)], |
| 239 | + [-1, 1, math.sqrt(2)], |
| 240 | + [1, -1, math.sqrt(2)], |
| 241 | + [1, 1, math.sqrt(2)], |
| 242 | + ] |
| 243 | + |
| 244 | + return motion |
0 commit comments