File size: 4,684 Bytes
1f5351c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """
map_graph.py
Gère la structure du graphe, la persistance JSON et les algorithmes spatiaux (BFS).
"""
import json
import os
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Set, Tuple
from collections import deque
@dataclass
class Connection:
to_id: int
direction: str
is_verified: bool = True # True si on l'a vraiment traversé
@dataclass
class RoomNode:
id: int
name: str
# Connexions: direction -> Connection object
exits: Dict[str, Connection] = field(default_factory=dict)
# Murs/Bloquages: direction -> nombre d'échecs
failed_exits: Dict[str, int] = field(default_factory=dict)
visited_count: int = 1
class MapGraph:
def __init__(self):
self.rooms: Dict[int, RoomNode] = {}
# Mapping des directions inverses pour l'inférence
self.inverse_dirs = {
"n": "s", "s": "n", "e": "w", "w": "e",
"north": "south", "south": "north",
"east": "west", "west": "east",
"up": "down", "down": "up",
"ne": "sw", "sw": "ne", "nw": "se", "se": "nw",
"enter": "exit", "exit": "enter", "in": "out", "out": "in"
}
def add_or_update_room(self, room_id: int, name: str):
"""Crée ou met à jour un noeud."""
if room_id not in self.rooms:
self.rooms[room_id] = RoomNode(id=room_id, name=name)
else:
self.rooms[room_id].visited_count += 1
# On garde le nom le plus long (souvent le plus descriptif)
if len(name) > len(self.rooms[room_id].name):
self.rooms[room_id].name = name
def add_connection(self, from_id: int, to_id: int, direction: str):
"""Ajoute une arête dirigée."""
if from_id not in self.rooms or to_id not in self.rooms:
return
# 1. Connexion directe (Vérifiée)
self.rooms[from_id].exits[direction] = Connection(to_id=to_id, direction=direction, is_verified=True)
# 2. Si une connexion précédente échouée existait, on la nettoie (Pruning)
if direction in self.rooms[from_id].failed_exits:
del self.rooms[from_id].failed_exits[direction]
# 3. Connexion inverse (Inférée/Non vérifiée)
# On suppose que si on va au Nord, le Sud ramène au départ.
rev_dir = self.inverse_dirs.get(direction)
if rev_dir:
target_room = self.rooms[to_id]
# On ajoute seulement si l'exit n'existe pas déjà
if rev_dir not in target_room.exits:
target_room.exits[rev_dir] = Connection(to_id=from_id, direction=rev_dir, is_verified=False)
def record_failure(self, room_id: int, direction: str):
"""Enregistre un mur/bloquage."""
if room_id in self.rooms:
self.rooms[room_id].failed_exits[direction] = self.rooms[room_id].failed_exits.get(direction, 0) + 1
def get_shortest_path(self, start_id: int, target_id: int) -> Optional[List[str]]:
"""BFS pour trouver le chemin le plus court."""
queue = deque([(start_id, [])])
visited = {start_id}
while queue:
curr, path = queue.popleft()
if curr == target_id:
return path
if curr in self.rooms:
for d, conn in self.rooms[curr].exits.items():
if conn.to_id not in visited:
visited.add(conn.to_id)
queue.append((conn.to_id, path + [d]))
return None
def get_hubs(self) -> List[str]:
"""Retourne les noms des salles importantes (Hubs) avec >3 sorties."""
return [r.name for r in self.rooms.values() if len(r.exits) > 3]
# --- Persistance JSON ---
def save(self, filepath: str):
data = {
str(rid): asdict(node) for rid, node in self.rooms.items()
}
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
def load(self, filepath: str):
if not os.path.exists(filepath): return
try:
with open(filepath, 'r') as f:
data = json.load(f)
self.rooms = {}
for rid, rdata in data.items():
node = RoomNode(id=int(rid), name=rdata['name'], visited_count=rdata['visited_count'])
node.failed_exits = rdata['failed_exits']
# Reconstruire les objets Connection
node.exits = {d: Connection(**c) for d, c in rdata['exits'].items()}
self.rooms[int(rid)] = node
except Exception as e:
print(f"Error loading map: {e}") |