2026-rff_mp/lukovnikovde/docs/data/2-nd-exercise/main.py

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

# ---------- 1. Модель клетки и лабиринта ----------
class Tile:
    def __init__(self, x, y):
        self.x = x
        self.y = y
        self.wall = False
        self.start = False
        self.exit = False

    def passable(self):
        return not self.wall


class Labyrinth:
    def __init__(self, width, height):
        self.width = width
        self.height = height
        self._tiles = [[Tile(x, y) for x in range(width)] for y in range(height)]
        self.start_tile = None
        self.exit_tile = None

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

    def set_tile(self, x, y, kind):
        tile = self.get_tile(x, y)
        if tile is None:
            return
        if kind == 'wall':
            tile.wall = True
        elif kind == 'start':
            if self.start_tile:
                self.start_tile.start = False
            tile.start = True
            tile.wall = False
            self.start_tile = tile
        elif kind == 'exit':
            if self.exit_tile:
                self.exit_tile.exit = False
            tile.exit = True
            tile.wall = False
            self.exit_tile = tile
        elif kind == 'path':
            tile.wall = False

    def neighbours(self, tile):
        dirs = [(0, -1), (0, 1), (-1, 0), (1, 0)]
        result = []
        for dx, dy in dirs:
            nx, ny = tile.x + dx, tile.y + dy
            nb = self.get_tile(nx, ny)
            if nb and nb.passable():
                result.append(nb)
        return result


# ---------- 2. Загрузка лабиринта из файла ----------
class LabyrinthLoader:
    def load(self, filename):
        raise NotImplementedError


class TextLabyrinthLoader(LabyrinthLoader):
    def load(self, filename):
        with open(filename, 'r', encoding='utf-8') as f:
            lines = [line.rstrip('\n') for line in f]
        h = len(lines)
        w = max(len(line) for line in lines) if h > 0 else 0
        start_cnt = 0
        exit_cnt = 0
        lab = Labyrinth(w, h)

        for y, line in enumerate(lines):
            for x, ch in enumerate(line):
                if ch == '#':
                    lab.set_tile(x, y, 'wall')
                elif ch == 'S':
                    lab.set_tile(x, y, 'start')
                    start_cnt += 1
                elif ch == 'E':
                    lab.set_tile(x, y, 'exit')
                    exit_cnt += 1
                else:
                    lab.set_tile(x, y, 'path')
        if start_cnt != 1 or exit_cnt != 1:
            raise ValueError(f"Нужны ровно S и E, найдено S={start_cnt}, E={exit_cnt}")
        return lab


# ---------- 3. Алгоритмы поиска пути (стратегии) ----------
class Pathfinder:
    def find_path(self, lab, start, goal):
        raise NotImplementedError

    def _build_path(self, came_from, start, goal):
        path = []
        cur = goal
        while cur is not None:
            path.append(cur)
            cur = came_from.get(cur)
        path.reverse()
        return path

    def visited_count(self):
        return getattr(self, '_visited', 0)


class BFS(Pathfinder):
    def find_path(self, lab, start, goal):
        q = deque([start])
        parent = {start: None}
        visited = {start}
        while q:
            cur = q.popleft()
            if cur == goal:
                self._visited = len(visited)
                return self._build_path(parent, start, goal)
            for nb in lab.neighbours(cur):
                if nb not in visited:
                    visited.add(nb)
                    parent[nb] = cur
                    q.append(nb)
        self._visited = len(visited)
        return []


class DFS(Pathfinder):
    def find_path(self, lab, start, goal):
        stack = [start]
        parent = {start: None}
        visited = {start}
        while stack:
            cur = stack.pop()
            if cur == goal:
                self._visited = len(visited)
                return self._build_path(parent, start, goal)
            for nb in lab.neighbours(cur):
                if nb not in visited:
                    visited.add(nb)
                    parent[nb] = cur
                    stack.append(nb)
        self._visited = len(visited)
        return []


