大規模経路探索の高度最適化
大規模グリッド上の経路探索では、A*では性能が不足する場合があります。ここではJump Point Search (JPS)や階層的経路探索などの高速化手法を学びます。
🔥 Vibe プロンプト
「200x200グリッド(障害物20%)でA*とJPSを比較。探索ノード数と実行時間を計測。」
Jump Point Search (JPS)
JPSは均一コストグリッドに特化したA*の高速化手法で、探索ノード数を10〜100分の1に削減します。
原理
通常のA*は隣接4方向(または8方向)全てをオープンリストに追加しますが、JPSは**跳躍点(Jump Point)**まで一気にスキップすることでノード数を削減します。
A* の探索(20×20グリッド):
● ● ● ● ● ● ● ● ● ● ● ● ●
● ○ ○ ○ ○ ○ ○ ○ ○ ○ ●
● ○ ○ ○ ○ ○ ● ● ● ○ ●
● → → → → → ● S → → G ●
探索ノード数: 約200
JPS の探索:
● ● ● ● ● ● ● ● ● ● ● ● ●
● ●
● ●
● → → → → → → → → → → G ●
探索ノード数: 約30
実装
from typing import List, Tuple, Optional
def jump(grid: List[List[int]], current: Tuple[int, int],
direction: Tuple[int, int]) -> Optional[Tuple[int, int]]:
"""指定方向にジャンプし、Jump Pointがあれば返す"""
r, c = current[0] + direction[0], current[1] + direction[1]
rows, cols = len(grid), len(grid[0])
if not (0 <= r < rows and 0 <= c < cols) or grid[r][c] == 1:
return None
# ゴールに到達
if grid[r][c] == 0 and ...:
return (r, c)
dr, dc = direction
# 対角移動の場合
if dr != 0 and dc != 0:
# 水平または垂直のJump Pointがあるか確認
if (jump(grid, (r, c), (dr, 0)) is not None or
jump(grid, (r, c), (0, dc)) is not None):
return (r, c)
# 水平移動: 上下に障害物があるか?
if dr == 0 and dc != 0:
if (grid[r+1][c] == 1 and grid[r+1][c-dc] == 0) or \
(grid[r-1][c] == 1 and grid[r-1][c-dc] == 0):
return (r, c)
# 垂直移動: 左右に障害物があるか?
if dr != 0 and dc == 0:
if (grid[r][c+1] == 1 and grid[r-dr][c+1] == 0) or \
(grid[r][c-1] == 1 and grid[r-dr][c-1] == 0):
return (r, c)
return jump(grid, (r, c), direction)
JPSの完全なアルゴリズム
import heapq
def jps(grid: List[List[int]], start: Tuple[int, int], goal: Tuple[int, int]):
"""Jump Point Search - メインループ"""
rows, cols = len(grid), len(grid[0])
open_set = [(0, start)]
came_from = {}
g_score = {start: 0}
directions = [(1,0), (-1,0), (0,1), (0,-1),
(1,1), (1,-1), (-1,1), (-1,-1)]
while open_set:
_, current = heapq.heappop(open_set)
if current == goal:
path = []
while current in came_from:
path.append(current)
current = came_from[current]
path.append(start)
return path[::-1]
for d in directions:
jp = jump(grid, current, d)
if jp is not None:
tentative_g = g_score[current] + abs(jp[0]-current[0]) + abs(jp[1]-current[1])
if jp not in g_score or tentative_g < g_score[jp]:
g_score[jp] = tentative_g
f_score = tentative_g + abs(jp[0]-goal[0]) + abs(jp[1]-goal[1])
heapq.heappush(open_set, (f_score, jp))
came_from[jp] = current
return None
A* vs JPS の性能比較
| 指標 | A* | JPS | |------|------|------| | 探索ノード数 | 約10,000 | 約500 | | 実行時間 | 150ms | 8ms | | メモリ使用量 | 5MB | 0.5MB | | 実装の複雑さ | 簡単 | 中程度 | | 汎用性 | 任意のグラフ | 均一コストグリッドのみ |
階層的経路探索 (HPA*)
HPA*はマップを複数の領域に分割し、各領域内の経路を事前計算することで高速化します。
- マップをクラスターに分割(例:10×10のブロック)
- 各クラスターの入口・出口を特定
- 入口間の抽象グラフを作成
- 経路探索は抽象グラフ上で行い、詳細経路は各クラスター内で計算
動的障害物回避
class DynamicObstacleAvoidance:
"""移動する障害物を考慮した経路探索"""
def __init__(self, grid, timesteps=50):
self.grid = grid
self.timesteps = timesteps
def is_safe(self, pos: Tuple[int, int], t: int) -> bool:
"""時刻tにおける位置posの安全性を確認"""
# 静的障害物
if self.grid[pos[0]][pos[1]] == 1:
return False
# 動的障害物(時刻tにその位置にいるか)
for obstacle in self.get_dynamic_obstacles(t):
if obstacle == pos:
return False
return True
def time_augmented_astar(self, start, goal):
"""時間拡張A* - 状態を (位置, 時刻) として探索"""
open_set = [(0, start, 0)] # (f, pos, time)
visited = set()
while open_set:
_, current, t = heapq.heappop(open_set)
if (current, t) in visited:
continue
visited.add((current, t))
if current == goal:
return self.reconstruct_path(...)
for neighbor in self.get_neighbors(current):
if self.is_safe(neighbor, t + 1):
heapq.heappush(open_set, (
t + 1 + abs(neighbor[0]-goal[0]) + abs(neighbor[1]-goal[1]),
neighbor, t + 1
))
return None
実践的な使い分けガイド
| シナリオ | 推奨アルゴリズム | |---------|----------------| | 小規模マップ (< 50×50) | A* | | 大規模グリッドマップ | JPS | | 超巨大マップ(RPGなど) | HPA* | | 動的障害物がある環境 | D* Lite, 時間拡張A* | | 道路ネットワーク | Contraction Hierarchies |
まとめ
- JPS: 均一グリッドでA*より10〜100倍高速
- HPA*: 階層化で超巨大マップに対応
- 動的回避: 時間軸を含めた状態空間で探索
- アルゴリズム選択はマップの種類と要件に応じて判断