2026-rff_mp/fomichevks/docs/data/plots.py

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

from maze import DATA_PATH



class Tile:
    def __init__(self, x: int, y: int):
        self._x = x
        self._y = y
        self._wall = False
        self._start = False
        self._exit = False

    @property
    def x(self) -> int:
        return self._x

    @property
    def y(self) -> int:
        return self._y

    @property
    def is_wall(self) -> bool:
        return self._wall

    @is_wall.setter
    def is_wall(self, v: bool):
        self._wall = v

    @property
    def is_start(self) -> bool:
        return self._start

    @is_start.setter
    def is_start(self, v: bool):
        self._start = v

    @property
    def is_exit(self) -> bool:
        return self._exit

    @is_exit.setter
    def is_exit(self, v: bool):
        self._exit = v

    def passable(self) -> bool:
        return not self._wall

    def __hash__(self):
        return hash((self._x, self._y))

    def __eq__(self, other):
        if not isinstance(other, Tile):
            return False
        return self._x == other._x and self._y == other._y


class Maze:
    def __init__(self, w: int, h: int):
        self._w = w
        self._h = h
        self._cells = [[Tile(x, y) for x in range(w)] for y in range(h)]
        self._start = None
        self._exit = None

    @property
    def width(self) -> int:
        return self._w

    @property
    def height(self) -> int:
        return self._h

    @property
    def start(self):
        return self._start

    @property
    def exit(self):
        return self._exit

    def get_cell(self, x: int, y: int):
        if 0 <= x < self._w and 0 <= y < self._h:
            return self._cells[y][x]
        return None

    def set_cell(self, x: int, y: int, kind: str):
        c = self.get_cell(x, y)
        if not c:
            return
        if kind == 'wall':
            c.is_wall = True
        elif kind == 'start':
            if self._start:
                self._start.is_start = False
            c.is_start = True
            c.is_wall = False
            self._start = c
        elif kind == 'exit':
            if self._exit:
                self._exit.is_exit = False
            c.is_exit = True
            c.is_wall = False
            self._exit = c
        elif kind == 'path':
            c.is_wall = False

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


class TextMazeLoader:
    def load(self, filename: str):
        with open(filename, 'r', encoding='utf-8') as f:
            lines = [line.rstrip('\n') for line in f.readlines()]

        h = len(lines)
        w = max(len(line) for line in lines) if h else 0

        start_count = 0
        exit_count = 0
        maze = Maze(w, h)

        for y, line in enumerate(lines):
            for x, ch in enumerate(line):
                if ch == '#':
                    maze.set_cell(x, y, 'wall')
                elif ch == 'S':
                    maze.set_cell(x, y, 'start')
                    start_count += 1
                elif ch == 'E':
                    maze.set_cell(x, y, 'exit')
                    exit_count += 1
                else:
                    maze.set_cell(x, y, 'path')

        if start_count != 1 or exit_count != 1:
            raise ValueError(f"Maze must have one S and one E. Found: S={start_count}, E={exit_count}")

        return maze


