forked from UNN/2026-rff_mp
247 lines
7.8 KiB
Python
247 lines
7.8 KiB
Python
|
|
"""Модель лабиринта и алгоритмы поиска с паттернами Builder, Strategy, Observer."""
|
|||
|
|
|
|||
|
|
from abc import ABC, abstractmethod
|
|||
|
|
from collections import deque
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from heapq import heappop, heappush
|
|||
|
|
from itertools import count
|
|||
|
|
from pathlib import Path
|
|||
|
|
from time import perf_counter
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(frozen=True)
|
|||
|
|
class Cell:
|
|||
|
|
x: int
|
|||
|
|
y: int
|
|||
|
|
is_wall: bool = False
|
|||
|
|
is_start: bool = False
|
|||
|
|
is_exit: bool = False
|
|||
|
|
|
|||
|
|
def is_passable(self):
|
|||
|
|
return not self.is_wall
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Maze:
|
|||
|
|
def __init__(self, cells):
|
|||
|
|
if not cells or not cells[0]:
|
|||
|
|
raise ValueError("Лабиринт не может быть пустым")
|
|||
|
|
self.cells = cells
|
|||
|
|
self.height = len(cells)
|
|||
|
|
self.width = len(cells[0])
|
|||
|
|
self.start = next((cell for row in cells for cell in row if cell.is_start), None)
|
|||
|
|
self.exit = next((cell for row in cells for cell in row if cell.is_exit), None)
|
|||
|
|
|
|||
|
|
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), (1, 0), (0, 1), (-1, 0)):
|
|||
|
|
neighbor = self.get_cell(cell.x + dx, cell.y + dy)
|
|||
|
|
if neighbor is not None and neighbor.is_passable():
|
|||
|
|
neighbors.append(neighbor)
|
|||
|
|
return neighbors
|
|||
|
|
|
|||
|
|
|
|||
|
|
class MazeBuilder(ABC):
|
|||
|
|
@abstractmethod
|
|||
|
|
def build_from_file(self, filename):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TextFileMazeBuilder(MazeBuilder):
|
|||
|
|
SYMBOLS = {"#", " ", "S", "E"}
|
|||
|
|
|
|||
|
|
def build_from_file(self, filename):
|
|||
|
|
lines = Path(filename).read_text(encoding="utf-8").splitlines()
|
|||
|
|
if not lines or not lines[0] or any(len(line) != len(lines[0]) for line in lines):
|
|||
|
|
raise ValueError("Строки лабиринта должны иметь одинаковую ненулевую длину")
|
|||
|
|
unknown = {character for line in lines for character in line} - self.SYMBOLS
|
|||
|
|
if unknown:
|
|||
|
|
raise ValueError(f"Недопустимые символы: {sorted(unknown)}")
|
|||
|
|
if sum(line.count("S") for line in lines) != 1 or sum(line.count("E") for line in lines) != 1:
|
|||
|
|
raise ValueError("Лабиринт должен содержать ровно один старт S и один выход E")
|
|||
|
|
|
|||
|
|
cells = []
|
|||
|
|
for y, line in enumerate(lines):
|
|||
|
|
cells.append([
|
|||
|
|
Cell(x, y, symbol == "#", symbol == "S", symbol == "E")
|
|||
|
|
for x, symbol in enumerate(line)
|
|||
|
|
])
|
|||
|
|
return Maze(cells)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class PathFindingStrategy(ABC):
|
|||
|
|
name = "Неизвестно"
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.visited_count = 0
|
|||
|
|
|
|||
|
|
@abstractmethod
|
|||
|
|
def find_path(self, maze, start, exit_cell):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def restore_path(parent, start, exit_cell):
|
|||
|
|
if exit_cell not in parent:
|
|||
|
|
return []
|
|||
|
|
path, current = [], exit_cell
|
|||
|
|
while current is not None:
|
|||
|
|
path.append(current)
|
|||
|
|
current = parent[current]
|
|||
|
|
path.reverse()
|
|||
|
|
return path if path and path[0] == start else []
|
|||
|
|
|
|||
|
|
|
|||
|
|
class BFSStrategy(PathFindingStrategy):
|
|||
|
|
name = "BFS"
|
|||
|
|
|
|||
|
|
def find_path(self, maze, start, exit_cell):
|
|||
|
|
queue = deque([start])
|
|||
|
|
parent = {start: None}
|
|||
|
|
visited = 0
|
|||
|
|
while queue:
|
|||
|
|
current = queue.popleft()
|
|||
|
|
visited += 1
|
|||
|
|
if current == exit_cell:
|
|||
|
|
break
|
|||
|
|
for neighbor in maze.get_neighbors(current):
|
|||
|
|
if neighbor not in parent:
|
|||
|
|
parent[neighbor] = current
|
|||
|
|
queue.append(neighbor)
|
|||
|
|
self.visited_count = visited
|
|||
|
|
return self.restore_path(parent, start, exit_cell)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class DFSStrategy(PathFindingStrategy):
|
|||
|
|
name = "DFS"
|
|||
|
|
|
|||
|
|
def find_path(self, maze, start, exit_cell):
|
|||
|
|
stack = [start]
|
|||
|
|
parent = {start: None}
|
|||
|
|
visited = 0
|
|||
|
|
while stack:
|
|||
|
|
current = stack.pop()
|
|||
|
|
visited += 1
|
|||
|
|
if current == exit_cell:
|
|||
|
|
break
|
|||
|
|
for neighbor in maze.get_neighbors(current):
|
|||
|
|
if neighbor not in parent:
|
|||
|
|
parent[neighbor] = current
|
|||
|
|
stack.append(neighbor)
|
|||
|
|
self.visited_count = visited
|
|||
|
|
return self.restore_path(parent, start, exit_cell)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class AStarStrategy(PathFindingStrategy):
|
|||
|
|
name = "A*"
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def heuristic(first, second):
|
|||
|
|
return abs(first.x - second.x) + abs(first.y - second.y)
|
|||
|
|
|
|||
|
|
def find_path(self, maze, start, exit_cell):
|
|||
|
|
order = count()
|
|||
|
|
queue = [(self.heuristic(start, exit_cell), next(order), start)]
|
|||
|
|
distance = {start: 0}
|
|||
|
|
parent = {start: None}
|
|||
|
|
closed = set()
|
|||
|
|
while queue:
|
|||
|
|
_, _, current = heappop(queue)
|
|||
|
|
if current in closed:
|
|||
|
|
continue
|
|||
|
|
closed.add(current)
|
|||
|
|
if current == exit_cell:
|
|||
|
|
break
|
|||
|
|
for neighbor in maze.get_neighbors(current):
|
|||
|
|
new_distance = distance[current] + 1
|
|||
|
|
if new_distance < distance.get(neighbor, float("inf")):
|
|||
|
|
distance[neighbor] = new_distance
|
|||
|
|
parent[neighbor] = current
|
|||
|
|
priority = new_distance + self.heuristic(neighbor, exit_cell)
|
|||
|
|
heappush(queue, (priority, next(order), neighbor))
|
|||
|
|
self.visited_count = len(closed)
|
|||
|
|
return self.restore_path(parent, start, exit_cell)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class SearchStats:
|
|||
|
|
strategy: str
|
|||
|
|
time_ms: float
|
|||
|
|
visited_cells: int
|
|||
|
|
path_length: int
|
|||
|
|
path: list = field(repr=False)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Observer(ABC):
|
|||
|
|
@abstractmethod
|
|||
|
|
def update(self, event):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ConsoleView(Observer):
|
|||
|
|
def __init__(self, verbose=True):
|
|||
|
|
self.verbose = verbose
|
|||
|
|
self.events = []
|
|||
|
|
|
|||
|
|
def update(self, event):
|
|||
|
|
self.events.append(event)
|
|||
|
|
if self.verbose:
|
|||
|
|
print(event["message"])
|
|||
|
|
|
|||
|
|
def render(self, maze, path=None):
|
|||
|
|
path_cells = set(path or [])
|
|||
|
|
rows = []
|
|||
|
|
for row in maze.cells:
|
|||
|
|
symbols = []
|
|||
|
|
for cell in row:
|
|||
|
|
if cell.is_start:
|
|||
|
|
symbols.append("S")
|
|||
|
|
elif cell.is_exit:
|
|||
|
|
symbols.append("E")
|
|||
|
|
elif cell.is_wall:
|
|||
|
|
symbols.append("#")
|
|||
|
|
elif cell in path_cells:
|
|||
|
|
symbols.append(".")
|
|||
|
|
else:
|
|||
|
|
symbols.append(" ")
|
|||
|
|
rows.append("".join(symbols))
|
|||
|
|
return "\n".join(rows)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class MazeSolver:
|
|||
|
|
def __init__(self, maze, strategy):
|
|||
|
|
self.maze = maze
|
|||
|
|
self.strategy = strategy
|
|||
|
|
self.observers = []
|
|||
|
|
|
|||
|
|
def set_strategy(self, strategy):
|
|||
|
|
self.strategy = strategy
|
|||
|
|
|
|||
|
|
def attach(self, observer):
|
|||
|
|
if observer not in self.observers:
|
|||
|
|
self.observers.append(observer)
|
|||
|
|
|
|||
|
|
def notify(self, event):
|
|||
|
|
for observer in self.observers:
|
|||
|
|
observer.update(event)
|
|||
|
|
|
|||
|
|
def solve(self):
|
|||
|
|
self.notify({"type": "search_started", "message": f"Запущен алгоритм {self.strategy.name}"})
|
|||
|
|
started = perf_counter()
|
|||
|
|
path = self.strategy.find_path(self.maze, self.maze.start, self.maze.exit)
|
|||
|
|
elapsed_ms = max((perf_counter() - started) * 1000, 1e-9)
|
|||
|
|
stats = SearchStats(
|
|||
|
|
strategy=self.strategy.name,
|
|||
|
|
time_ms=elapsed_ms,
|
|||
|
|
visited_cells=self.strategy.visited_count,
|
|||
|
|
path_length=max(len(path) - 1, 0),
|
|||
|
|
path=path,
|
|||
|
|
)
|
|||
|
|
event_type = "path_found" if path else "path_not_found"
|
|||
|
|
message = f"{self.strategy.name}: путь длиной {stats.path_length}" if path else f"{self.strategy.name}: путь не найден"
|
|||
|
|
self.notify({"type": event_type, "message": message, "stats": stats})
|
|||
|
|
return stats
|