527 lines
18 KiB
Python
527 lines
18 KiB
Python
import sys
|
||
import os
|
||
from collections import deque
|
||
import heapq
|
||
import time
|
||
import csv
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
|
||
# ---------- 1. Модель клетки и лабиринта ----------
|
||
class Tile:
|
||
def __init__(self, x, y):
|
||
self.x = x
|
||
self.y = y
|
||
self.wall = False
|
||
self.start = False
|
||
self.exit = False
|
||
|
||
def passable(self):
|
||
return not self.wall
|
||
|
||
|
||
class Labyrinth:
|
||
def __init__(self, width, height):
|
||
self.width = width
|
||
self.height = height
|
||
self._tiles = [[Tile(x, y) for x in range(width)] for y in range(height)]
|
||
self.start_tile = None
|
||
self.exit_tile = None
|
||
|
||
def get_tile(self, x, y):
|
||
if 0 <= x < self.width and 0 <= y < self.height:
|
||
return self._tiles[y][x]
|
||
return None
|
||
|
||
def set_tile(self, x, y, kind):
|
||
tile = self.get_tile(x, y)
|
||
if tile is None:
|
||
return
|
||
if kind == 'wall':
|
||
tile.wall = True
|
||
elif kind == 'start':
|
||
if self.start_tile:
|
||
self.start_tile.start = False
|
||
tile.start = True
|
||
tile.wall = False
|
||
self.start_tile = tile
|
||
elif kind == 'exit':
|
||
if self.exit_tile:
|
||
self.exit_tile.exit = False
|
||
tile.exit = True
|
||
tile.wall = False
|
||
self.exit_tile = tile
|
||
elif kind == 'path':
|
||
tile.wall = False
|
||
|
||
def neighbours(self, tile):
|
||
dirs = [(0, -1), (0, 1), (-1, 0), (1, 0)]
|
||
result = []
|
||
for dx, dy in dirs:
|
||
nx, ny = tile.x + dx, tile.y + dy
|
||
nb = self.get_tile(nx, ny)
|
||
if nb and nb.passable():
|
||
result.append(nb)
|
||
return result
|
||
|
||
|
||
# ---------- 2. Загрузка лабиринта из файла ----------
|
||
class LabyrinthLoader:
|
||
def load(self, filename):
|
||
raise NotImplementedError
|
||
|
||
|
||
class TextLabyrinthLoader(LabyrinthLoader):
|
||
def load(self, filename):
|
||
with open(filename, 'r', encoding='utf-8') as f:
|
||
lines = [line.rstrip('\n') for line in f]
|
||
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(x, y, 'wall')
|
||
elif ch == 'S':
|
||
lab.set_tile(x, y, 'start')
|
||
start_cnt += 1
|
||
elif ch == 'E':
|
||
lab.set_tile(x, y, 'exit')
|
||
exit_cnt += 1
|
||
else:
|
||
lab.set_tile(x, y, 'path')
|
||
if start_cnt != 1 or exit_cnt != 1:
|
||
raise ValueError(f"Нужны ровно S и E, найдено S={start_cnt}, E={exit_cnt}")
|
||
return lab
|
||
|
||
|
||
# ---------- 3. Алгоритмы поиска пути (стратегии) ----------
|
||
class Pathfinder:
|
||
def find_path(self, lab, start, goal):
|
||
raise NotImplementedError
|
||
|
||
def _build_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_count(self):
|
||
return getattr(self, '_visited', 0)
|
||
|
||
|
||
class BFS(Pathfinder):
|
||
def find_path(self, lab, start, goal):
|
||
q = deque([start])
|
||
parent = {start: None}
|
||
visited = {start}
|
||
while q:
|
||
cur = q.popleft()
|
||
if cur == goal:
|
||
self._visited = len(visited)
|
||
return self._build_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(Pathfinder):
|
||
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._build_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(Pathfinder):
|
||
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._build_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 []
|
||
|
||
|
||
# ---------- 4. Оркестратор с поддержкой наблюдателей ----------
|
||
class LabyrinthSolver:
|
||
def __init__(self, lab):
|
||
self.lab = lab
|
||
self.strategy = None
|
||
self.observers = []
|
||
|
||
def attach(self, obs):
|
||
self.observers.append(obs)
|
||
|
||
def notify(self, event, data):
|
||
for obs in self.observers:
|
||
obs.notify(event, data)
|
||
|
||
def set_strategy(self, strategy):
|
||
self.strategy = strategy
|
||
|
||
def solve(self):
|
||
if not self.strategy:
|
||
return None
|
||
t0 = time.perf_counter()
|
||
path = self.strategy.find_path(self.lab, self.lab.start_tile, self.lab.exit_tile)
|
||
t1 = time.perf_counter()
|
||
self.notify("path_found", path)
|
||
return {
|
||
'time_ms': (t1 - t0) * 1000,
|
||
'visited': self.strategy.visited_count(),
|
||
'length': len(path)
|
||
}
|
||
|
||
|
||
# ---------- 5. Визуализация (наблюдатель) ----------
|
||
class EventListener:
|
||
def notify(self, event, data):
|
||
raise NotImplementedError
|
||
|
||
|
||
class ConsoleRenderer(EventListener):
|
||
def __init__(self, walker=None):
|
||
self.last_path = None
|
||
self.walker = walker
|
||
|
||
def notify(self, event, data):
|
||
if event == "maze_loaded":
|
||
self._draw_maze(data)
|
||
elif event == "path_found":
|
||
self.last_path = data
|
||
self._show_path_info(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(" ЛАБИРИНТ")
|
||
print("=" * (lab.width * 2 + 4))
|
||
for y in range(lab.height):
|
||
print(" ", end='')
|
||
for x in range(lab.width):
|
||
t = lab.get_tile(x, y)
|
||
if t == lab.start_tile:
|
||
print('S', end=' ')
|
||
elif t == lab.exit_tile:
|
||
print('E', end=' ')
|
||
elif t.wall:
|
||
print('#', end=' ')
|
||
else:
|
||
print('.', end=' ')
|
||
print()
|
||
print("=" * (lab.width * 2 + 4))
|
||
print(" S - старт E - выход # - стена . - проход")
|
||
|
||
def _draw_maze_with_player(self, lab):
|
||
os.system('cls' if os.name == 'nt' else 'clear')
|
||
print("=" * (lab.width * 2 + 4))
|
||
print(" ЛАБИРИНТ (игрок = P)")
|
||
print("=" * (lab.width * 2 + 4))
|
||
for y in range(lab.height):
|
||
print(" ", end='')
|
||
for x in range(lab.width):
|
||
t = lab.get_tile(x, y)
|
||
if self.walker and t == self.walker.current:
|
||
print('P', end=' ')
|
||
elif t == lab.start_tile:
|
||
print('S', end=' ')
|
||
elif t == lab.exit_tile:
|
||
print('E', end=' ')
|
||
elif t.wall:
|
||
print('#', end=' ')
|
||
else:
|
||
print('.', end=' ')
|
||
print()
|
||
print("=" * (lab.width * 2 + 4))
|
||
if self.walker:
|
||
print(f" Позиция игрока: ({self.walker.current.x}, {self.walker.current.y})")
|
||
|
||
def _show_path_info(self, path):
|
||
if not path:
|
||
print("\n Путь не найден!")
|
||
else:
|
||
print(f"\n Путь найден! Длина = {len(path)}")
|
||
|
||
|
||
# ---------- 6. Игрок и команды с отменой ----------
|
||
class Walker:
|
||
def __init__(self, start_tile, lab):
|
||
self.current = start_tile
|
||
self.prev = None
|
||
self.lab = lab
|
||
|
||
def move(self, target_tile):
|
||
if target_tile and target_tile.passable():
|
||
self.prev = self.current
|
||
self.current = target_tile
|
||
return True
|
||
return False
|
||
|
||
def undo(self):
|
||
if self.prev:
|
||
self.current, self.prev = self.prev, None
|
||
return True
|
||
return False
|
||
|
||
|
||
class Action:
|
||
def do(self):
|
||
raise NotImplementedError
|
||
def undo(self):
|
||
raise NotImplementedError
|
||
|
||
|
||
class MoveAction(Action):
|
||
def __init__(self, walker, dx, dy, lab):
|
||
self.walker = walker
|
||
self.dx = dx
|
||
self.dy = dy
|
||
self.lab = lab
|
||
self.done = False
|
||
|
||
def do(self):
|
||
new_x = self.walker.current.x + self.dx
|
||
new_y = self.walker.current.y + self.dy
|
||
target = self.lab.get_tile(new_x, new_y)
|
||
if target and target.passable():
|
||
self.walker.move(target)
|
||
self.done = True
|
||
return True
|
||
return False
|
||
|
||
def undo(self):
|
||
if self.done:
|
||
self.walker.undo()
|
||
self.done = False
|
||
return True
|
||
return False
|
||
|
||
|
||
# ---------- 7. Экспериментальные функции ----------
|
||
def run_benchmark(maze_file, strategy, runs=5):
|
||
loader = TextLabyrinthLoader()
|
||
lab = loader.load(maze_file)
|
||
total_time = 0.0
|
||
total_visited = 0
|
||
total_len = 0
|
||
for _ in range(runs):
|
||
solver = LabyrinthSolver(lab)
|
||
solver.set_strategy(strategy)
|
||
stats = solver.solve()
|
||
if stats:
|
||
total_time += stats['time_ms']
|
||
total_visited += stats['visited']
|
||
total_len += stats['length']
|
||
return {
|
||
'time_ms': total_time / runs,
|
||
'visited': total_visited / runs,
|
||
'length': total_len / runs
|
||
}
|
||
|
||
|
||
def make_plots(results):
|
||
mazes = list({r['maze'] for r in results})
|
||
algos = ['BFS', 'DFS', 'AStar']
|
||
fig, axes = plt.subplots(1, 3, figsize=(15, 5))
|
||
x = np.arange(len(mazes))
|
||
width = 0.25
|
||
|
||
for i, algo in enumerate(algos):
|
||
times = []
|
||
for m in mazes:
|
||
val = next((r['time_ms'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
|
||
times.append(val)
|
||
axes[0].bar(x + i*width, times, width, label=algo)
|
||
axes[0].set_xlabel('Лабиринт')
|
||
axes[0].set_ylabel('Время (мс)')
|
||
axes[0].set_title('Сравнение времени выполнения')
|
||
axes[0].set_xticks(x + width)
|
||
axes[0].set_xticklabels(mazes, rotation=45, ha='right')
|
||
axes[0].legend()
|
||
axes[0].grid(True, alpha=0.3)
|
||
|
||
for i, algo in enumerate(algos):
|
||
visited_vals = []
|
||
for m in mazes:
|
||
val = next((r['visited'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
|
||
visited_vals.append(val)
|
||
axes[1].bar(x + i*width, visited_vals, width, label=algo)
|
||
axes[1].set_xlabel('Лабиринт')
|
||
axes[1].set_ylabel('Посещено клеток')
|
||
axes[1].set_title('Сравнение посещённых клеток')
|
||
axes[1].set_xticks(x + width)
|
||
axes[1].set_xticklabels(mazes, rotation=45, ha='right')
|
||
axes[1].legend()
|
||
axes[1].grid(True, alpha=0.3)
|
||
|
||
for i, algo in enumerate(algos):
|
||
lengths = []
|
||
for m in mazes:
|
||
val = next((r['length'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
|
||
lengths.append(val)
|
||
axes[2].bar(x + i*width, lengths, width, label=algo)
|
||
axes[2].set_xlabel('Лабиринт')
|
||
axes[2].set_ylabel('Длина пути')
|
||
axes[2].set_title('Сравнение длины пути')
|
||
axes[2].set_xticks(x + width)
|
||
axes[2].set_xticklabels(mazes, rotation=45, ha='right')
|
||
axes[2].legend()
|
||
axes[2].grid(True, alpha=0.3)
|
||
|
||
plt.tight_layout()
|
||
plt.savefig('performance_comparison.png', dpi=150, bbox_inches='tight')
|
||
plt.show()
|
||
|
||
|
||
# ---------- 8. Главный блок ----------
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1 and sys.argv[1] == 'experiment':
|
||
# Режим экспериментов
|
||
test_mazes = [
|
||
("maze/maze1.txt", "Small 10x6"),
|
||
("maze/maze10x10.txt", "Medium 10x10"),
|
||
("maze/maze20x20.txt", "Large 20x20"),
|
||
("maze/maze_empty.txt", "Empty 15x15"),
|
||
("maze/maze_no_exit.txt", "No exit 10x10")
|
||
]
|
||
strategies = [
|
||
("BFS", BFS()),
|
||
("DFS", DFS()),
|
||
("AStar", AStar())
|
||
]
|
||
results = []
|
||
for maze_path, maze_name in test_mazes:
|
||
print(f"Тестируем {maze_name}...")
|
||
for algo_name, algo in strategies:
|
||
try:
|
||
stats = run_benchmark(maze_path, algo, runs=3)
|
||
results.append({
|
||
'maze': maze_name,
|
||
'algo': algo_name,
|
||
'time_ms': stats['time_ms'],
|
||
'visited': stats['visited'],
|
||
'length': stats['length']
|
||
})
|
||
print(f" {algo_name}: время={stats['time_ms']:.3f}мс, посещено={stats['visited']:.0f}, длина={stats['length']:.0f}")
|
||
except Exception as e:
|
||
print(f" {algo_name}: ошибка - {e}")
|
||
# Сохраняем CSV
|
||
with open('experiment_results.csv', 'w', newline='', encoding='utf-8') as f:
|
||
writer = csv.DictWriter(f, fieldnames=['maze', 'algo', 'time_ms', 'visited', 'length'])
|
||
writer.writeheader()
|
||
writer.writerows(results)
|
||
# Строим графики
|
||
if results:
|
||
make_plots(results)
|
||
print("\nРезультаты сохранены в experiment_results.csv и performance_comparison.png")
|
||
else:
|
||
# Интерактивный режим
|
||
loader = TextLabyrinthLoader()
|
||
lab = loader.load("maze/maze1.txt")
|
||
|
||
player = Walker(lab.start_tile, lab)
|
||
view = ConsoleRenderer(player)
|
||
view.notify("maze_loaded", lab)
|
||
|
||
solver = LabyrinthSolver(lab)
|
||
solver.attach(view)
|
||
|
||
print("\n УПРАВЛЕНИЕ:")
|
||
print(" H (влево) J (вниз) K (вверх) L (вправо)")
|
||
print(" U - отменить ход Q - выход")
|
||
print("\n АВТОПОИСК:")
|
||
print(" B - BFS D - DFS A - A*")
|
||
print("\n" + "=" * 50)
|
||
|
||
command_stack = []
|
||
while True:
|
||
key = input("\n Команда > ").lower()
|
||
if key == 'q':
|
||
print("\n До свидания!")
|
||
break
|
||
elif key == 'b':
|
||
solver.set_strategy(BFS())
|
||
stats = solver.solve()
|
||
if stats:
|
||
print(f"\n BFS: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
|
||
elif key == 'd':
|
||
solver.set_strategy(DFS())
|
||
stats = solver.solve()
|
||
print(f"\n DFS: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
|
||
elif key == 'a':
|
||
solver.set_strategy(AStar())
|
||
stats = solver.solve()
|
||
print(f"\n A*: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
|
||
elif key in ['h', 'j', 'k', 'l']:
|
||
dirs = {'h': (-1,0), 'l': (1,0), 'k': (0,-1), 'j': (0,1)}
|
||
act = MoveAction(player, dirs[key][0], dirs[key][1], lab)
|
||
if act.do():
|
||
command_stack.append(act)
|
||
view.notify("player_moved", lab)
|
||
if player.current == lab.exit_tile:
|
||
print("\n ПОЗДРАВЛЯЮ! ВЫ НАШЛИ ВЫХОД!")
|
||
print(f" Сделано ходов: {len(command_stack)}")
|
||
break
|
||
else:
|
||
print("\n Туда нельзя – стена!")
|
||
elif key == 'u':
|
||
if command_stack:
|
||
cmd = command_stack.pop()
|
||
cmd.undo()
|
||
view.notify("player_moved", lab)
|
||
print("\n Отмена последнего хода")
|
||
else:
|
||
print("\n Нечего отменять")
|
||
else:
|
||
print("\n Неизвестная команда. Используйте h,j,k,l, u, b, d, a, q")
|
||
|
||
print("\n Игра окончена. Спасибо!") |