class BFS:
    def __init__(self):
        self._visited = 0

    def find(self, maze, start, goal):
        from collections import deque
        queue = deque([start])
        parent = {start: None}
        visited = {start}

        while queue:
            current = queue.popleft()

            if current == goal:
                self._visited = len(visited)
                return self._reconstruct(parent, start, goal)

            for neighbor in maze.neighbours(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    parent[neighbor] = current
                    queue.append(neighbor)

        self._visited = len(visited)
        return []

    def _reconstruct(self, parent, start, goal):
        path = []
        current = goal
        while current is not None:
            path.append(current)
            current = parent.get(current)
        path.reverse()
        return path if path and path[0] == start else []

    @property
    def visited_count(self):
        return self._visited


class DFS:
    def __init__(self):
        self._visited = 0

    def find(self, maze, start, goal):
        stack = [start]
        parent = {start: None}
        visited = {start}

        while stack:
            current = stack.pop()

            if current == goal:
                self._visited = len(visited)
                return self._reconstruct(parent, start, goal)

            for neighbor in maze.neighbours(current):
                if neighbor not in visited:
                    visited.add(neighbor)
                    parent[neighbor] = current
                    stack.append(neighbor)

        self._visited = len(visited)
        return []

    def _reconstruct(self, parent, start, goal):
        path = []
        current = goal
        while current is not None:
            path.append(current)
            current = parent.get(current)
        path.reverse()
        return path if path and path[0] == start else []

    @property
    def visited_count(self):
        return self._visited


class AStar:
    def __init__(self):
        self._visited = 0

    def _heuristic(self, cell, goal):
        return abs(cell.x - goal.x) + abs(cell.y - goal.y)

    def find(self, maze, start, goal):
        import heapq
        heap = []
        counter = 0
        start_f = self._heuristic(start, goal)
        heapq.heappush(heap, (start_f, counter, start))
        counter += 1

        parent = {}
        g_score = {start: 0}
        f_score = {start: start_f}
        visited = set()

        while heap:
            current_f, _, current = heapq.heappop(heap)
            visited.add(current)

            if current == goal:
                self._visited = len(visited)
                return self._reconstruct(parent, start, goal)

            if current_f > f_score.get(current, float('inf')):
                continue

            for neighbor in maze.neighbours(current):
                tentative_g = g_score[current] + 1

                if tentative_g < g_score.get(neighbor, float('inf')):
                    parent[neighbor] = current
                    g_score[neighbor] = tentative_g
                    new_f = tentative_g + self._heuristic(neighbor, goal)
                    f_score[neighbor] = new_f
                    heapq.heappush(heap, (new_f, counter, neighbor))
                    counter += 1

        self._visited = len(visited)
        return []

    def _reconstruct(self, parent, start, goal):
        path = []
        current = goal
        while current is not None:
            path.append(current)
            current = parent.get(current)
        path.reverse()
        return path if path and path[0] == start else []

    @property
    def visited_count(self):
        return self._visited


class MazeSolver:
    def __init__(self, maze):
        self._maze = maze
        self._algorithm = None

    def set_algorithm(self, algorithm):
        self._algorithm = algorithm

    def solve(self):
        if not self._algorithm:
            raise ValueError("Algorithm not set")

        start_time = time.perf_counter()
        path = self._algorithm.find(self._maze, self._maze.start, self._maze.exit)
        end_time = time.perf_counter()

        elapsed_ms = (end_time - start_time) * 1000

        return {
            'time_ms': elapsed_ms,
            'visited': self._algorithm.visited_count,
            'path_length': len(path),
            'path': path
        }





DATA_PATH = r"C:\Users\Kirill\2026-rff_mp\fomichevks\docs\data"


class ExperimentRunner:
    def __init__(self):
        self.algorithms = {
            "BFS": BFS(),
            "DFS": DFS(),
            "A*": AStar()
        }
        self.loader = TextMazeLoader()

    def run_benchmark(self, maze_file: str, algorithm: str, runs: int = 5):
        try:
            maze = self.loader.load(maze_file)
        except Exception as e:
            return None

        total_time = 0.0
        total_visited = 0
        total_length = 0
        successes = 0

        for _ in range(runs):
            solver = MazeSolver(maze)
            solver.set_algorithm(self.algorithms[algorithm])
            result = solver.solve()

            if result and result['path_length'] > 0:
                total_time += result['time_ms']
                total_visited += result['visited']
                total_length += result['path_length']
                successes += 1

        if successes == 0:
            return None

        return {
            'time_ms': total_time / successes,
            'visited_cells': total_visited / successes,
            'path_length': total_length / successes,
            'success_rate': successes / runs
        }

    def run_all_experiments(self, runs: int = 5):
        mazes_list = [
            (os.path.join(DATA_PATH, "small.txt"), "Small (10x10)"),
            (os.path.join(DATA_PATH, "medium.txt"), "Medium (50x50)"),
            (os.path.join(DATA_PATH, "large.txt"), "Large (100x100)"),
            (os.path.join(DATA_PATH, "empty.txt"), "Empty"),
            (os.path.join(DATA_PATH, "no_exit.txt"), "No exit")
        ]

        results = []


        print("running experiments")

        print(f"Data path: {DATA_PATH}")


        for maze_file, maze_name in mazes_list:
            if not os.path.exists(maze_file):
                print(f"\n[warn] File not found: {maze_file}")
                continue

            print(f"\nTesting: {maze_name}")

            for algo_name in self.algorithms.keys():
                stats = self.run_benchmark(maze_file, algo_name, runs)

                if stats:
                    print(
                        f"  {algo_name}: time={stats['time_ms']:.3f}ms, visited={stats['visited_cells']:.0f}, length={stats['path_length']:.0f}")
                    results.append({
                        'maze': maze_name,
                        'strategy': algo_name,
                        'time_ms': stats['time_ms'],
                        'visited_cells': stats['visited_cells'],
                        'path_length': stats['path_length'],
                        'success_rate': stats['success_rate']
                    })
                else:
                    print(f"  {algo_name}: no path found")
                    results.append({
                        'maze': maze_name,
                        'strategy': algo_name,
                        'time_ms': -1,
                        'visited_cells': -1,
                        'path_length': -1,
                        'success_rate': 0
                    })

        return results


def create_visualizations(results):
    valid_results = [r for r in results if r['time_ms'] > 0]
    if not valid_results:
        print("no valid results for visualization")
        return

    mazes = sorted(set(r['maze'] for r in valid_results))
    algorithms = ['BFS', 'DFS', 'A*']

    fig, axes = plt.subplots(1, 3, figsize=(15, 5))
    fig.suptitle('pathfinding algorithms comparison', fontsize=14)

    x = np.arange(len(mazes))
    width = 0.25

    # Time chart
    for i, algo in enumerate(algorithms):
        times = []
        for maze in mazes:
            val = next((r['time_ms'] for r in valid_results
                        if r['maze'] == maze and r['strategy'] == algo), 0)
            times.append(val)
        bars = axes[0].bar(x + i * width, times, width, label=algo, alpha=0.8)
        for bar, val in zip(bars, times):
            if val > 0:
                axes[0].text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.5,
                             f'{val:.1f}', ha='center', va='bottom', fontsize=7)

    axes[0].set_title('execution Time (ms)')
    axes[0].set_ylabel('time (ms)')
    axes[0].set_xticks(x + width)
    axes[0].set_xticklabels(mazes, rotation=45, ha='right', fontsize=8)
    axes[0].legend()
    axes[0].grid(alpha=0.3, axis='y')

    # Visited cells chart
    for i, algo in enumerate(algorithms):
        visited = []
        for maze in mazes:
            val = next((r['visited_cells'] for r in valid_results
                        if r['maze'] == maze and r['strategy'] == algo), 0)
            visited.append(val)
        bars = axes[1].bar(x + i * width, visited, width, label=algo, alpha=0.8)
        for bar, val in zip(bars, visited):
            if val > 0:
                axes[1].text(bar.get_x() + bar.get_width() / 2, bar.get_height(),
                             f'{val:.0f}', ha='center', va='bottom', fontsize=7)

    axes[1].set_title('visited Cells')
    axes[1].set_ylabel('count')
    axes[1].set_xticks(x + width)
    axes[1].set_xticklabels(mazes, rotation=45, ha='right', fontsize=8)
    axes[1].legend()
    axes[1].grid(alpha=0.3, axis='y')

    # Path length chart
    for i, algo in enumerate(algorithms):
        lengths = []
        for maze in mazes:
            val = next((r['path_length'] for r in valid_results
                        if r['maze'] == maze and r['strategy'] == algo), 0)
            lengths.append(val)
        bars = axes[2].bar(x + i * width, lengths, width, label=algo, alpha=0.8)
        for bar, val in zip(bars, lengths):
            if val > 0:
                axes[2].text(bar.get_x() + bar.get_width() / 2, bar.get_height(),
                             f'{val:.0f}', ha='center', va='bottom', fontsize=7)

    axes[2].set_title('path Length')
    axes[2].set_ylabel('steps')
    axes[2].set_xticks(x + width)
    axes[2].set_xticklabels(mazes, rotation=45, ha='right', fontsize=8)
    axes[2].legend()
    axes[2].grid(alpha=0.3, axis='y')

    plt.tight_layout()

    output_path = os.path.join(DATA_PATH, 'experiment_results.png')
    plt.savefig(output_path, dpi=150, bbox_inches='tight')
    print(f"\nPlot saved to: {output_path}")
    plt.show()


