506 lines
15 KiB
Python
506 lines
15 KiB
Python
import sys
|
|
from collections import deque
|
|
import heapq
|
|
import time
|
|
import os
|
|
|
|
|
|
|
|
class Tile:
|
|
def __init__(self, column, row):
|
|
self._col = column
|
|
self._row = row
|
|
self._blocked = False
|
|
self._is_start = False
|
|
self._is_exit = False
|
|
|
|
@property
|
|
def col(self):
|
|
return self._col
|
|
|
|
@property
|
|
def row(self):
|
|
return self._row
|
|
|
|
@property
|
|
def blocked(self):
|
|
return self._blocked
|
|
|
|
@blocked.setter
|
|
def blocked(self, value):
|
|
self._blocked = value
|
|
|
|
@property
|
|
def is_start(self):
|
|
return self._is_start
|
|
|
|
@is_start.setter
|
|
def is_start(self, value):
|
|
self._is_start = value
|
|
|
|
@property
|
|
def is_exit(self):
|
|
return self._is_exit
|
|
|
|
@is_exit.setter
|
|
def is_exit(self, value):
|
|
self._is_exit = value
|
|
|
|
def passable(self):
|
|
return not self._blocked
|
|
|
|
|
|
class Labyrinth:
|
|
def __init__(self, width, height):
|
|
self._width = width
|
|
self._height = height
|
|
self._grid = [[Tile(x, y) for x in range(width)] for y in range(height)]
|
|
self._start_tile = None
|
|
self._exit_tile = None
|
|
|
|
@property
|
|
def width(self):
|
|
return self._width
|
|
|
|
@property
|
|
def height(self):
|
|
return self._height
|
|
|
|
@property
|
|
def start_tile(self):
|
|
return self._start_tile
|
|
|
|
@property
|
|
def exit_tile(self):
|
|
return self._exit_tile
|
|
|
|
def get_tile(self, x, y):
|
|
if 0 <= x < self._width and 0 <= y < self._height:
|
|
return self._grid[y][x]
|
|
return None
|
|
|
|
def set_tile_type(self, x, y, kind):
|
|
tile = self.get_tile(x, y)
|
|
if tile is None:
|
|
return
|
|
|
|
if kind == 'wall':
|
|
tile.blocked = True
|
|
elif kind == 'start':
|
|
if self._start_tile:
|
|
self._start_tile.is_start = False
|
|
tile.is_start = True
|
|
tile.blocked = False
|
|
self._start_tile = tile
|
|
elif kind == 'exit':
|
|
if self._exit_tile:
|
|
self._exit_tile.is_exit = False
|
|
tile.is_exit = True
|
|
tile.blocked = False
|
|
self._exit_tile = tile
|
|
elif kind == 'path':
|
|
tile.blocked = False
|
|
|
|
def neighbors_of(self, tile):
|
|
result = []
|
|
directions = [(0, -1), (0, 1), (-1, 0), (1, 0)]
|
|
for dx, dy in directions:
|
|
nx, ny = tile.col + dx, tile.row + dy
|
|
nb = self.get_tile(nx, ny)
|
|
if nb and nb.passable():
|
|
result.append(nb)
|
|
return result
|
|
|
|
|
|
class LabyrinthLoader:
|
|
def load(self, filepath):
|
|
raise NotImplementedError
|
|
|
|
|
|
class TextFileLoader(LabyrinthLoader):
|
|
def load(self, filepath):
|
|
with open(filepath, 'r') as f:
|
|
lines = [line.rstrip('\n') for line in f.readlines()]
|
|
h = len(lines)
|
|
w = max(len(line) for line in lines) if h > 0 else 0
|
|
|
|
start_count = 0
|
|
exit_count = 0
|
|
lab = Labyrinth(w, h)
|
|
|
|
for row, line in enumerate(lines):
|
|
for col, ch in enumerate(line):
|
|
if ch == "#":
|
|
lab.set_tile_type(col, row, "wall")
|
|
elif ch == "S":
|
|
lab.set_tile_type(col, row, "start")
|
|
start_count += 1
|
|
elif ch == "E":
|
|
lab.set_tile_type(col, row, "exit")
|
|
exit_count += 1
|
|
else:
|
|
lab.set_tile_type(col, row, "path")
|
|
|
|
if start_count != 1 or exit_count != 1:
|
|
raise ValueError(f"Maze must have exactly one 'S' and one 'E'. Found: S={start_count}, E={exit_count}")
|
|
return lab
|
|
|
|
|
|
class SearchAlgorithm:
|
|
def find_route(self, maze, start, goal):
|
|
raise NotImplementedError
|
|
|
|
def _reconstruct(self, came_from, start, goal):
|
|
path = []
|
|
cur = goal
|
|
while cur is not None:
|
|
path.append(cur)
|
|
cur = came_from.get(cur)
|
|
path.reverse()
|
|
return path
|
|
|
|
def visited_cells(self):
|
|
return getattr(self, '_visited', 0)
|
|
|
|
|
|
class BreadthFirstSearch(SearchAlgorithm):
|
|
def find_route(self, maze, start, goal):
|
|
q = deque()
|
|
q.append(start)
|
|
parent = {start: None}
|
|
seen = {start}
|
|
|
|
while q:
|
|
current = q.popleft()
|
|
if current == goal:
|
|
self._visited = len(seen)
|
|
return self._reconstruct(parent, start, goal)
|
|
for nb in maze.neighbors_of(current):
|
|
if nb not in seen:
|
|
seen.add(nb)
|
|
parent[nb] = current
|
|
q.append(nb)
|
|
self._visited = len(seen)
|
|
return []
|
|
|
|
|
|
class DepthFirstSearch(SearchAlgorithm):
|
|
def find_route(self, maze, start, goal):
|
|
stack = [start]
|
|
parent = {start: None}
|
|
seen = {start}
|
|
|
|
while stack:
|
|
current = stack.pop()
|
|
if current == goal:
|
|
self._visited = len(seen)
|
|
return self._reconstruct(parent, start, goal)
|
|
for nb in maze.neighbors_of(current):
|
|
if nb not in seen:
|
|
seen.add(nb)
|
|
parent[nb] = current
|
|
stack.append(nb)
|
|
self._visited = len(seen)
|
|
return []
|
|
|
|
|
|
class AStarSearch(SearchAlgorithm):
|
|
def _heuristic(self, tile, goal):
|
|
return abs(tile.col - goal.col) + abs(tile.row - goal.row)
|
|
|
|
def find_route(self, maze, start, goal):
|
|
heap = []
|
|
counter = 0
|
|
start_f = self._heuristic(start, goal)
|
|
heapq.heappush(heap, (start_f, counter, start))
|
|
counter += 1
|
|
|
|
parent = {}
|
|
g = {start: 0}
|
|
f = {start: start_f}
|
|
closed = set()
|
|
|
|
while heap:
|
|
cur_f, _, cur = heapq.heappop(heap)
|
|
closed.add(cur)
|
|
|
|
if cur == goal:
|
|
self._visited = len(closed)
|
|
return self._reconstruct(parent, start, goal)
|
|
|
|
if cur_f > f.get(cur, float('inf')):
|
|
continue
|
|
|
|
for nb in maze.neighbors_of(cur):
|
|
tentative_g = g[cur] + 1
|
|
if tentative_g < g.get(nb, float('inf')):
|
|
parent[nb] = cur
|
|
g[nb] = tentative_g
|
|
new_f = tentative_g + self._heuristic(nb, goal)
|
|
f[nb] = new_f
|
|
heapq.heappush(heap, (new_f, counter, nb))
|
|
counter += 1
|
|
|
|
self._visited = len(closed)
|
|
return []
|
|
|
|
|
|
class SearchStats:
|
|
def __init__(self, elapsed_ms, visited, path_len):
|
|
self.elapsed_ms = elapsed_ms
|
|
self.visited_cells = visited
|
|
self.path_length = path_len
|
|
|
|
|
|
class EventListener:
|
|
def on_event(self, event_type, data):
|
|
raise NotImplementedError
|
|
|
|
|
|
class TerminalView(EventListener):
|
|
def __init__(self, player=None):
|
|
self._current_path = None
|
|
self._player = player
|
|
|
|
def on_event(self, event_type, data):
|
|
if event_type == "maze_loaded":
|
|
self._display_maze(data)
|
|
elif event_type == "path_found":
|
|
self._current_path = data
|
|
self._display_path(data)
|
|
elif event_type == "player_moved":
|
|
self._display_maze_with_player(data)
|
|
|
|
def _display_maze(self, maze):
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
|
print("=" * (maze.width * 2 + 4))
|
|
print(" LABYRINTH")
|
|
print("=" * (maze.width * 2 + 4))
|
|
|
|
for y in range(maze.height):
|
|
print(" ", end='')
|
|
for x in range(maze.width):
|
|
cell = maze.get_tile(x, y)
|
|
if cell == maze.start_tile:
|
|
print('S', end=' ')
|
|
elif cell == maze.exit_tile:
|
|
print('E', end=' ')
|
|
elif cell.blocked:
|
|
print('#', end=' ')
|
|
else:
|
|
print('.', end=' ')
|
|
print()
|
|
print("=" * (maze.width * 2 + 4))
|
|
print(" S - start E - exit # - wall . - path")
|
|
|
|
def _display_maze_with_player(self, maze):
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
|
print("=" * (maze.width * 2 + 4))
|
|
print(" LABYRINTH (P = player)")
|
|
print("=" * (maze.width * 2 + 4))
|
|
|
|
for y in range(maze.height):
|
|
print(" ", end='')
|
|
for x in range(maze.width):
|
|
cell = maze.get_tile(x, y)
|
|
if self._player and cell == self._player.position:
|
|
print('P', end=' ')
|
|
elif cell == maze.start_tile:
|
|
print('S', end=' ')
|
|
elif cell == maze.exit_tile:
|
|
print('E', end=' ')
|
|
elif cell.blocked:
|
|
print('#', end=' ')
|
|
else:
|
|
print('.', end=' ')
|
|
print()
|
|
print("=" * (maze.width * 2 + 4))
|
|
print(f" Player at: ({self._player.position.col}, {self._player.position.row})")
|
|
print(" S - start E - exit # - wall . - path P - player")
|
|
|
|
def _display_path(self, path):
|
|
if not path:
|
|
print("\n No route found!")
|
|
else:
|
|
print(f"\n Path found! Length = {len(path)}")
|
|
|
|
|
|
class Player:
|
|
def __init__(self, start_tile, labyrinth):
|
|
self._pos = start_tile
|
|
self._prev = None
|
|
self._lab = labyrinth
|
|
|
|
@property
|
|
def position(self):
|
|
return self._pos
|
|
|
|
def move_to(self, new_tile):
|
|
if new_tile and new_tile.passable():
|
|
self._prev = self._pos
|
|
self._pos = new_tile
|
|
return True
|
|
return False
|
|
|
|
def undo(self):
|
|
if self._prev:
|
|
self._pos, self._prev = self._prev, None
|
|
return True
|
|
return False
|
|
|
|
|
|
class Command:
|
|
def do(self):
|
|
raise NotImplementedError
|
|
|
|
def undo(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class MoveCommand(Command):
|
|
def __init__(self, player, direction, labyrinth):
|
|
self._player = player
|
|
self._dx, self._dy = direction
|
|
self._lab = labyrinth
|
|
self._done = False
|
|
|
|
def do(self):
|
|
nx = self._player.position.col + self._dx
|
|
ny = self._player.position.row + self._dy
|
|
target = self._lab.get_tile(nx, ny)
|
|
if target and target.passable():
|
|
self._player.move_to(target)
|
|
self._done = True
|
|
return True
|
|
return False
|
|
|
|
def undo(self):
|
|
if self._done:
|
|
self._player.undo()
|
|
self._done = False
|
|
return True
|
|
return False
|
|
|
|
|
|
|
|
class MazeSolver:
|
|
"""Controls the search process and notifies observers."""
|
|
|
|
def __init__(self, labyrinth):
|
|
self._lab = labyrinth
|
|
self._algorithm = None
|
|
self._listeners = []
|
|
|
|
def add_listener(self, listener):
|
|
self._listeners.append(listener)
|
|
|
|
def notify(self, event, data):
|
|
for lst in self._listeners:
|
|
lst.on_event(event, data)
|
|
|
|
def set_algorithm(self, algo):
|
|
self._algorithm = algo
|
|
|
|
def solve(self):
|
|
if self._algorithm is None:
|
|
return None
|
|
|
|
start_time = time.perf_counter()
|
|
route = self._algorithm.find_route(self._lab, self._lab.start_tile, self._lab.exit_tile)
|
|
end_time = time.perf_counter()
|
|
elapsed_ms = (end_time - start_time) * 1000
|
|
|
|
self.notify("path_found", route)
|
|
return SearchStats(elapsed_ms, self._algorithm.visited_cells(), len(route))
|
|
|
|
|
|
def run_experiment(maze_file, algorithm, repetitions=5):
|
|
loader = TextFileLoader()
|
|
maze = loader.load(maze_file)
|
|
|
|
total_time = 0.0
|
|
total_visited = 0
|
|
total_length = 0
|
|
|
|
for _ in range(repetitions):
|
|
solver = MazeSolver(maze)
|
|
solver.set_algorithm(algorithm)
|
|
stats = solver.solve()
|
|
if stats:
|
|
total_time += stats.elapsed_ms
|
|
total_visited += stats.visited_cells
|
|
total_length += stats.path_length
|
|
|
|
return {
|
|
'time_ms': total_time / repetitions,
|
|
'visited_cells': total_visited / repetitions,
|
|
'path_length': total_length / repetitions
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1 and sys.argv[1] == 'experiment':
|
|
print("Running experiments (use plots.py for full test suite)...")
|
|
sys.exit(0)
|
|
|
|
loader = TextFileLoader()
|
|
maze = loader.load("maze1.txt")
|
|
|
|
player = Player(maze.start_tile, maze)
|
|
view = TerminalView(player)
|
|
view.on_event("maze_loaded", maze)
|
|
|
|
solver = MazeSolver(maze)
|
|
solver.add_listener(view)
|
|
|
|
print("\n CONTROLS:")
|
|
print(" H (left) J (down) K (up) L (right)")
|
|
print(" U - undo Q - quit")
|
|
print("\n AUTO SEARCH:")
|
|
print(" B - BFS D - DFS A - A*")
|
|
print("\n" + "=" * 50)
|
|
|
|
history = []
|
|
|
|
while True:
|
|
cmd = input("\n Command > ").lower()
|
|
|
|
if cmd == 'q':
|
|
print("\n Goodbye!")
|
|
break
|
|
elif cmd == 'b':
|
|
solver.set_algorithm(BreadthFirstSearch())
|
|
stats = solver.solve()
|
|
print(f"\n BFS: time={stats.elapsed_ms:.3f}ms, visited={stats.visited_cells}, length={stats.path_length}")
|
|
elif cmd == 'd':
|
|
solver.set_algorithm(DepthFirstSearch())
|
|
stats = solver.solve()
|
|
print(f"\n DFS: time={stats.elapsed_ms:.3f}ms, visited={stats.visited_cells}, length={stats.path_length}")
|
|
elif cmd == 'a':
|
|
solver.set_algorithm(AStarSearch())
|
|
stats = solver.solve()
|
|
print(f"\n A*: time={stats.elapsed_ms:.3f}ms, visited={stats.visited_cells}, length={stats.path_length}")
|
|
elif cmd in ['h', 'j', 'k', 'l']:
|
|
dir_map = {'h': (-1, 0), 'l': (1, 0), 'k': (0, -1), 'j': (0, 1)}
|
|
move = MoveCommand(player, dir_map[cmd], maze)
|
|
if move.do():
|
|
history.append(move)
|
|
view.on_event("player_moved", maze)
|
|
if player.position == maze.exit_tile:
|
|
print("\n *** YOU ESCAPED! ***")
|
|
print(f" Total moves: {len(history)}")
|
|
break
|
|
else:
|
|
print("\n Blocked by a wall!")
|
|
elif cmd == 'u':
|
|
if history:
|
|
last = history.pop()
|
|
last.undo()
|
|
view.on_event("player_moved", maze)
|
|
print("\n Undo successful")
|
|
else:
|
|
print("\n Nothing to undo")
|
|
else:
|
|
print("\n Unknown command. Use h,j,k,l to move, u to undo, q to quit")
|
|
|
|
print("\n Game over. Thanks for playing!") |