Skip to content

Commit bcd3e1f

Browse files
author
Uzia Peer
committed
Implement Basic Theta star path search
1 parent 9d48ba3 commit bcd3e1f

1 file changed

Lines changed: 213 additions & 6 deletions

File tree

‎graphs/theta_star.py‎

Lines changed: 213 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,154 @@
77
Date: 2026-10-12
88
"""
99

10+
import heapq
11+
import logging
12+
from itertools import count, pairwise
13+
from math import hypot, inf
1014

11-
def theta_star(
12-
grid: list[list[int]], # noqa: ARG001
13-
start: tuple[int, int], # noqa: ARG001
14-
goal: tuple[int, int], # noqa: ARG001
15-
) -> list[tuple[int, int]]:
15+
Point = tuple[int, int]
16+
logger = logging.getLogger(__name__)
17+
18+
19+
class CornerGrid:
20+
"""A rectangular grid whose path vertices are cell corners.
21+
22+
The bottom row is ``cells[0]``. Obstacles are marked by 1.
23+
24+
>>> grid = CornerGrid([[0, 1], [0, 0]])
25+
>>> grid.width, grid.height
26+
(2, 2)
27+
>>> grid.line_of_sight((0, 0), (2, 2))
28+
True
29+
"""
30+
31+
def __init__(self, cells: list[list[int]]) -> None:
32+
rows = tuple(tuple(row) for row in cells)
33+
if not rows or not rows[0] or any(len(row) != len(rows[0]) for row in rows):
34+
raise ValueError("Grid must be a non-empty rectangular matrix.")
35+
if any(
36+
type(cell) is not int or cell not in (0, 1) for row in rows for cell in row
37+
):
38+
raise ValueError("Cells must be integers 0 (free) or 1 (blocked).")
39+
self.cells = rows
40+
self.height, self.width = len(rows), len(rows[0])
41+
42+
def blocked(self, x: int, y: int) -> bool:
43+
"""Treat cells outside the grid as blocked.
44+
45+
>>> CornerGrid([[0]]).blocked(-1, 0)
46+
True
47+
"""
48+
return not (0 <= x < self.width and 0 <= y < self.height) or bool(
49+
self.cells[y][x]
50+
)
51+
52+
def valid_corner(self, point: Point) -> bool:
53+
"""A usable corner touches at least one free cell.
54+
55+
>>> CornerGrid([[1]]).valid_corner((0, 0))
56+
False
57+
"""
58+
if not isinstance(point, tuple) or len(point) != 2:
59+
return False
60+
x, y = point
61+
if type(x) is not int or type(y) is not int:
62+
return False
63+
return (
64+
0 <= x <= self.width
65+
and 0 <= y <= self.height
66+
and any(not self.blocked(cx, cy) for cx in (x - 1, x) for cy in (y - 1, y))
67+
)
68+
69+
def line_of_sight(self, a: Point, b: Point) -> bool:
70+
"""Test visibility with integer grid traversal, as in the paper.
71+
72+
Diagonal contact at one corner is allowed, but moving along an edge
73+
shared by two blocked cells is not.
74+
75+
>>> CornerGrid([[0, 1], [1, 0]]).line_of_sight((0, 0), (2, 2))
76+
True
77+
>>> CornerGrid([[1], [1]]).line_of_sight((0, 0), (0, 2))
78+
False
79+
"""
80+
if not self.valid_corner(a) or not self.valid_corner(b):
81+
return False
82+
x, y = a
83+
dx, dy = abs(b[0] - x), abs(b[1] - y)
84+
sx = 1 if b[0] >= x else -1
85+
sy = 1 if b[1] >= y else -1
86+
ox, oy = (sx - 1) // 2, (sy - 1) // 2
87+
error = 0
88+
if dx >= dy:
89+
while x != b[0]:
90+
error += dy
91+
if error >= dx:
92+
if self.blocked(x + ox, y + oy):
93+
return False
94+
y += sy
95+
error -= dx
96+
if error and self.blocked(x + ox, y + oy):
97+
return False
98+
if dy == 0 and self.blocked(x + ox, y) and self.blocked(x + ox, y - 1):
99+
return False
100+
x += sx
101+
else:
102+
while y != b[1]:
103+
error += dx
104+
if error >= dy:
105+
if self.blocked(x + ox, y + oy):
106+
return False
107+
x += sx
108+
error -= dy
109+
if error and self.blocked(x + ox, y + oy):
110+
return False
111+
if dx == 0 and self.blocked(x, y + oy) and self.blocked(x - 1, y + oy):
112+
return False
113+
y += sy
114+
return True
115+
116+
def neighbors(self, point: Point) -> list[Point]:
117+
"""Return visible neighboring corners in a fixed order.
118+
119+
>>> CornerGrid([[0]]).neighbors((0, 0))
120+
[(0, 1), (1, 0), (1, 1)]
121+
"""
122+
x, y = point
123+
candidates = ((x + dx, y + dy) for dx in (-1, 0, 1) for dy in (-1, 0, 1))
124+
return [q for q in candidates if q != point and self.line_of_sight(point, q)]
125+
126+
127+
def distance(a: Point, b: Point) -> float:
128+
"""Euclidean distance ``c(a,b)`` from the paper.
129+
130+
>>> distance((0, 0), (3, 4))
131+
5.0
132+
"""
133+
return hypot(a[0] - b[0], a[1] - b[1])
134+
135+
136+
def path_length(path: list[Point]) -> float:
137+
"""Add the lengths of the path segments.
138+
139+
>>> path_length([(0, 0), (3, 4), (6, 4)])
140+
8.0
141+
"""
142+
return sum(distance(a, b) for a, b in pairwise(path))
143+
144+
145+
def reconstruct_path(parent: dict[Point, Point], goal: Point) -> list[Point]:
146+
"""Follow parent pointers from the goal back to the start.
147+
148+
>>> reconstruct_path({(0, 0): (0, 0), (1, 1): (0, 0)}, (1, 1))
149+
[(0, 0), (1, 1)]
150+
"""
151+
path = [goal]
152+
while parent[path[-1]] != path[-1]:
153+
path.append(parent[path[-1]])
154+
return path[::-1]
155+
156+
157+
def theta_star(grid: list[list[int]], start: Point, goal: Point) -> list[Point]:
16158
"""Find an any-angle path through a grid using Basic Theta* (Algorithms 1-2).
17159
18160
``grid[y][x]`` is 0 for a free cell and 1 for an obstacle. The first row
@@ -75,5 +217,70 @@ def theta_star(
75217
>>> path
76218
[(0, 0), (6, 7), (9, 10), (10, 10)]
77219
"""
78-
# Assignment 4: search intentionally left empty. Tests should fail.
220+
corner_grid = CornerGrid(grid)
221+
if not corner_grid.valid_corner(start) or not corner_grid.valid_corner(goal):
222+
raise ValueError("Start and goal must be usable integer grid corners.")
223+
return _search(corner_grid, start, goal)
224+
225+
226+
def _search(corner_grid: CornerGrid, start: Point, goal: Point) -> list[Point]:
227+
"""Run Basic Theta* on a validated grid (shared with the cache experiment).
228+
229+
>>> _search(CornerGrid([[0]]), (0, 0), (1, 1))
230+
[(0, 0), (1, 1)]
231+
"""
232+
233+
# g, parent, open, and closed use the same meanings as in Algorithms 1-2.
234+
# h is the Euclidean distance to goal. Equal f prefers smaller g.
235+
g = {start: 0.0}
236+
parent = {start: start}
237+
closed: set[Point] = set()
238+
serial = count()
239+
opened = [(distance(start, goal), 0.0, next(serial), start)]
240+
241+
while opened:
242+
_, old_g, _, current = heapq.heappop(opened)
243+
if current in closed or old_g != g[current]:
244+
continue
245+
logger.debug("Expand %s g=%.6f parent=%s", current, g[current], parent[current])
246+
if current == goal:
247+
path = reconstruct_path(parent, goal)
248+
logger.info("Found path of length %.6f", path_length(path))
249+
return path
250+
251+
closed.add(current)
252+
for neighbor in corner_grid.neighbors(current):
253+
if neighbor in closed:
254+
logger.debug("Skip closed neighbor %s", neighbor)
255+
continue
256+
257+
# Path 2 goes directly from parent(current) to neighbor if visible.
258+
# Otherwise Path 1 goes through current.
259+
previous = parent[current]
260+
shortcut = corner_grid.line_of_sight(previous, neighbor)
261+
candidate_parent = previous if shortcut else current
262+
branch = "Path 2" if shortcut else "Path 1"
263+
candidate = g[candidate_parent] + distance(candidate_parent, neighbor)
264+
if candidate < g.get(neighbor, inf):
265+
g[neighbor], parent[neighbor] = candidate, candidate_parent
266+
heapq.heappush(
267+
opened,
268+
(
269+
candidate + distance(neighbor, goal),
270+
candidate,
271+
next(serial),
272+
neighbor,
273+
),
274+
)
275+
logger.debug(
276+
"%s: %s -> %s; g=%.6f",
277+
branch,
278+
candidate_parent,
279+
neighbor,
280+
candidate,
281+
)
282+
else:
283+
logger.debug("%s: no improvement for %s", branch, neighbor)
284+
285+
logger.info("No path from %s to %s", start, goal)
79286
return []

0 commit comments

Comments
 (0)