forked from UNN/2026-rff_mp
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
"""
|
||
maze_solver/solver.py - оркестратор MazeSolver + паттерн Observer.
|
||
|
||
MazeSolver знает лабиринт и текущую стратегию (Strategy). Перед поиском
|
||
он уведомляет наблюдателей (Observer) о старте, после поиска - о результате.
|
||
"""
|
||
|
||
import time
|
||
from abc import ABC, abstractmethod
|
||
|
||
|
||
# ---------- Observer ----------
|
||
|
||
class Observer(ABC):
|
||
"""Интерфейс наблюдателя."""
|
||
|
||
@abstractmethod
|
||
def update(self, event):
|
||
"""event - dict с ключом 'type' и сопровождающими данными."""
|
||
|
||
|
||
class ConsoleView(Observer):
|
||
"""Простой текстовый наблюдатель."""
|
||
|
||
def __init__(self, verbose=True):
|
||
self.verbose = verbose
|
||
|
||
def update(self, event):
|
||
if not self.verbose:
|
||
return
|
||
t = event["type"]
|
||
if t == "maze_loaded":
|
||
m = event["maze"]
|
||
print(f"[ConsoleView] лабиринт {m.width}x{m.height} загружен")
|
||
elif t == "search_start":
|
||
print(f"[ConsoleView] старт поиска: {event['strategy']}")
|
||
elif t == "search_end":
|
||
stats = event["stats"]
|
||
print(f"[ConsoleView] поиск окончен: путь={stats['path_length']}, "
|
||
f"посещено={stats['visited']}, время={stats['elapsed_ms']:.3f} мс")
|
||
elif t == "move":
|
||
print(f"[ConsoleView] игрок -> ({event['x']},{event['y']})")
|
||
elif t == "path_found":
|
||
print("[ConsoleView] путь найден")
|
||
elif t == "no_path":
|
||
print("[ConsoleView] пути нет")
|
||
|
||
|
||
# ---------- MazeSolver ----------
|
||
|
||
class SearchStats(dict):
|
||
"""Простой dict-подобный контейнер статистики поиска."""
|
||
pass
|
||
|
||
|
||
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 RuntimeError("Стратегия не задана")
|
||
if self.maze.start is None or self.maze.exit_ is None:
|
||
raise RuntimeError("В лабиринте нет старта или выхода")
|
||
|
||
self._notify({"type": "search_start", "strategy": self.strategy.name})
|
||
|
||
t0 = time.perf_counter()
|
||
result = self.strategy.find_path(self.maze,
|
||
self.maze.start,
|
||
self.maze.exit_)
|
||
elapsed = (time.perf_counter() - t0) * 1000.0
|
||
|
||
path = result["path"]
|
||
stats = SearchStats(
|
||
strategy=self.strategy.name,
|
||
elapsed_ms=elapsed,
|
||
visited=result["visited"],
|
||
path_length=len(path),
|
||
path=path,
|
||
)
|
||
self._notify({"type": "search_end", "stats": stats})
|
||
if path:
|
||
self._notify({"type": "path_found"})
|
||
else:
|
||
self._notify({"type": "no_path"})
|
||
return stats
|