490 lines
16 KiB
Python
490 lines
16 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
import csv
|
|
from collections import deque
|
|
import heapq
|
|
from dataclasses import dataclass
|
|
from abc import ABC, abstractmethod
|
|
|
|
class Cell:
|
|
def __init__(self, x, y, is_wall=False, is_start=False, is_exit=False):
|
|
self.x = x
|
|
self.y = y
|
|
self.is_wall = is_wall
|
|
self.is_start = is_start
|
|
self.is_exit = is_exit
|
|
|
|
def is_passable(self):
|
|
return not self.is_wall
|
|
|
|
|
|
class Maze:
|
|
def __init__(self, width, height, cells, start=None, exit=None):
|
|
self.width = width
|
|
self.height = height
|
|
self.cells = cells
|
|
self.start = start
|
|
self.exit = exit
|
|
|
|
def get_cell(self, x, y):
|
|
if 0 <= x < self.width and 0 <= y < self.height:
|
|
return self.cells[y][x]
|
|
return None
|
|
|
|
def get_neighbors(self, cell):
|
|
neighbors = []
|
|
for dx, dy in ((0, -1), (0, 1), (-1, 0), (1, 0)):
|
|
nx, ny = cell.x + dx, cell.y + dy
|
|
neighbor = self.get_cell(nx, ny)
|
|
if neighbor and neighbor.is_passable():
|
|
neighbors.append(neighbor)
|
|
return neighbors
|
|
|
|
|
|
class MazeBuilder:
|
|
@staticmethod
|
|
def build_from_file(filename):
|
|
with open(filename, 'r') as f:
|
|
lines = [line.rstrip('\n') for line in f]
|
|
if not lines:
|
|
raise ValueError("Empty file")
|
|
height = len(lines)
|
|
width = max(len(line) for line in lines)
|
|
cells = []
|
|
start = None
|
|
exit_cell = None
|
|
for y, line in enumerate(lines):
|
|
row = []
|
|
for x in range(width):
|
|
ch = line[x] if x < len(line) else ' '
|
|
is_wall = (ch == '#')
|
|
is_start = (ch == 'S')
|
|
is_exit = (ch == 'E')
|
|
if is_start:
|
|
start = Cell(x, y, False, True, False)
|
|
row.append(start)
|
|
elif is_exit:
|
|
exit_cell = Cell(x, y, False, False, True)
|
|
row.append(exit_cell)
|
|
else:
|
|
row.append(Cell(x, y, is_wall, False, False))
|
|
cells.append(row)
|
|
if start is None:
|
|
raise ValueError("No start cell (S) found")
|
|
if exit_cell is None:
|
|
raise ValueError("No exit cell (E) found")
|
|
return Maze(width, height, cells, start, exit_cell)
|
|
|
|
|
|
class PathFindingStrategy(ABC):
|
|
@abstractmethod
|
|
def find_path(self, maze, start, exit, visit_callback=None):
|
|
pass
|
|
|
|
|
|
class BFSStrategy(PathFindingStrategy):
|
|
def find_path(self, maze, start, exit, visit_callback=None):
|
|
if start == exit:
|
|
if visit_callback: visit_callback(start)
|
|
return [start], 1
|
|
queue = deque([start])
|
|
visited = {start}
|
|
parent = {start: None}
|
|
visited_count = 1
|
|
while queue:
|
|
current = queue.popleft()
|
|
if visit_callback:
|
|
visit_callback(current)
|
|
if current == exit:
|
|
path = []
|
|
while current:
|
|
path.append(current)
|
|
current = parent[current]
|
|
path.reverse()
|
|
return path, visited_count
|
|
for neighbor in maze.get_neighbors(current):
|
|
if neighbor not in visited:
|
|
visited.add(neighbor)
|
|
parent[neighbor] = current
|
|
queue.append(neighbor)
|
|
visited_count += 1
|
|
return [], visited_count
|
|
|
|
|
|
class DFSStrategy(PathFindingStrategy):
|
|
def find_path(self, maze, start, exit, visit_callback=None):
|
|
if start == exit:
|
|
if visit_callback: visit_callback(start)
|
|
return [start], 1
|
|
stack = [start]
|
|
visited = {start}
|
|
parent = {start: None}
|
|
visited_count = 1
|
|
while stack:
|
|
current = stack.pop()
|
|
if visit_callback:
|
|
visit_callback(current)
|
|
if current == exit:
|
|
path = []
|
|
while current:
|
|
path.append(current)
|
|
current = parent[current]
|
|
path.reverse()
|
|
return path, visited_count
|
|
for neighbor in maze.get_neighbors(current):
|
|
if neighbor not in visited:
|
|
visited.add(neighbor)
|
|
parent[neighbor] = current
|
|
stack.append(neighbor)
|
|
visited_count += 1
|
|
return [], visited_count
|
|
|
|
|
|
class AStarStrategy(PathFindingStrategy):
|
|
@staticmethod
|
|
def manhattan(cell, target):
|
|
return abs(cell.x - target.x) + abs(cell.y - target.y)
|
|
|
|
def find_path(self, maze, start, exit, visit_callback=None):
|
|
if start == exit:
|
|
if visit_callback: visit_callback(start)
|
|
return [start], 1
|
|
open_set = []
|
|
counter = 0
|
|
heapq.heappush(open_set, (0, counter, start))
|
|
g_score = {start: 0}
|
|
f_score = {start: self.manhattan(start, exit)}
|
|
parent = {start: None}
|
|
visited_count = 0
|
|
visited = set()
|
|
while open_set:
|
|
_, _, current = heapq.heappop(open_set)
|
|
if current in visited:
|
|
continue
|
|
visited.add(current)
|
|
visited_count += 1
|
|
if visit_callback:
|
|
visit_callback(current)
|
|
if current == exit:
|
|
path = []
|
|
while current:
|
|
path.append(current)
|
|
current = parent[current]
|
|
path.reverse()
|
|
return path, visited_count
|
|
for neighbor in maze.get_neighbors(current):
|
|
tentative_g = g_score[current] + 1
|
|
if neighbor not in g_score or tentative_g < g_score[neighbor]:
|
|
parent[neighbor] = current
|
|
g_score[neighbor] = tentative_g
|
|
f = tentative_g + self.manhattan(neighbor, exit)
|
|
f_score[neighbor] = f
|
|
counter += 1
|
|
heapq.heappush(open_set, (f, counter, neighbor))
|
|
return [], visited_count
|
|
|
|
|
|
@dataclass
|
|
class SearchStats:
|
|
time_ms: float
|
|
visited_cells: int
|
|
path_length: int
|
|
|
|
|
|
class MazeSolver:
|
|
def __init__(self, maze, strategy=None):
|
|
self.maze = maze
|
|
self.strategy = strategy
|
|
|
|
def set_strategy(self, strategy):
|
|
self.strategy = strategy
|
|
|
|
def solve(self, visit_callback=None):
|
|
if self.strategy is None:
|
|
raise ValueError("Strategy not set")
|
|
start_time = time.perf_counter()
|
|
path, visited = self.strategy.find_path(self.maze, self.maze.start, self.maze.exit, visit_callback)
|
|
end_time = time.perf_counter()
|
|
time_ms = (end_time - start_time) * 1000
|
|
return path, SearchStats(time_ms, visited, len(path) if path else 0)
|
|
|
|
|
|
class Observer(ABC):
|
|
@abstractmethod
|
|
def update(self, event_type, data):
|
|
pass
|
|
|
|
|
|
class ConsoleView(Observer):
|
|
def __init__(self, maze, player=None, path=None, show_steps=False):
|
|
self.maze = maze
|
|
self.player = player
|
|
self.path = path or []
|
|
self.show_steps = show_steps
|
|
self.visited = set()
|
|
self._clear_screen()
|
|
|
|
def _clear_screen(self):
|
|
os.system('cls' if os.name == 'nt' else 'clear')
|
|
|
|
def update(self, event_type, data):
|
|
if event_type == 'player_moved':
|
|
self.player = data['player']
|
|
self.render()
|
|
elif event_type == 'path_found':
|
|
self.path = data['path']
|
|
self.visited.clear()
|
|
self.render()
|
|
elif event_type == 'search_step':
|
|
if self.show_steps:
|
|
cell = data['cell']
|
|
self.visited.add(cell)
|
|
self.render()
|
|
elif event_type == 'clear_visited':
|
|
self.visited.clear()
|
|
self.render()
|
|
elif event_type == 'clear':
|
|
self._clear_screen()
|
|
|
|
def render(self):
|
|
self._clear_screen()
|
|
player_pos = self.player.current_cell if self.player else None
|
|
path_set = set(self.path) if self.path else set()
|
|
for y in range(self.maze.height):
|
|
row = []
|
|
for x in range(self.maze.width):
|
|
cell = self.maze.get_cell(x, y)
|
|
if player_pos and cell == player_pos:
|
|
row.append('@')
|
|
elif cell.is_start:
|
|
row.append('S')
|
|
elif cell.is_exit:
|
|
row.append('E')
|
|
elif cell in path_set and not cell.is_start and not cell.is_exit:
|
|
row.append('*')
|
|
elif cell in self.visited and not cell.is_start and not cell.is_exit and not cell.is_wall:
|
|
row.append('.')
|
|
elif cell.is_wall:
|
|
row.append('#')
|
|
else:
|
|
row.append(' ')
|
|
print(''.join(row))
|
|
if player_pos:
|
|
print(f"Player at ({player_pos.x},{player_pos.y})")
|
|
if self.path:
|
|
print(f"Path length: {len(self.path)}")
|
|
print("(Use W/A/S/D to move, U to undo, F to find path, Q to quit)")
|
|
|
|
|
|
class Command(ABC):
|
|
@abstractmethod
|
|
def execute(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def undo(self):
|
|
pass
|
|
|
|
|
|
class MoveCommand(Command):
|
|
def __init__(self, player, dx, dy):
|
|
self.player = player
|
|
self.dx = dx
|
|
self.dy = dy
|
|
self.previous_cell = None
|
|
|
|
def execute(self):
|
|
self.previous_cell = self.player.current_cell
|
|
nx = self.player.current_cell.x + self.dx
|
|
ny = self.player.current_cell.y + self.dy
|
|
target = self.player.maze.get_cell(nx, ny)
|
|
if target and target.is_passable():
|
|
self.player.move_to(target)
|
|
return True
|
|
return False
|
|
|
|
def undo(self):
|
|
if self.previous_cell:
|
|
self.player.move_to(self.previous_cell)
|
|
return True
|
|
return False
|
|
|
|
|
|
class Player:
|
|
def __init__(self, maze, start_cell):
|
|
self.maze = maze
|
|
self.current_cell = start_cell
|
|
|
|
def move_to(self, cell):
|
|
self.current_cell = cell
|
|
|
|
|
|
|
|
def run_experiments():
|
|
test_files = ['maze1.txt', 'maze10x10.txt', 'maze20x20.txt', 'maze_empty.txt', 'maze_no_exit.txt']
|
|
strategies = {
|
|
'BFS': BFSStrategy(),
|
|
'DFS': DFSStrategy(),
|
|
'AStar': AStarStrategy()
|
|
}
|
|
results = []
|
|
runs = 5
|
|
for fname in test_files:
|
|
if not os.path.exists(fname):
|
|
print(f"File {fname} not found, skipping.")
|
|
continue
|
|
try:
|
|
maze = MazeBuilder.build_from_file(fname)
|
|
except Exception as e:
|
|
print(f"Error loading {fname}: {e}")
|
|
continue
|
|
print(f"Testing on {fname} ({maze.width}x{maze.height})")
|
|
for name, strategy in strategies.items():
|
|
total_time = 0.0
|
|
total_visited = 0
|
|
total_length = 0
|
|
success = True
|
|
for _ in range(runs):
|
|
solver = MazeSolver(maze, strategy)
|
|
path, stats = solver.solve()
|
|
if not path:
|
|
success = False
|
|
total_time += stats.time_ms
|
|
total_visited += stats.visited_cells
|
|
total_length += 0
|
|
else:
|
|
total_time += stats.time_ms
|
|
total_visited += stats.visited_cells
|
|
total_length += len(path)
|
|
avg_time = total_time / runs
|
|
avg_visited = total_visited / runs
|
|
avg_length = total_length / runs if success else 0
|
|
results.append({
|
|
'maze': fname,
|
|
'strategy': name,
|
|
'avg_time_ms': avg_time,
|
|
'avg_visited': avg_visited,
|
|
'avg_path_length': avg_length,
|
|
'path_found': success
|
|
})
|
|
print(f" {name}: time={avg_time:.3f}ms, visited={avg_visited:.1f}, length={avg_length:.1f}")
|
|
csv_file = 'experiment_results_2-nd-exercise.csv'
|
|
with open(csv_file, 'w', newline='') as csvfile:
|
|
fieldnames = ['maze', 'strategy', 'avg_time_ms', 'avg_visited', 'avg_path_length', 'path_found']
|
|
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerows(results)
|
|
print(f"Results saved to {csv_file}")
|
|
print("\nSummary Table:")
|
|
print(f"{'Maze':<15} {'Strategy':<10} {'Time(ms)':<12} {'Visited':<10} {'Length':<10} {'Found'}")
|
|
for r in results:
|
|
print(f"{r['maze']:<15} {r['strategy']:<10} {r['avg_time_ms']:<12.3f} {r['avg_visited']:<10.1f} {r['avg_path_length']:<10.1f} {r['path_found']}")
|
|
|
|
|
|
def manual_mode(maze):
|
|
player = Player(maze, maze.start)
|
|
view = ConsoleView(maze, player, show_steps=True)
|
|
command_history = []
|
|
view.render()
|
|
while True:
|
|
cmd = input().strip().lower()
|
|
if cmd == 'q':
|
|
break
|
|
elif cmd == 'u':
|
|
if command_history:
|
|
cmd_obj = command_history.pop()
|
|
cmd_obj.undo()
|
|
view.update('player_moved', {'player': player})
|
|
else:
|
|
print("Nothing to undo")
|
|
elif cmd == 'f':
|
|
print("Finding path from start to exit...")
|
|
strategy = BFSStrategy()
|
|
solver = MazeSolver(maze, strategy)
|
|
path, stats = solver.solve(visit_callback=lambda cell: view.update('search_step', {'cell': cell}))
|
|
view.update('clear_visited', {})
|
|
if path:
|
|
view.update('path_found', {'path': path})
|
|
print(f"Path found! Length: {len(path)}")
|
|
else:
|
|
print("No path found.")
|
|
elif cmd in ('w', 'a', 's', 'd'):
|
|
dx, dy = 0, 0
|
|
if cmd == 'w':
|
|
dy = -1
|
|
elif cmd == 's':
|
|
dy = 1
|
|
elif cmd == 'a':
|
|
dx = -1
|
|
elif cmd == 'd':
|
|
dx = 1
|
|
move_cmd = MoveCommand(player, dx, dy)
|
|
if move_cmd.execute():
|
|
command_history.append(move_cmd)
|
|
view.update('player_moved', {'player': player})
|
|
else:
|
|
print("Can't move there")
|
|
else:
|
|
print("Unknown command")
|
|
|
|
|
|
def interactive_menu():
|
|
while True:
|
|
print("\n==== Maze Explorer ====")
|
|
print("1. Load maze and solve (auto)")
|
|
print("2. Manual control")
|
|
print("3. Run experiments")
|
|
print("4. Quit")
|
|
choice = input("Choose option: ").strip()
|
|
if choice == '1':
|
|
filename = input("Enter maze filename (default maze1.txt): ").strip()
|
|
if not filename:
|
|
filename = 'maze1.txt'
|
|
try:
|
|
maze = MazeBuilder.build_from_file(filename)
|
|
print("Maze loaded.")
|
|
print("Select algorithm: (1) BFS, (2) DFS, (3) A*")
|
|
algo = input("Choice: ").strip()
|
|
if algo == '1':
|
|
strategy = BFSStrategy()
|
|
elif algo == '2':
|
|
strategy = DFSStrategy()
|
|
elif algo == '3':
|
|
strategy = AStarStrategy()
|
|
else:
|
|
print("Invalid, using BFS")
|
|
strategy = BFSStrategy()
|
|
solver = MazeSolver(maze, strategy)
|
|
view = ConsoleView(maze, show_steps=True)
|
|
path, stats = solver.solve(visit_callback=lambda cell: view.update('search_step', {'cell': cell}))
|
|
view.update('clear_visited', {})
|
|
if path:
|
|
view.update('path_found', {'path': path})
|
|
print(f"Path found! Length: {len(path)}, Visited: {stats.visited_cells}, Time: {stats.time_ms:.4f} ms")
|
|
else:
|
|
print("No path found.")
|
|
print(f"Visited: {stats.visited_cells}, Time: {stats.time_ms:.4f} ms")
|
|
input("Press Enter to continue...")
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
elif choice == '2':
|
|
filename = input("Enter maze filename (default maze1.txt): ").strip()
|
|
if not filename:
|
|
filename = 'maze1.txt'
|
|
try:
|
|
maze = MazeBuilder.build_from_file(filename)
|
|
manual_mode(maze)
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
elif choice == '3':
|
|
run_experiments()
|
|
input("Press Enter to continue...")
|
|
elif choice == '4':
|
|
break
|
|
else:
|
|
print("Invalid choice")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
interactive_menu() |