596 lines
17 KiB
Python
596 lines
17 KiB
Python
import numpy as np
|
||
import abc
|
||
from collections import deque
|
||
import heapq
|
||
import time
|
||
import os
|
||
import keyboard
|
||
import csv
|
||
|
||
#Классы клетки и лабиринта
|
||
|
||
class Cell:
|
||
|
||
def __init__(self, coords, isWall = False, isStart = False,
|
||
isExit = False):
|
||
self.coords = coords
|
||
self.isWall = isWall
|
||
self.isStart = isStart
|
||
self.isExit = isExit
|
||
|
||
def isPassable(self):
|
||
if self.isWall:
|
||
return False
|
||
return True
|
||
|
||
class Maze:
|
||
|
||
def __init__(self, cells, width, height, st, ex):
|
||
self.cells = cells
|
||
self.width = width
|
||
self.height = height
|
||
self.st = st
|
||
self.ex = ex
|
||
|
||
def getCell(self,x,y):
|
||
try:
|
||
return self.cells[x][y]
|
||
except:
|
||
return None
|
||
|
||
def getNeighbors(self,cell):
|
||
x,y = cell.coords
|
||
res = []
|
||
for i,j in (x,y+1),(x,y-1),(x-1,y),(x+1,y):
|
||
cellij = self.getCell(i,j)
|
||
if i <= self.width-1 and j <= self.height-1 and 0 <= i and 0 <= j and cellij is not None:
|
||
if cellij.isPassable():
|
||
res.append(cellij)
|
||
else:
|
||
res.append(None)
|
||
else:
|
||
res.append(None)
|
||
|
||
return res
|
||
|
||
#Тестирование классов клетки и лабиринта
|
||
|
||
# cell1 = Cell((1,2), isExit = True, isWall = True, isStart = False)
|
||
|
||
# print(cell1.isPassable())
|
||
# print(cell1.isStart)
|
||
# print(cell1.coords)
|
||
|
||
# width, height = 3,3
|
||
|
||
# cells = np.full((width,height), None, dtype=object)
|
||
|
||
# for x in range(width):
|
||
# for y in range(height):
|
||
# if x != 0 and x != width-1 and y != 0 and y != height-1:
|
||
# cells[x][y] = Cell((x,y), isWall = False)
|
||
# else:
|
||
# cells[x][y] = Cell((x,y), isWall = True)
|
||
|
||
# print(cells)
|
||
|
||
# maze1 = Maze(cells, width, height, cells[0], cells[-1])
|
||
|
||
# for column in cells:
|
||
# for cell in column:
|
||
# print(cell.coords)
|
||
# print(maze1.getNeighbors(cell))
|
||
|
||
# print('\n')
|
||
|
||
#Интерфейс постройки лабиринта
|
||
|
||
class MazeBuilder(abc.ABC):
|
||
|
||
@abc.abstractmethod
|
||
def buildFromFile(filename): pass
|
||
|
||
#Наследуем от него класс постройки из текстового файла
|
||
|
||
class TextFileMazeBuilder(MazeBuilder):
|
||
|
||
def buildFromFile(filename):
|
||
with open(filename, "r") as file:
|
||
|
||
rows = file.read().splitlines()
|
||
|
||
#print(rows)
|
||
|
||
width = 0
|
||
height = 0
|
||
for row in rows:
|
||
height += 1
|
||
if len(row) > width:
|
||
width = len(row)
|
||
|
||
#print(width, height)
|
||
|
||
st = (0,0)
|
||
ex = (width,height)
|
||
|
||
cells = np.full((width,height), None, dtype=object)
|
||
|
||
flagst = False
|
||
flagex = False
|
||
for y in range(height):
|
||
for x in range(width):
|
||
isWall = False
|
||
isStart = False
|
||
isExit = False
|
||
|
||
if rows[-(y+1)][x] == '#':
|
||
isWall = True
|
||
elif rows[-(y+1)][x] == 'S':
|
||
isStart = True
|
||
st = (x,y)
|
||
flagst = True
|
||
#print('Старт в',x,y)
|
||
elif rows[-(y+1)][x] == 'E':
|
||
isExit = True
|
||
ex = (x,y)
|
||
flagex = True
|
||
#print('Выход в',x,y)
|
||
elif rows[-(y+1)][x] != ' ':
|
||
raise ValueError("Неверный формат лабиринта! Пожалуйста, используйте только символы #,S,E и пробелы")
|
||
|
||
cells[x][y] = Cell((x,y), isWall, isStart, isExit)
|
||
|
||
if flagst and flagex:
|
||
|
||
return Maze(cells, width, height, cells[st[0]][st[1]], cells[ex[0]][ex[1]])
|
||
|
||
raise ValueError('В лабаиринте должны быть вход и выход (S и E)')
|
||
|
||
|
||
# builder = TextFileMazeBuilder
|
||
|
||
# maze = builder.buildFromFile('maze1.txt')
|
||
|
||
# print(maze)
|
||
|
||
|
||
#Интерфейс поиска пути
|
||
|
||
class PathFindingStrategy(abc.ABC):
|
||
|
||
@abc.abstractmethod
|
||
def findPath(self, maze, st, ex): pass
|
||
|
||
#Поиск в глубину
|
||
|
||
class DFS(PathFindingStrategy):
|
||
|
||
def findPath(self,maze,st,ex):
|
||
|
||
stack = [st]
|
||
|
||
self.visited = {st.coords} #по координатам надёжнее, а то вдруг адрес изменится
|
||
|
||
pathmap = {}
|
||
|
||
while stack:
|
||
cell = stack.pop()
|
||
|
||
if cell.coords == ex.coords:
|
||
|
||
#маршрут выстраивается в обратном порядке и разворачивается
|
||
path = []
|
||
while cell.coords != st.coords:
|
||
path.append(cell)
|
||
cell = pathmap[cell.coords]
|
||
path.append(st)
|
||
path = path[::-1]
|
||
return path
|
||
|
||
for n in maze.getNeighbors(cell):
|
||
if n != None and n.coords not in self.visited:
|
||
self.visited.add(n.coords)
|
||
pathmap[n.coords] = cell
|
||
stack.append(n)
|
||
|
||
return None
|
||
|
||
# path = DFS().findPath(maze,maze.st,maze.ex)
|
||
|
||
# print('путь поиском в глубину:')
|
||
# for cell in path:
|
||
# print(cell.coords)
|
||
|
||
|
||
class BFS(PathFindingStrategy):
|
||
|
||
def findPath(self,maze,st,ex):
|
||
|
||
queue = deque([st])
|
||
|
||
self.visited = {st.coords} #по координатам надёжнее, а то вдруг адрес изменится
|
||
|
||
pathmap = {}
|
||
|
||
while queue:
|
||
cell = queue.popleft()
|
||
|
||
if cell.coords == ex.coords:
|
||
|
||
|
||
path = []
|
||
while cell.coords != st.coords:
|
||
path.append(cell)
|
||
cell = pathmap[cell.coords]
|
||
path.append(st)
|
||
path = path[::-1]
|
||
return path
|
||
|
||
for n in maze.getNeighbors(cell):
|
||
if n != None and n.coords not in self.visited:
|
||
self.visited.add(n.coords)
|
||
pathmap[n.coords] = cell
|
||
queue.append(n)
|
||
|
||
return None
|
||
|
||
# path = BFS().findPath(maze,maze.st,maze.ex)
|
||
|
||
# print('путь поиском в ширину:')
|
||
# for cell in path:
|
||
# print(cell.coords)
|
||
|
||
class Astar(PathFindingStrategy):
|
||
|
||
def findPath(self,maze,st,ex):
|
||
|
||
c = 0
|
||
hp_queue = [(0,c,st)]
|
||
|
||
self.g_score = {st.coords: 0}
|
||
|
||
pathmap = {}
|
||
|
||
hp_queue_coords = {st.coords} #нам важна скорость
|
||
|
||
while hp_queue:
|
||
|
||
cell = heapq.heappop(hp_queue)[2]
|
||
hp_queue_coords.remove(cell.coords)
|
||
|
||
if cell.coords == ex.coords:
|
||
|
||
path = []
|
||
while cell.coords != st.coords:
|
||
path.append(cell)
|
||
cell = pathmap[cell.coords]
|
||
path.append(st)
|
||
path = path[::-1]
|
||
self.visited = set(self.g_score.keys()) #экий костыль
|
||
return path
|
||
|
||
for n in maze.getNeighbors(cell):
|
||
new_g_score = self.g_score[cell.coords] + 1
|
||
|
||
if n is not None and new_g_score < self.g_score.get(n.coords, float('inf')):
|
||
pathmap[n.coords] = cell
|
||
self.g_score[n.coords] = new_g_score
|
||
|
||
h_score = abs(n.coords[0]-ex.coords[0]) + abs(n.coords[1]-ex.coords[1])
|
||
|
||
#f = g + h
|
||
#h - манхэттенское расстояние
|
||
full_score = new_g_score + h_score
|
||
|
||
if n.coords not in hp_queue_coords:
|
||
c += 1
|
||
heapq.heappush(hp_queue, (full_score, c, n))
|
||
hp_queue_coords.add(n.coords)
|
||
|
||
self.visited = set(self.g_score.keys()) #экий костыль 2: возвращение ситхов
|
||
return None
|
||
|
||
# path = Astar().findPath(maze,maze.st,maze.ex)
|
||
|
||
# print('путь с A*:')
|
||
# for cell in path:
|
||
# print(cell.coords)
|
||
|
||
#Класс статистики поиска пути и класс оркестратор
|
||
|
||
class SearchStats():
|
||
|
||
def __init__(self, timeMs, visitedCells, pathLength):
|
||
self.timeMs = timeMs
|
||
self.visitedCells = visitedCells
|
||
self.pathLength = pathLength
|
||
|
||
class MazeSolver():
|
||
|
||
def __init__(self, maze, strategy):
|
||
self.maze = maze
|
||
self.strategy = strategy
|
||
self.observers = [ConsoleView(maze)]
|
||
|
||
for observer in self.observers:
|
||
observer.update(MazeEvent('maze_loaded',maze,maze.st.coords))
|
||
|
||
def setStrategy(self,strategy):
|
||
self.strategy = strategy
|
||
|
||
def solve(self):
|
||
|
||
start = time.perf_counter()
|
||
path = self.strategy.findPath(self.maze,self.maze.st,self.maze.ex)
|
||
end = time.perf_counter()
|
||
|
||
elapsed = end - start
|
||
|
||
visitedCells = len(self.strategy.visited)
|
||
|
||
if path is not None:
|
||
pathLength = len(path)
|
||
|
||
for observer in self.observers:
|
||
observer.update(MazeEvent('path_found',self.maze,path[-1].coords,path))
|
||
|
||
else:
|
||
pathLength = 0
|
||
for observer in self.observers:
|
||
observer.update(MazeEvent('path_found',self.maze,None,path))
|
||
|
||
return SearchStats(elapsed*1000, visitedCells, pathLength)
|
||
|
||
# MS = MazeSolver(maze, DFS())
|
||
# Stats = MS.solve()
|
||
# print(Stats.timeMs)
|
||
# print(Stats.visitedCells)
|
||
# print(Stats.pathLength)
|
||
|
||
# MS = MazeSolver(maze, BFS())
|
||
# Stats = MS.solve()
|
||
# print(Stats.timeMs)
|
||
# print(Stats.visitedCells)
|
||
# print(Stats.pathLength)
|
||
|
||
# MS = MazeSolver(maze, Astar())
|
||
# Stats = MS.solve()
|
||
# print(Stats.timeMs)
|
||
# print(Stats.visitedCells)
|
||
# print(Stats.pathLength)
|
||
|
||
|
||
#Класс для событий
|
||
|
||
class MazeEvent():
|
||
|
||
def __init__(self,event_type, maze, player_position = None, path = []):
|
||
if player_position is None:
|
||
player_position = maze.st.coords
|
||
self.event_type = event_type
|
||
self.maze = maze
|
||
self.player_position = player_position
|
||
self.path = path
|
||
|
||
#Интерфейс наблюдатель
|
||
|
||
class Observer(abc.ABC):
|
||
|
||
@abc.abstractmethod
|
||
def update(self, event):
|
||
|
||
if not isinstance(event, (str, MazeEvent)):
|
||
raise TypeError('Только строки и объекты события')
|
||
|
||
elif isinstance(event, MazeEvent) and event.event_type not in ('path_found','move','maze_loaded'):
|
||
raise TypeError('Только события "path_found","move","maze_loaded"')
|
||
|
||
#Класс консольного просмотра
|
||
|
||
class ConsoleView(Observer):
|
||
|
||
def __init__(self, maze, player_position = (0,0), path = []):
|
||
|
||
self.maze = maze
|
||
self.player_position = player_position
|
||
self.path = path
|
||
|
||
def update(self, event):
|
||
|
||
super().update(event) #проверка через сам интерфейс
|
||
|
||
if isinstance(event, str):
|
||
print('')
|
||
print(event+'\n')
|
||
self.render(self.maze, self.player_position, self.path)
|
||
|
||
else:
|
||
print('')
|
||
print(event.event_type+'\n')
|
||
if event.player_position is not None:
|
||
self.player_position = event.player_position
|
||
if event.path is not None and event.path:
|
||
self.path = event.path
|
||
self.render(event.maze, self.player_position, self.path)
|
||
|
||
def render(self, maze, player_position, path):
|
||
|
||
os.system('cls' if os.name == 'nt' else 'clear')
|
||
|
||
#из-за системы координат надо всё опять транспонировать
|
||
|
||
res = []
|
||
for row in maze.cells.T[::-1]:
|
||
subres = []
|
||
for cell in row:
|
||
if cell.isWall:
|
||
subres += '#'
|
||
elif cell.isStart:
|
||
subres += 'S'
|
||
elif cell.isExit:
|
||
subres += 'E'
|
||
else:
|
||
subres += ' '
|
||
res.append(subres)
|
||
|
||
for cell in path:
|
||
x,y = cell.coords
|
||
if res[-(y+1)][x] != 'S':
|
||
res[-(y+1)][x] = '*'
|
||
|
||
res[-(player_position[1]+1)][player_position[0]] = 'X'
|
||
|
||
for row in res:
|
||
print(''.join(row))
|
||
|
||
# builder = TextFileMazeBuilder
|
||
|
||
# maze = builder.buildFromFile('maze1.txt')
|
||
|
||
# print(maze)
|
||
|
||
# CV = ConsoleView(maze, (0,0))
|
||
# CV.update('Что-то случилось')
|
||
|
||
# ME = MazeEvent('maze_loaded', maze, (0,0))
|
||
|
||
# CV.update(ME)
|
||
|
||
# CV.update('Что-то случилось')
|
||
|
||
#Интерфейс для команд
|
||
|
||
class Command(abc.ABC):
|
||
|
||
@abc.abstractmethod
|
||
def execute(self): pass
|
||
|
||
@abc.abstractmethod
|
||
def undo(self): pass
|
||
|
||
#Класс команды движения
|
||
|
||
class MoveCommand(Command):
|
||
|
||
def __init__(self):
|
||
self.previousCell = (0,0)
|
||
|
||
def execute(self,player,direction):
|
||
|
||
self.previousCell = player.currentCell
|
||
|
||
resCell = (self.previousCell[0]+direction.dir[0],self.previousCell[1]+direction.dir[1])
|
||
|
||
player.moveTo(resCell)
|
||
|
||
def undo(self,player):
|
||
|
||
player.moveTo(self.previousCell)
|
||
|
||
#Класс игрока
|
||
|
||
class Player():
|
||
|
||
#Он хранит не текущую клетку, а только её координаты. Поскольку
|
||
#нам надо перемещать игрока динамически, а команда для перемещения
|
||
#не принимает лабиринт в качестве аргумента, следующую клетку мы
|
||
#как объект получить не можем, а можем получить только её координаты.
|
||
|
||
def __init__(self, currentCell):
|
||
|
||
self.currentCell = currentCell
|
||
|
||
def moveTo(self, cell):
|
||
|
||
self.currentCell = cell
|
||
|
||
#Класс направление
|
||
|
||
class Direction():
|
||
|
||
def __init__(self, x,y):
|
||
self.dir = (x,y)
|
||
|
||
#Тест системы перемещения клавиатурой :D
|
||
|
||
builder = TextFileMazeBuilder
|
||
|
||
maze = builder.buildFromFile('maze1.txt')
|
||
|
||
MS = MazeSolver(maze, DFS())
|
||
|
||
MS.solve()
|
||
|
||
MC = MoveCommand()
|
||
|
||
CV = MS.observers[0]
|
||
|
||
player1 = Player(CV.player_position)
|
||
|
||
instruct = '\nПеремещайтесь на W/A/S/D. Для отмены используйте ctrl+Z. Для выхода из режима перемещения команда X.\n'
|
||
|
||
def move(player, direction):
|
||
|
||
resCoords = (player.currentCell[0]+direction.dir[0], player.currentCell[1]+direction.dir[1])
|
||
resCell = maze.getCell(resCoords[0], resCoords[1])
|
||
if resCell == None or resCell.isWall:
|
||
return
|
||
MC.execute(player, direction)
|
||
CV.update(MazeEvent('move', maze, player.currentCell))
|
||
print(instruct)
|
||
|
||
def undo(player):
|
||
|
||
MC.undo(player)
|
||
CV.update(MazeEvent('move', maze, player.currentCell))
|
||
|
||
print(instruct)
|
||
|
||
keyboard.add_hotkey('w', move, args=[player1, Direction(0,1)])
|
||
|
||
keyboard.add_hotkey('s', move, args=[player1, Direction(0,-1)])
|
||
|
||
keyboard.add_hotkey('a', move, args=[player1, Direction(-1,0)])
|
||
|
||
keyboard.add_hotkey('d', move, args=[player1, Direction(1,0)])
|
||
|
||
keyboard.add_hotkey('ctrl+z', undo, args=[player1])
|
||
|
||
keyboard.wait('x')
|
||
keyboard.unhook_all()
|
||
|
||
|
||
|
||
#Эксперимент
|
||
|
||
res = []
|
||
|
||
strategyList = [BFS(),DFS(),Astar()]
|
||
sNamesList = ['BFS','DFS','Astar']
|
||
labNamesList = ['10x10','50x50','100x100','empty','no exit']
|
||
for strategy in range(3):
|
||
for i in range(1,6):
|
||
|
||
subres1 = []
|
||
subres2 = []
|
||
subres3 = []
|
||
|
||
maze_name = 'expmaze' + str(i) + '.txt'
|
||
|
||
maze = TextFileMazeBuilder.buildFromFile(maze_name)
|
||
|
||
MS = MazeSolver(maze, strategyList[strategy])
|
||
|
||
for j in range(10):
|
||
Stats = MS.solve()
|
||
subres1.append(Stats.timeMs)
|
||
subres2.append(Stats.visitedCells)
|
||
subres3.append(Stats.pathLength)
|
||
|
||
res.append([labNamesList[i-1],sNamesList[strategy],sum(subres1)/10., sum(subres2)/10., sum(subres3)/10.])
|
||
|
||
print(res)
|
||
|
||
with open("results.csv", "w", newline="") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerows(res) |