class AStar(Pathfinder):
    def _heuristic(self, a, b):
        return abs(a.x - b.x) + abs(a.y - b.y)

    def find_path(self, lab, start, goal):
        heap = []
        counter = 0
        start_f = self._heuristic(start, goal)
        heapq.heappush(heap, (start_f, counter, start))
        counter += 1
        parent = {}
        g = {start: 0}
        f = {start: start_f}
        visited = set()
        while heap:
            cur_f, _, cur = heapq.heappop(heap)
            visited.add(cur)
            if cur == goal:
                self._visited = len(visited)
                return self._build_path(parent, start, goal)
            if cur_f > f.get(cur, float('inf')):
                continue
            for nb in lab.neighbours(cur):
                new_g = g[cur] + 1
                if new_g < g.get(nb, float('inf')):
                    parent[nb] = cur
                    g[nb] = new_g
                    new_f = new_g + self._heuristic(nb, goal)
                    f[nb] = new_f
                    heapq.heappush(heap, (new_f, counter, nb))
                    counter += 1
        self._visited = len(visited)
        return []


# ---------- 4. Оркестратор с поддержкой наблюдателей ----------
class LabyrinthSolver:
    def __init__(self, lab):
        self.lab = lab
        self.strategy = None
        self.observers = []

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

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

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

    def solve(self):
        if not self.strategy:
            return None
        t0 = time.perf_counter()
        path = self.strategy.find_path(self.lab, self.lab.start_tile, self.lab.exit_tile)
        t1 = time.perf_counter()
        self.notify("path_found", path)
        return {
            'time_ms': (t1 - t0) * 1000,
            'visited': self.strategy.visited_count(),
            'length': len(path)
        }


# ---------- 5. Визуализация (наблюдатель) ----------
class EventListener:
    def notify(self, event, data):
        raise NotImplementedError


class ConsoleRenderer(EventListener):
    def __init__(self, walker=None):
        self.last_path = None
        self.walker = walker

    def notify(self, event, data):
        if event == "maze_loaded":
            self._draw_maze(data)
        elif event == "path_found":
            self.last_path = data
            self._show_path_info(data)
        elif event == "player_moved":
            self._draw_maze_with_player(data)

    def _draw_maze(self, lab):
        os.system('cls' if os.name == 'nt' else 'clear')
        print("=" * (lab.width * 2 + 4))
        print("   ЛАБИРИНТ")
        print("=" * (lab.width * 2 + 4))
        for y in range(lab.height):
            print("  ", end='')
            for x in range(lab.width):
                t = lab.get_tile(x, y)
                if t == lab.start_tile:
                    print('S', end=' ')
                elif t == lab.exit_tile:
                    print('E', end=' ')
                elif t.wall:
                    print('#', end=' ')
                else:
                    print('.', end=' ')
            print()
        print("=" * (lab.width * 2 + 4))
        print("  S - старт  E - выход  # - стена  . - проход")

    def _draw_maze_with_player(self, lab):
        os.system('cls' if os.name == 'nt' else 'clear')
        print("=" * (lab.width * 2 + 4))
        print("   ЛАБИРИНТ (игрок = P)")
        print("=" * (lab.width * 2 + 4))
        for y in range(lab.height):
            print("  ", end='')
            for x in range(lab.width):
                t = lab.get_tile(x, y)
                if self.walker and t == self.walker.current:
                    print('P', end=' ')
                elif t == lab.start_tile:
                    print('S', end=' ')
                elif t == lab.exit_tile:
                    print('E', end=' ')
                elif t.wall:
                    print('#', end=' ')
                else:
                    print('.', end=' ')
            print()
        print("=" * (lab.width * 2 + 4))
        if self.walker:
            print(f"  Позиция игрока: ({self.walker.current.x}, {self.walker.current.y})")

    def _show_path_info(self, path):
        if not path:
            print("\n  Путь не найден!")
        else:
            print(f"\n  Путь найден! Длина = {len(path)}")


