2026-rff_mp/lab2/maze_solver.py

from abc import ABC, abstractmethod
from collections import deque
import heapq
import time

class Cell:
    def __init__(self, x, y):
        self.x = x
        self.y = y
        self.is_wall = False
        self.is_start = False
        self.is_exit = False

    def is_passable(self):
        return not self.is_wall

class Maze:
    def __init__(self, width, height):
        self.width = width
        self.height = height
        self.cells = []
        self.start = None
        self.exit = None

        for y in range(height):
            row = []

            for x in range(width):
                row.append(Cell(x, y))

            self.cells.append(row)

    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 = []

        directions = [
            (0, -1),
            (0, 1),
            (-1, 0),
            (1, 0)
        ]

        for dx, dy in directions:
            neighbor = self.get_cell(
                cell.x + dx,
                cell.y + dy
            )

            if neighbor and neighbor.is_passable():
                neighbors.append(neighbor)

        return neighbors

class MazeBuilder(ABC):
    @abstractmethod
    def build_from_file(self, filename):
        pass

class TextFileMazeBuilder(MazeBuilder):
    def build_from_file(self, filename):
        with open(filename, "r", encoding="utf-8") as file:
            lines = [line.rstrip("\n") for line in file]

        if not lines:
            raise ValueError("Файл лабиринта пустой")

        width = len(lines[0])

        for line in lines:
            if len(line) != width:
                raise ValueError("Строки лабиринта имеют разную длину")

        maze = Maze(width, len(lines))

        for y, line in enumerate(lines):
            for x, symbol in enumerate(line):
                cell = maze.get_cell(x, y)

                if symbol == "#":
                    cell.is_wall = True

                elif symbol == "S":
                    if maze.start is not None:
                        raise ValueError("В лабиринте несколько стартов")

                    maze.start = cell
                    cell.is_start = True

                elif symbol == "E":
                    if maze.exit is not None:
                        raise ValueError("В лабиринте несколько выходов")

                    maze.exit = cell
                    cell.is_exit = True

                elif symbol == " ":
                    pass

                else:
                    raise ValueError("Неизвестный символ в лабиринте")

        if maze.start is None:
            raise ValueError("В лабиринте нет старта")

        if maze.exit is None:
            raise ValueError("В лабиринте нет выхода")

        return maze

class PathFindingStrategy(ABC):
    @abstractmethod
    def find_path(self, maze, start, exit):
        pass

class BFSStrategy(PathFindingStrategy):
    def find_path(self, maze, start, exit):
        if start is None or exit is None:
            return [], 0

        queue = deque([(start, [start])])
        visited = {start}

        while queue:
            current, path = queue.popleft()

            if current == exit:
                return path, len(visited)

            for neighbor in maze.get_neighbors(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    queue.append((neighbor, path + [neighbor]))

        return [], len(visited)

class DFSStrategy(PathFindingStrategy):
    def find_path(self, maze, start, exit):
        if start is None or exit is None:
            return [], 0

        stack = [(start, [start])]
        visited = {start}

        while stack:
            current, path = stack.pop()

            if current == exit:
                return path, len(visited)

            for neighbor in maze.get_neighbors(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    stack.append((neighbor, path + [neighbor]))

        return [], len(visited)

class AStarStrategy(PathFindingStrategy):
    def heuristic(self, a, b):
        return abs(a.x - b.x) + abs(a.y - b.y)

    def find_path(self, maze, start, exit):
        if start is None or exit is None:
            return [], 0

        heap = []
        counter = 0

        heapq.heappush(
            heap,
            (self.heuristic(start, exit), counter, start, [start])
        )

        g_score = {start: 0}
        visited = set()

        while heap:
            _, _, current, path = heapq.heappop(heap)

            if current in visited:
                continue

            visited.add(current)

            if current == exit:
                return path, len(visited)

            for neighbor in maze.get_neighbors(current):
                new_cost = g_score[current] + 1

                if neighbor not in g_score or new_cost < g_score[neighbor]:
                    g_score[neighbor] = new_cost
                    counter += 1

                    priority = new_cost + self.heuristic(
                        neighbor,
                        exit
                    )

                    heapq.heappush(
                        heap,
                        (
                            priority,
                            counter,
                            neighbor,
                            path + [neighbor]
                        )
                    )

        return [], len(visited)

class SearchStats:
    def __init__(self, path, time_ms, visited_count):
        self.path = path
        self.time_ms = time_ms
        self.visited_count = visited_count
        self.path_length = len(path) if path else 0

class MazeSolver:
    def __init__(self, maze, strategy=None):
        self.maze = maze
        self.strategy = strategy
        self.observers = []

    def attach(self, observer):
        self.observers.append(observer)

    def detach(self, observer):
        self.observers.remove(observer)

    def notify(self, event, data=None):
        for observer in self.observers:
            observer.update(event, data)

    def set_strategy(self, strategy):
        self.strategy = strategy

    def solve(self):
        if self.strategy is None:
            raise ValueError("Стратегия не установлена")

        self.notify("search_started")

        start_time = time.perf_counter()

        path, visited_count = self.strategy.find_path(
            self.maze,
            self.maze.start,
            self.maze.exit
        )

        end_time = time.perf_counter()

        time_ms = (end_time - start_time) * 1000

        self.notify("search_finished", time_ms)
        self.notify("path_found", path)

        return SearchStats(
            path,
            time_ms,
            visited_count
        )

class Observer(ABC):
    @abstractmethod
    def update(self, event, data=None):
        pass

class ConsoleView(Observer):
    def update(self, event, data=None):
        if event == "search_started":
            print("Поиск начат")

        elif event == "search_finished":
            print(f"Поиск завершен за {data:.3f} мс")

        elif event == "path_found":
            print(f"Длина пути: {len(data)}")