def save_results_to_csv(results, filename='experiment_results.csv'):
    if not results:
        return

    filepath = os.path.join(DATA_PATH, filename)
    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        fieldnames = ['maze', 'strategy', 'time_ms', 'visited_cells', 'path_length', 'success_rate']
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)

    print(f"Results saved to: {filepath}")


def analyze_efficiency(results):
    valid_results = [r for r in results if r['time_ms'] > 0]
    if not valid_results:
        print("no valid results for analysis")
        return

    algo_stats = {}
    for algo in ['BFS', 'DFS', 'A*']:
        algo_data = [r for r in valid_results if r['strategy'] == algo]
        if algo_data:
            algo_stats[algo] = {
                'avg_time': sum(r['time_ms'] for r in algo_data) / len(algo_data),
                'avg_visited': sum(r['visited_cells'] for r in algo_data) / len(algo_data),
                'avg_length': sum(r['path_length'] for r in algo_data) / len(algo_data)
            }


    print("average values across all mazes")
    print(f"{'Algorithm':<12} {'Time (ms)':<15} {'Visited':<15} {'Path length':<15}")

    for algo, stats in algo_stats.items():
        print(f"{algo:<12} {stats['avg_time']:<15.3f} {stats['avg_visited']:<15.1f} {stats['avg_length']:<15.1f}")

    fastest = min(algo_stats.items(), key=lambda x: x[1]['avg_time'])
    optimal = min(algo_stats.items(), key=lambda x: x[1]['avg_length'])
    efficient = min(algo_stats.items(), key=lambda x: x[1]['avg_visited'])

    print("conclusions:")
    print(f"  fastest algorithm: {fastest[0]} ({fastest[1]['avg_time']:.3f} ms avg)")
    print(f"  optimal path: {optimal[0]} ({optimal[1]['avg_length']:.1f} steps avg)")
    print(f"  most efficient (fewest visits): {efficient[0]} ({efficient[1]['avg_visited']:.0f} cells avg)")
    print("=" * 70)


def main():


    if not os.path.exists(DATA_PATH):
        print(f"\nerr: directory not found: {DATA_PATH}")
        print("please create the directory and place maze files there.")
        print("\nexpected structure:")
        print(f"  {DATA_PATH}/")
        print("    ├── small.txt")
        print("    ├── medium.txt")
        print("    ├── large.txt")
        print("    ├── empty.txt")
        print("    └── no_exit.txt")
        return

    runner = ExperimentRunner()
    results = runner.run_all_experiments(runs=5)

    if not results:
        print("\nNo results. Check if maze files exist in:", DATA_PATH)
        return

    save_results_to_csv(results)
    analyze_efficiency(results)
    create_visualizations(results)




if __name__ == "__main__":
    main()