# ---------- 6. Игрок и команды с отменой ----------
class Walker:
    def __init__(self, start_tile, lab):
        self.current = start_tile
        self.prev = None
        self.lab = lab

    def move(self, target_tile):
        if target_tile and target_tile.passable():
            self.prev = self.current
            self.current = target_tile
            return True
        return False

    def undo(self):
        if self.prev:
            self.current, self.prev = self.prev, None
            return True
        return False


class Action:
    def do(self):
        raise NotImplementedError
    def undo(self):
        raise NotImplementedError


class MoveAction(Action):
    def __init__(self, walker, dx, dy, lab):
        self.walker = walker
        self.dx = dx
        self.dy = dy
        self.lab = lab
        self.done = False

    def do(self):
        new_x = self.walker.current.x + self.dx
        new_y = self.walker.current.y + self.dy
        target = self.lab.get_tile(new_x, new_y)
        if target and target.passable():
            self.walker.move(target)
            self.done = True
            return True
        return False

    def undo(self):
        if self.done:
            self.walker.undo()
            self.done = False
            return True
        return False


# ---------- 7. Экспериментальные функции ----------
def run_benchmark(maze_file, strategy, runs=5):
    loader = TextLabyrinthLoader()
    lab = loader.load(maze_file)
    total_time = 0.0
    total_visited = 0
    total_len = 0
    for _ in range(runs):
        solver = LabyrinthSolver(lab)
        solver.set_strategy(strategy)
        stats = solver.solve()
        if stats:
            total_time += stats['time_ms']
            total_visited += stats['visited']
            total_len += stats['length']
    return {
        'time_ms': total_time / runs,
        'visited': total_visited / runs,
        'length': total_len / runs
    }


