2026-rff_mp/AgapovaDS/docs/data/2-nd/maze.py

import time
import random
import csv
import os
from collections import deque
import heapq
import matplotlib.pyplot as plt
import numpy as np

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):
        self.width = width
        self.height = height
        self.grid = [[Cell(x, y) for y in range(height)] for x in range(width)]
        self.start_cell = None
        self.exit_cell = None

    def get_cell(self, x, y):
        if 0 <= x < self.width and 0 <= y < self.height:
            return self.grid[x][y]
        return None

    def get_neighbors(self, cell):
        neighbors = []
        for dx, dy in [(-1,0), (1,0), (0,-1), (0,1)]:
            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

    def save_to_file(self, filename):
        """Сохраняет лабиринт в текстовый файл."""
        with open(filename, 'w', encoding='utf-8') as f:
            for y in range(self.height):
                row = ''
                for x in range(self.width):
                    cell = self.get_cell(x, y)
                    if cell.is_wall:
                        row += '#'
                    elif cell.is_start:
                        row += 'S'
                    elif cell.is_exit:
                        row += 'E'
                    else:
                        row += ' '
                f.write(row + '\n')



class MazeBuilder:
    def build_from_file(self, filename):
        raise NotImplementedError


class TextFileMazeBuilder(MazeBuilder):
    def build_from_file(self, filename):
        with open(filename, 'r', encoding='utf-8') as f:
            lines = f.readlines()
        lines = [line.rstrip('\n') for line in lines if line.strip() != '']
        if not lines:
            raise ValueError("Файл пуст")
        height = len(lines)
        width = max(len(line) for line in lines)
        maze = Maze(width, height)
        for y, line in enumerate(lines):
            for x, ch in enumerate(line):
                if x >= width:
                    break
                cell = maze.get_cell(x, y)
                if ch == '#':
                    cell.is_wall = True
                elif ch == 'S':
                    cell.is_start = True
                    maze.start_cell = cell
                elif ch == 'E':
                    cell.is_exit = True
                    maze.exit_cell = cell
        if maze.start_cell is None or maze.exit_cell is None:
            raise ValueError("В лабиринте должны быть S и E")
        return maze


class PathFindingStrategy:
    def find_path(self, maze, start, exit):
        raise NotImplementedError


