490 lines
14 KiB
Python
490 lines
14 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 x(self):
|
||
|
|
return self.col
|
||
|
|
|
||
|
|
@property
|
||
|
|
def y(self):
|
||
|
|
return self.row
|
||
|
|
|
||
|
|
@property
|
||
|
|
def is_wall(self):
|
||
|
|
return self.blocked
|
||
|
|
|
||
|
|
@is_wall.setter
|
||
|
|
def is_wall(self, value):
|
||
|
|
self.blocked = value
|
||
|
|
|
||
|
|
def can_step(self):
|
||
|
|
return not self.blocked
|
||
|
|
|
||
|
|
|
||
|
|
class Labyrinth:
|
||
|
|
def __init__(self, width, height):
|
||
|
|
self._w = width
|
||
|
|
self._h = 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._w
|
||
|
|
|
||
|
|
@property
|
||
|
|
def height(self):
|
||
|
|
return self._h
|
||
|
|
|
||
|
|
@property
|
||
|
|
def start(self):
|
||
|
|
return self._start_tile
|
||
|
|
|
||
|
|
@property
|
||
|
|
def exit(self):
|
||
|
|
return self._exit_tile
|
||
|
|
|
||
|
|
def get_tile(self, x, y):
|
||
|
|
if 0 <= x < self._w and 0 <= y < self._h:
|
||
|
|
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 neighbours(self, tile):
|
||
|
|
"""Возвращает список проходимых соседей"""
|
||
|
|
result = []
|
||
|
|
directions = [(0, -1), (0, 1), (-1, 0), (1, 0)] # вверх, вниз, влево, вправо
|
||
|
|
for dx, dy in directions:
|
||
|
|
nx, ny = tile.x + dx, tile.y + dy
|
||
|
|
nb = self.get_tile(nx, ny)
|
||
|
|
if nb and nb.can_step():
|
||
|
|
result.append(nb)
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Загрузка лабиринта ----------
|
||
|
|
class MazeLoader:
|
||
|
|
def load(self, filename):
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
|
||
|
|
class TxtMazeLoader(MazeLoader):
|
||
|
|
def load(self, filename):
|
||
|
|
with open(filename, '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_cnt = 0
|
||
|
|
exit_cnt = 0
|
||
|
|
lab = Labyrinth(w, h)
|
||
|
|
|
||
|
|
for y, line in enumerate(lines):
|
||
|
|
for x, ch in enumerate(line):
|
||
|
|
if ch == "#":
|
||
|
|
lab.set_tile_type(x, y, "wall")
|
||
|
|
elif ch == "S":
|
||
|
|
lab.set_tile_type(x, y, "start")
|
||
|
|
start_cnt += 1
|
||
|
|
elif ch == "E":
|
||
|
|
lab.set_tile_type(x, y, "exit")
|
||
|
|
exit_cnt += 1
|
||
|
|
else:
|
||
|
|
lab.set_tile_type(x, y, 'path')
|
||
|
|
|
||
|
|
if start_cnt != 1 or exit_cnt != 1:
|
||
|
|
raise ValueError(f"Maze error: S={start_cnt}, E={exit_cnt} (need exactly one each)")
|
||
|
|
return lab
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Стратегии поиска пути ----------
|
||
|
|
class SearchStrategy:
|
||
|
|
def find_path(self, lab, start, goal):
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
def _rebuild_path(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 BFS(SearchStrategy):
|
||
|
|
def find_path(self, lab, start, goal):
|
||
|
|
q = deque()
|
||
|
|
q.append(start)
|
||
|
|
parent = {start: None}
|
||
|
|
visited = {start}
|
||
|
|
|
||
|
|
while q:
|
||
|
|
cur = q.popleft()
|
||
|
|
if cur == goal:
|
||
|
|
self._visited = len(visited)
|
||
|
|
return self._rebuild_path(parent, start, goal)
|
||
|
|
for nb in lab.neighbours(cur):
|
||
|
|
if nb not in visited:
|
||
|
|
visited.add(nb)
|
||
|
|
parent[nb] = cur
|
||
|
|
q.append(nb)
|
||
|
|
self._visited = len(visited)
|
||
|
|
return []
|
||
|
|
|
||
|
|
|
||
|
|
class DFS(SearchStrategy):
|
||
|
|
def find_path(self, lab, start, goal):
|
||
|
|
stack = [start]
|
||
|
|
parent = {start: None}
|
||
|
|
visited = {start}
|
||
|
|
|
||
|
|
while stack:
|
||
|
|
cur = stack.pop()
|
||
|
|
if cur == goal:
|
||
|
|
self._visited = len(visited)
|
||
|
|
return self._rebuild_path(parent, start, goal)
|
||
|
|
for nb in lab.neighbours(cur):
|
||
|
|
if nb not in visited:
|
||
|
|
visited.add(nb)
|
||
|
|
parent[nb] = cur
|
||
|
|
stack.append(nb)
|
||
|
|
self._visited = len(visited)
|
||
|
|
return []
|
||
|
|
|
||
|
|
|
||
|
|
class AStar(SearchStrategy):
|
||
|
|
def _heuristic(self, a, b):
|
||
|
|
return abs(a.x - b.x) + abs(a.y - b.y)
|
||
|
|
|
||
|
|
def find_path(self, lab, 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}
|
||
|
|
visited = set()
|
||
|
|
|
||
|
|
while heap:
|
||
|
|
cur_f, _, cur = heapq.heappop(heap)
|
||
|
|
visited.add(cur)
|
||
|
|
if cur == goal:
|
||
|
|
self._visited = len(visited)
|
||
|
|
return self._rebuild_path(parent, start, goal)
|
||
|
|
if cur_f > f.get(cur, float('inf')):
|
||
|
|
continue
|
||
|
|
for nb in lab.neighbours(cur):
|
||
|
|
new_g = g[cur] + 1
|
||
|
|
if new_g < g.get(nb, float('inf')):
|
||
|
|
parent[nb] = cur
|
||
|
|
g[nb] = new_g
|
||
|
|
new_f = new_g + self._heuristic(nb, goal)
|
||
|
|
f[nb] = new_f
|
||
|
|
heapq.heappush(heap, (new_f, counter, nb))
|
||
|
|
counter += 1
|
||
|
|
self._visited = len(visited)
|
||
|
|
return []
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Статистика ----------
|
||
|
|
class SearchStats:
|
||
|
|
def __init__(self, time_ms, visited, path_len):
|
||
|
|
self.time_ms = time_ms
|
||
|
|
self.visited_cells = visited
|
||
|
|
self.path_length = path_len
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Наблюдатель ----------
|
||
|
|
class Observer:
|
||
|
|
def notify(self, event, data):
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
|
||
|
|
class ConsoleDisplay(Observer):
|
||
|
|
def __init__(self, player=None):
|
||
|
|
self._last_path = None
|
||
|
|
self._player = player
|
||
|
|
|
||
|
|
def notify(self, event, data):
|
||
|
|
if event == "maze_loaded":
|
||
|
|
self._draw_maze(data)
|
||
|
|
elif event == "path_found":
|
||
|
|
self._last_path = data
|
||
|
|
self._show_path(data)
|
||
|
|
elif event == "player_moved":
|
||
|
|
self._draw_maze_with_player(data)
|
||
|
|
|
||
|
|
def _draw_maze(self, lab):
|
||
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
print(" LABYRINTH")
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
|
||
|
|
for y in range(lab.height):
|
||
|
|
print(" ", end='')
|
||
|
|
for x in range(lab.width):
|
||
|
|
cell = lab.get_tile(x, y)
|
||
|
|
if cell == lab.start:
|
||
|
|
print('S', end=' ')
|
||
|
|
elif cell == lab.exit:
|
||
|
|
print('E', end=' ')
|
||
|
|
elif cell.is_wall:
|
||
|
|
print('#', end=' ')
|
||
|
|
else:
|
||
|
|
print('.', end=' ')
|
||
|
|
print()
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
print(" S - start E - exit # - wall . - free")
|
||
|
|
|
||
|
|
def _draw_maze_with_player(self, lab):
|
||
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
print(" LABYRINTH (P = player)")
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
|
||
|
|
for y in range(lab.height):
|
||
|
|
print(" ", end='')
|
||
|
|
for x in range(lab.width):
|
||
|
|
cell = lab.get_tile(x, y)
|
||
|
|
if self._player and cell == self._player.position:
|
||
|
|
print('P', end=' ')
|
||
|
|
elif cell == lab.start:
|
||
|
|
print('S', end=' ')
|
||
|
|
elif cell == lab.exit:
|
||
|
|
print('E', end=' ')
|
||
|
|
elif cell.is_wall:
|
||
|
|
print('#', end=' ')
|
||
|
|
else:
|
||
|
|
print('.', end=' ')
|
||
|
|
print()
|
||
|
|
print("=" * (lab.width * 2 + 4))
|
||
|
|
print(f" Player at: ({self._player.position.x}, {self._player.position.y})")
|
||
|
|
print(" S - start E - exit # - wall . - free P - player")
|
||
|
|
|
||
|
|
def _show_path(self, path):
|
||
|
|
if not path:
|
||
|
|
print("\n No route found!")
|
||
|
|
return
|
||
|
|
print(f"\n Route found! Length = {len(path)}")
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Игрок и команды ----------
|
||
|
|
class Player:
|
||
|
|
def __init__(self, start_cell, lab):
|
||
|
|
self._pos = start_cell
|
||
|
|
self._prev = None
|
||
|
|
self._lab = lab
|
||
|
|
|
||
|
|
@property
|
||
|
|
def position(self):
|
||
|
|
return self._pos
|
||
|
|
|
||
|
|
def move(self, target):
|
||
|
|
if target and target.can_step():
|
||
|
|
self._prev = self._pos
|
||
|
|
self._pos = target
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
def undo(self):
|
||
|
|
if self._prev:
|
||
|
|
self._pos, self._prev = self._prev, None
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
class Action:
|
||
|
|
def execute(self):
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
def revert(self):
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
|
||
|
|
class MoveAction(Action):
|
||
|
|
def __init__(self, player, direction, lab):
|
||
|
|
self._player = player
|
||
|
|
self._dx, self._dy = direction
|
||
|
|
self._lab = lab
|
||
|
|
self._done = False
|
||
|
|
|
||
|
|
def execute(self):
|
||
|
|
new_x = self._player.position.x + self._dx
|
||
|
|
new_y = self._player.position.y + self._dy
|
||
|
|
target = self._lab.get_tile(new_x, new_y)
|
||
|
|
if target and target.can_step():
|
||
|
|
self._player.move(target)
|
||
|
|
self._done = True
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
def revert(self):
|
||
|
|
if self._done:
|
||
|
|
self._player.undo()
|
||
|
|
self._done = False
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Решатель лабиринта ----------
|
||
|
|
class LabyrinthSolver:
|
||
|
|
def __init__(self, lab):
|
||
|
|
self._lab = lab
|
||
|
|
self._strategy = None
|
||
|
|
self._watchers = []
|
||
|
|
|
||
|
|
def attach(self, observer):
|
||
|
|
self._watchers.append(observer)
|
||
|
|
|
||
|
|
def _broadcast(self, event, data):
|
||
|
|
for obs in self._watchers:
|
||
|
|
obs.notify(event, data)
|
||
|
|
|
||
|
|
def set_algorithm(self, strategy):
|
||
|
|
self._strategy = strategy
|
||
|
|
|
||
|
|
def solve(self):
|
||
|
|
if self._strategy is None:
|
||
|
|
return None
|
||
|
|
|
||
|
|
t0 = time.perf_counter()
|
||
|
|
path = self._strategy.find_path(self._lab, self._lab.start, self._lab.exit)
|
||
|
|
t1 = time.perf_counter()
|
||
|
|
elapsed_ms = (t1 - t0) * 1000
|
||
|
|
|
||
|
|
self._broadcast("path_found", path)
|
||
|
|
return SearchStats(elapsed_ms, self._strategy.visited_cells(), len(path))
|
||
|
|
|
||
|
|
|
||
|
|
def run_experiment(maze_file, algorithm, runs=5):
|
||
|
|
loader = TxtMazeLoader()
|
||
|
|
lab = loader.load(maze_file)
|
||
|
|
|
||
|
|
total_time = 0.0
|
||
|
|
total_visited = 0
|
||
|
|
total_length = 0
|
||
|
|
|
||
|
|
for _ in range(runs):
|
||
|
|
solver = LabyrinthSolver(lab)
|
||
|
|
solver.set_algorithm(algorithm)
|
||
|
|
stats = solver.solve()
|
||
|
|
if stats:
|
||
|
|
total_time += stats.time_ms
|
||
|
|
total_visited += stats.visited_cells
|
||
|
|
total_length += stats.path_length
|
||
|
|
|
||
|
|
return {
|
||
|
|
'time_ms': total_time / runs,
|
||
|
|
'visited_cells': total_visited / runs,
|
||
|
|
'path_length': total_length / runs
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
# ---------- Точка входа ----------
|
||
|
|
if __name__ == "__main__":
|
||
|
|
if len(sys.argv) > 1 and sys.argv[1] == 'experiment':
|
||
|
|
print("Running experiments...")
|
||
|
|
sys.exit(0)
|
||
|
|
|
||
|
|
loader = TxtMazeLoader()
|
||
|
|
lab = loader.load("maze1.txt")
|
||
|
|
|
||
|
|
player = Player(lab.start, lab)
|
||
|
|
view = ConsoleDisplay(player)
|
||
|
|
view.notify("maze_loaded", lab)
|
||
|
|
|
||
|
|
solver = LabyrinthSolver(lab)
|
||
|
|
solver.attach(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(BFS())
|
||
|
|
stats = solver.solve()
|
||
|
|
print(f"\n BFS: time={stats.time_ms:.3f}ms, visited={stats.visited_cells}, length={stats.path_length}")
|
||
|
|
elif cmd == 'd':
|
||
|
|
solver.set_algorithm(DFS())
|
||
|
|
stats = solver.solve()
|
||
|
|
print(f"\n DFS: time={stats.time_ms:.3f}ms, visited={stats.visited_cells}, length={stats.path_length}")
|
||
|
|
elif cmd == 'a':
|
||
|
|
solver.set_algorithm(AStar())
|
||
|
|
stats = solver.solve()
|
||
|
|
print(f"\n A*: time={stats.time_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)}
|
||
|
|
action = MoveAction(player, dir_map[cmd], lab)
|
||
|
|
if action.execute():
|
||
|
|
history.append(action)
|
||
|
|
view.notify("player_moved", lab)
|
||
|
|
if player.position == lab.exit:
|
||
|
|
print("\n *** VICTORY! EXIT REACHED ***")
|
||
|
|
print(f" Moves made: {len(history)}")
|
||
|
|
break
|
||
|
|
else:
|
||
|
|
print("\n Blocked by wall!")
|
||
|
|
elif cmd == 'u':
|
||
|
|
if history:
|
||
|
|
act = history.pop()
|
||
|
|
act.revert()
|
||
|
|
view.notify("player_moved", lab)
|
||
|
|
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!")
|