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
|