class BFSStrategy(PathFindingStrategy):
    def find_path(self, maze, start, exit):
        if start == exit:
            return [start]
        queue = deque([start])
        visited = {start}
        parent = {start: None}
        while queue:
            current = queue.popleft()
            if current == exit:
                break
            for neighbor in maze.get_neighbors(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    parent[neighbor] = current
                    queue.append(neighbor)
        if exit not in parent:
            return []
        path = []
        step = exit
        while step is not None:
            path.append(step)
            step = parent[step]
        path.reverse()
        return path


class DFSStrategy(PathFindingStrategy):
    def find_path(self, maze, start, exit):
        if start == exit:
            return [start]
        stack = [start]
        visited = {start}
        parent = {start: None}
        while stack:
            current = stack.pop()
            if current == exit:
                break
            for neighbor in maze.get_neighbors(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    parent[neighbor] = current
                    stack.append(neighbor)
        if exit not in parent:
            return []
        path = []
        step = exit
        while step is not None:
            path.append(step)
            step = parent[step]
        path.reverse()
        return path


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 == exit:
            return [start]
        open_set = []
        heapq.heappush(open_set, (0, id(start), start))
        came_from = {}
        g_score = {start: 0}
        f_score = {start: self.heuristic(start, exit)}
        while open_set:
            _, _, current = heapq.heappop(open_set)
            if current == exit:
                path = []
                step = current
                while step is not None:
                    path.append(step)
                    step = came_from.get(step)
                path.reverse()
                return path
            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]:
                    came_from[neighbor] = current
                    g_score[neighbor] = tentative_g
                    f_score[neighbor] = tentative_g + self.heuristic(neighbor, exit)
                    heapq.heappush(open_set, (f_score[neighbor], id(neighbor), neighbor))
        return []


class Observer:
    def update(self, event):
        raise NotImplementedError


class ConsoleView(Observer):
    def __init__(self, maze):
        self.maze = maze

    def update(self, event):
        if event == "Поиск начат":
            print("=== Поиск начат ===")
        elif event == "Поиск завершён":
            print("=== Поиск завершён ===")

    def render(self, path=None):
        path_set = set(path) if 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 cell.is_wall:
                    row += '#'
                elif cell.is_start:
                    row += 'S'
                elif cell.is_exit:
                    row += 'E'
                elif cell in path_set:
                    row += '*'
                else:
                    row += ' '
            print(row)
        print()


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

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

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

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

    def notify(self, event):
        for obs in self.observers:
            obs.update(event)

    def solve(self):
        if self.strategy is None:
            raise ValueError("Стратегия не установлена")
        start = self.maze.start_cell
        exit_cell = self.maze.exit_cell
        if start is None or exit_cell is None:
            raise ValueError("Лабиринт не содержит старта или выхода")
        self.notify("Поиск начат")
        start_time = time.perf_counter()
        path = self.strategy.find_path(self.maze, start, exit_cell)
        end_time = time.perf_counter()
        elapsed_ms = (end_time - start_time) * 1000
        self.notify("Поиск завершён")
        return path, elapsed_ms


def generate_empty_maze(width, height):
    maze = Maze(width, height)
    start = maze.get_cell(0, 0)
    exit_cell = maze.get_cell(width-1, height-1)
    start.is_start = True
    exit_cell.is_exit = True
    maze.start_cell = start
    maze.exit_cell = exit_cell
    return maze


def generate_random_maze(width, height, wall_prob=0.3):
    maze = Maze(width, height)
    for x in range(width):
        for y in range(height):
            cell = maze.get_cell(x, y)
            if random.random() < wall_prob:
                cell.is_wall = True
    start = maze.get_cell(0, 0)
    exit_cell = maze.get_cell(width-1, height-1)
    start.is_wall = False
    start.is_start = True
    exit_cell.is_wall = False
    exit_cell.is_exit = True
    maze.start_cell = start
    maze.exit_cell = exit_cell
    return maze


def generate_maze_with_dead_ends(width, height):
    maze = Maze(width, height)
    for x in range(width):
        for y in range(height):
            maze.get_cell(x, y).is_wall = True
    x, y = 0, 0
    while x < width and y < height:
        cell = maze.get_cell(x, y)
        cell.is_wall = False
        if x == width-1 and y == height-1:
            break
        if y+1 < height and (x == width-1 or random.choice([True, False])):
            y += 1
        else:
            x += 1
    start = maze.get_cell(0, 0)
    exit_cell = maze.get_cell(width-1, height-1)
    start.is_start = True
    exit_cell.is_exit = True
    maze.start_cell = start
    maze.exit_cell = exit_cell
    return maze


def generate_maze_no_exit(width, height):
    maze = generate_random_maze(width, height, 0.2)
    exit_cell = maze.get_cell(width-1, height-1)
    for dx, dy in [(-1,0), (1,0), (0,-1), (0,1)]:
        nx, ny = exit_cell.x + dx, exit_cell.y + dy
        neighbor = maze.get_cell(nx, ny)
        if neighbor:
            neighbor.is_wall = True
    start = maze.get_cell(0, 0)
    start.is_wall = False
    start.is_start = True
    maze.start_cell = start
    maze.exit_cell = exit_cell
    return maze


def ensure_maze_files():
    """Создаёт папку mazes и генерирует все лабиринты, если файлы отсутствуют."""
    os.makedirs("mazes", exist_ok=True)
    configs = [
        ("empty_10x10.txt", 10, 10, generate_empty_maze),
        ("empty_50x50.txt", 50, 50, generate_empty_maze),
        ("empty_100x100.txt", 100, 100, generate_empty_maze),
        ("random_10x10.txt", 10, 10, generate_random_maze),
        ("random_50x50.txt", 50, 50, generate_random_maze),
        ("random_100x100.txt", 100, 100, generate_random_maze),
        ("deadends_10x10.txt", 10, 10, generate_maze_with_dead_ends),
        ("deadends_50x50.txt", 50, 50, generate_maze_with_dead_ends),
        ("deadends_100x100.txt", 100, 100, generate_maze_with_dead_ends),
        ("no_exit_10x10.txt", 10, 10, generate_maze_no_exit),
        ("no_exit_50x50.txt", 50, 50, generate_maze_no_exit),
    ]
    for filename, w, h, gen_func in configs:
        filepath = os.path.join("mazes", filename)
        if not os.path.exists(filepath):
            print(f"Генерация {filename}...")
            maze = gen_func(w, h)
            maze.save_to_file(filepath)


def run_experiment():
    ensure_maze_files()
    builder = TextFileMazeBuilder()
    strategies = [
        ("BFS", BFSStrategy()),
        ("DFS", DFSStrategy()),
        ("AStar", AStarStrategy())
    ]
    maze_files = sorted([f for f in os.listdir("mazes") if f.endswith(".txt")])
    results = []
    repeats = 5

    for filename in maze_files:
        maze_name = filename.replace(".txt", "")
        print(f"Тестирование лабиринта: {maze_name}")
        try:
            maze = builder.build_from_file(os.path.join("mazes", filename))
        except Exception as e:
            print(f"Ошибка загрузки {filename}: {e}")
            continue

        solver = MazeSolver(maze)
        for strat_name, strat in strategies:
            solver.set_strategy(strat)
            total_time = 0
            total_path_len = 0
            path = []
            for rep in range(repeats):
                path, elapsed_ms = solver.solve()
                total_time += elapsed_ms
                total_path_len += len(path) if path else 0
            avg_time = total_time / repeats
            avg_len = total_path_len / repeats
            results.append({
                "Maze": maze_name,
                "Strategy": strat_name,
                "AvgTime_ms": avg_time,
                "AvgPathLen": avg_len,
                "PathFound": len(path) > 0 if path else False
            })
            print(f"  {strat_name}: время {avg_time:.3f} мс, длина пути {avg_len:.1f}")

    os.makedirs("results", exist_ok=True)
    csv_path = "results/experiment_results.csv"
    with open(csv_path, 'w', newline='', encoding='utf-8') as f:
        fieldnames = ["Maze", "Strategy", "AvgTime_ms", "AvgPathLen", "PathFound"]
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)
    print(f"Результаты сохранены в {csv_path}")

    maze_names = sorted(set(r["Maze"] for r in results))
    strategy_names = ["BFS", "DFS", "AStar"]
    data = {maze: {s: None for s in strategy_names} for maze in maze_names}
    for r in results:
        data[r["Maze"]][r["Strategy"]] = r["AvgTime_ms"]

    fig, ax = plt.subplots(figsize=(14, 6))
    x = np.arange(len(maze_names))
    width = 0.25
    colors = ['skyblue', 'lightgreen', 'salmon']

    for i, strat in enumerate(strategy_names):
        times = [data[maze][strat] if data[maze][strat] is not None else 0 for maze in maze_names]
        ax.bar(x + i*width, times, width, label=strat, color=colors[i])

    ax.set_xlabel('Лабиринт')
    ax.set_ylabel('Среднее время (мс)')
    ax.set_title('Сравнение стратегий поиска пути')
    ax.set_xticks(x + width)
    ax.set_xticklabels(maze_names, rotation=45, ha='right')
    ax.legend()

    plt.tight_layout()
    plt.savefig("results/performance.png", dpi=150)
    #plt.show()
    print("График сохранён в results/performance.png")


if __name__ == '__main__':
    run_experiment()