def make_plots(results):
    mazes = list({r['maze'] for r in results})
    algos = ['BFS', 'DFS', 'AStar']
    fig, axes = plt.subplots(1, 3, figsize=(15, 5))
    x = np.arange(len(mazes))
    width = 0.25

    for i, algo in enumerate(algos):
        times = []
        for m in mazes:
            val = next((r['time_ms'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
            times.append(val)
        axes[0].bar(x + i*width, times, width, label=algo)
    axes[0].set_xlabel('Лабиринт')
    axes[0].set_ylabel('Время (мс)')
    axes[0].set_title('Сравнение времени выполнения')
    axes[0].set_xticks(x + width)
    axes[0].set_xticklabels(mazes, rotation=45, ha='right')
    axes[0].legend()
    axes[0].grid(True, alpha=0.3)

    for i, algo in enumerate(algos):
        visited_vals = []
        for m in mazes:
            val = next((r['visited'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
            visited_vals.append(val)
        axes[1].bar(x + i*width, visited_vals, width, label=algo)
    axes[1].set_xlabel('Лабиринт')
    axes[1].set_ylabel('Посещено клеток')
    axes[1].set_title('Сравнение посещённых клеток')
    axes[1].set_xticks(x + width)
    axes[1].set_xticklabels(mazes, rotation=45, ha='right')
    axes[1].legend()
    axes[1].grid(True, alpha=0.3)

    for i, algo in enumerate(algos):
        lengths = []
        for m in mazes:
            val = next((r['length'] for r in results if r['maze'] == m and r['algo'] == algo), 0)
            lengths.append(val)
        axes[2].bar(x + i*width, lengths, width, label=algo)
    axes[2].set_xlabel('Лабиринт')
    axes[2].set_ylabel('Длина пути')
    axes[2].set_title('Сравнение длины пути')
    axes[2].set_xticks(x + width)
    axes[2].set_xticklabels(mazes, rotation=45, ha='right')
    axes[2].legend()
    axes[2].grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig('performance_comparison.png', dpi=150, bbox_inches='tight')
    plt.show()


# ---------- 8. Главный блок ----------
if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == 'experiment':
        # Режим экспериментов
        test_mazes = [
            ("maze/maze1.txt", "Small 10x6"),
            ("maze/maze10x10.txt", "Medium 10x10"),
            ("maze/maze20x20.txt", "Large 20x20"),
            ("maze/maze_empty.txt", "Empty 15x15"),
            ("maze/maze_no_exit.txt", "No exit 10x10")
        ]
        strategies = [
            ("BFS", BFS()),
            ("DFS", DFS()),
            ("AStar", AStar())
        ]
        results = []
        for maze_path, maze_name in test_mazes:
            print(f"Тестируем {maze_name}...")
            for algo_name, algo in strategies:
                try:
                    stats = run_benchmark(maze_path, algo, runs=3)
                    results.append({
                        'maze': maze_name,
                        'algo': algo_name,
                        'time_ms': stats['time_ms'],
                        'visited': stats['visited'],
                        'length': stats['length']
                    })
                    print(f"  {algo_name}: время={stats['time_ms']:.3f}мс, посещено={stats['visited']:.0f}, длина={stats['length']:.0f}")
                except Exception as e:
                    print(f"  {algo_name}: ошибка - {e}")
        # Сохраняем CSV
        with open('experiment_results.csv', 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=['maze', 'algo', 'time_ms', 'visited', 'length'])
            writer.writeheader()
            writer.writerows(results)
        # Строим графики
        if results:
            make_plots(results)
        print("\nРезультаты сохранены в experiment_results.csv и performance_comparison.png")
    else:
        # Интерактивный режим
        loader = TextLabyrinthLoader()
        lab = loader.load("maze/maze1.txt")

        player = Walker(lab.start_tile, lab)
        view = ConsoleRenderer(player)
        view.notify("maze_loaded", lab)

        solver = LabyrinthSolver(lab)
        solver.attach(view)

        print("\n  УПРАВЛЕНИЕ:")
        print("  H (влево)  J (вниз)  K (вверх)  L (вправо)")
        print("  U - отменить ход    Q - выход")
        print("\n  АВТОПОИСК:")
        print("  B - BFS    D - DFS    A - A*")
        print("\n" + "=" * 50)

        command_stack = []
        while True:
            key = input("\n  Команда > ").lower()
            if key == 'q':
                print("\n  До свидания!")
                break
            elif key == 'b':
                solver.set_strategy(BFS())
                stats = solver.solve()
                if stats:
                    print(f"\n  BFS: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
            elif key == 'd':
                solver.set_strategy(DFS())
                stats = solver.solve()
                print(f"\n  DFS: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
            elif key == 'a':
                solver.set_strategy(AStar())
                stats = solver.solve()
                print(f"\n  A*: время={stats['time_ms']:.3f}мс, посещено={stats['visited']}, длина={stats['length']}")
            elif key in ['h', 'j', 'k', 'l']:
                dirs = {'h': (-1,0), 'l': (1,0), 'k': (0,-1), 'j': (0,1)}
                act = MoveAction(player, dirs[key][0], dirs[key][1], lab)
                if act.do():
                    command_stack.append(act)
                    view.notify("player_moved", lab)
                    if player.current == lab.exit_tile:
                        print("\n  ПОЗДРАВЛЯЮ! ВЫ НАШЛИ ВЫХОД!")
                        print(f"  Сделано ходов: {len(command_stack)}")
                        break
                else:
                    print("\n  Туда нельзя – стена!")
            elif key == 'u':
                if command_stack:
                    cmd = command_stack.pop()
                    cmd.undo()
                    view.notify("player_moved", lab)
                    print("\n  Отмена последнего хода")
                else:
                    print("\n  Нечего отменять")
            else:
                print("\n  Неизвестная команда. Используйте h,j,k,l, u, b, d, a, q")

        print("\n  Игра окончена. Спасибо!")