Dijkstra algorithm on a grid with obstacles.

grid_with_obstacles_Dijkstra.py

@@ -0,0 +1,169 @@
+import matplotlib.patches as patches
+import numpy as np
+import matplotlib.pyplot as plt
+import networkx as nx
+import matplotlib.animation as animation
+import heapq
+import random
+
+# graph creation
+rows, cols = 17, 17
+grid = np.zeros((rows, cols))
+G = nx.grid_2d_graph(rows, cols)
+
+# initial and target nodes
+start = (0, 0)
+target = (12, 12)
+
+# obstacles creation
+obstacle_ratio = 0.2  
+num_obstacles = int(rows * cols * obstacle_ratio)
+all_positions = [(r, c) for r in range(rows) for c in range(cols) if (r, c) not in [start, target]]
+obstacles = random.sample(all_positions, num_obstacles)
+for obs in obstacles:
+    grid[obs] = 1 
+#removing them form the graph
+for obs in obstacles:
+    if obs in G:
+        G.remove_node(obs)
+
+# Function of Dijkstra algorithm. Shown in the animation
+def dijkstra_stepwise(G, start, target):
+    # Initialisation
+    distances = {node: float('inf') for node in G.nodes()}
+    distances[start] = 0
+    previous_nodes = {node: None for node in G.nodes()}
+    evaluated_nodes = []  # List of evaluated nodes
+    path_to_current = []  # History of the path to the current node
+
+    # priority queue on the non-evaluated nodes
+    priority_queue = [(0, start)]  # (distance, node)
+    heapq.heapify(priority_queue)
+
+    while priority_queue:
+        current_distance, current_node = heapq.heappop(priority_queue)
+
+        if current_node not in evaluated_nodes:
+            evaluated_nodes.append(current_node)
+
+        # Reconstruction of the path to the current node
+        temp_path = []
+        node = current_node
+        while node is not None:
+            temp_path.append(node)
+            node = previous_nodes[node]
+        temp_path.reverse()
+
+        path_to_current.append(temp_path)  
+
+        if current_node == target:
+            break
+
+        # Evaluation of the current nodes neighbors
+        for neighbor in G.neighbors(current_node):
+            if neighbor not in evaluated_nodes:
+                new_distance = current_distance + 1
+                if new_distance < distances[neighbor]:
+                    distances[neighbor] = new_distance
+                    previous_nodes[neighbor] = current_node
+                    heapq.heappush(priority_queue, (new_distance, neighbor))
+
+    return evaluated_nodes, path_to_current
+
+# Retrieving the results of Dijkstra's algorithm
+evaluated_nodes, path_history = dijkstra_stepwise(G, start, target)
+
+# Creation of the figure
+fig, ax = plt.subplots(figsize=(6, 6))
+ax.imshow(grid, cmap="Greys", origin="upper") # The origin is set on the top left corner
+
+# Display of the grid
+ax.set_xticks(np.arange(-0.5, cols, 1), minor=True)
+ax.set_yticks(np.arange(-0.5, rows, 1), minor=True)
+ax.grid(True, which="minor", color="black", linewidth=0.5)
+ax.tick_params(which="both", bottom=False, left=False, labelbottom=False, labelleft=False)
+
+# Display of the start and target nodes
+start_rect = patches.Rectangle((start[1] - 0.5, start[0] - 0.5), 1, 1, facecolor="green", alpha=0.8)
+target_rect = patches.Rectangle((target[1] - 0.5, target[0] - 0.5), 1, 1, facecolor="yellow", alpha=0.8)
+ax.add_patch(start_rect)
+ax.add_patch(target_rect)
+
+# Initiate the plots
+evaluated_patches = []
+path_patches = []
+
+# List in use for the confetti animationn
+confetti_patches = []
+confetti_velocities = [] 
+
+# Update function for the animation
+def update(frame):
+    # Diplays the evaluated nodes in blue
+    if frame < len(evaluated_nodes):
+        node = evaluated_nodes[frame]
+        if node != start:
+            rect = patches.Rectangle((node[1] - 0.5, node[0] - 0.5), 1, 1, facecolor="blue", alpha=0.6)
+        else:
+            rect = patches.Rectangle((node[1] - 0.5, node[0] - 0.5), 1, 1, facecolor="green", alpha=0.6)
+        ax.add_patch(rect)
+        evaluated_patches.append(rect)
+
+
+    # Delete previous path
+    if frame < len(path_history):
+        for patch in path_patches:
+            patch.remove()
+        path_patches.clear()
+
+    # Displays the current shortest path 
+    if frame < len(path_history):
+        for node in path_history[frame]:
+            if node != start:
+                rect = patches.Rectangle((node[1] - 0.5, node[0] - 0.5), 1, 1, facecolor="red", alpha=0.8)
+            else:
+                rect = patches.Rectangle((node[1] - 0.5, node[0] - 0.5), 1, 1, facecolor="green", alpha=0.8)
+            ax.add_patch(rect)
+            path_patches.append(rect)
+
+
+    # Getting rid of the previous shortest paths
+    if frame == len(path_history) - 1:
+        for patch in path_patches:
+            patch.set_facecolor("red")  
+        path_patches[-1].set_facecolor("cyan") # When the target is reached, becomes cyan
+
+    # confetti effect when we reach the target node
+    if frame == len(path_history) - 1:
+        # creation of particules around the final node
+        confetti_patches.clear()  
+        confetti_velocities.clear()  
+        for _ in range(50):  # 50 particules
+            offset_x = random.uniform(-0.5, 0.5)
+            offset_y = random.uniform(-0.5, 0.5)
+            color = np.random.rand(3,)  # Random color
+            confetti_rect = patches.Circle(
+                (target[1] + offset_x, target[0] + offset_y),
+                radius=0.1, facecolor=color, alpha=0.7
+            )
+            ax.add_patch(confetti_rect)
+            confetti_patches.append(confetti_rect)
+
+            # Define velocity of the particales
+            velocity = [random.uniform(-0.1, 0.1), random.uniform(-0.1, 0.1)]  # Mouvement dans toutes les directions
+            confetti_velocities.append(velocity)
+
+    # Animation of the particales
+    for i, patch in enumerate(confetti_patches):
+        new_center = (
+            patch.center[0] + confetti_velocities[i][0],  # X movement
+            patch.center[1] + confetti_velocities[i][1]   # Y movement
+        )
+        patch.set_center(new_center)
+
+    return evaluated_patches + path_patches + confetti_patches
+
+# creation of the animation
+ani = animation.FuncAnimation(fig, update, frames=len(evaluated_nodes) + 20, interval=100, blit=False, repeat=False)
+
+plt.show()

quactography/classical/io.py

@@ -0,0 +1,104 @@
+import numpy as np
+import json
+import networkx as nx
+
+
+def save_graph(G, output_base, copies=1):
+    """
+    Save a graph to .npz files as arrays of nodes and edges.
+
+    Parameters
+    ----------
+    G : networkx.Graph
+        The graph to be saved.
+    output_base : str
+        The base name for output files (e.g., 'graph' will become 'graph_0.npz').
+    copies : int, optional
+        The number of copies to save. Default is 1.
+
+    Returns
+    -------
+    None
+    """
+    for i in range(copies):
+        output_file = f"{output_base}_{i}.npz"
+        nodes = list(G.nodes())
+        edges = list(G.edges())
+
+        nodes_array = np.array(nodes)
+        edges_array = np.array(edges)
+
+        np.savez(output_file, nodes=nodes_array, edges=edges_array)
+        print(f"Copie {i + 1}/{copies} saved as '{output_file}'.")
+
+
+def load_the_graph(file_path):
+    """
+    Load a graph from a .json or .npz file.
+
+    Parameters
+    ----------
+    file_path : str
+        Path to the graph file. Supported formats: .json, .npz.
+
+    Returns
+    -------
+    networkx.Graph
+        The loaded graph.
+
+    Raises
+    ------
+    ValueError
+        If the file format is not supported.
+    """
+    if file_path.endswith(".json"):
+        with open(file_path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+
+        G = nx.Graph()
+        for node in data["nodes"]:
+            G.add_node(tuple(node))
+
+        for edge in data["edges"]:
+            node1, node2 = tuple(map(tuple, edge[:2]))
+            weight = edge[2] if len(edge) > 2 else 1
+            G.add_edge(node1, node2, weight=weight)
+
+        return G
+
+    if file_path.endswith(".npz"):
+        data = np.load(file_path, allow_pickle=True)
+        adjacency_matrix = data["adjacency_matrix"]
+        raw_nodes = data["node_indices"]
+
+        node_indices = [
+            tuple(node) if isinstance(node, (list, np.ndarray, tuple)) else (node,)
+            for node in raw_nodes
+        ]
+        node_indices = [(n[0], 0) if len(n) == 1 else n for n in node_indices]
+
+        print(
+            f"Loaded adjacency matrix of shape {adjacency_matrix.shape} "
+            f"with {len(node_indices)} nodes."
+        )
+
+        G = nx.Graph()
+        for node in node_indices:
+            G.add_node(node)
+
+        for i in range(len(adjacency_matrix)):
+            for j in range(i + 1, len(adjacency_matrix[i])):
+                if adjacency_matrix[i, j] > 0:
+                    G.add_edge(
+                        node_indices[i], node_indices[j],
+                        weight=adjacency_matrix[i, j]
+                    )
+
+        print(f"Nodes in G: {list(G.nodes())}")
+        print(
+            f"Graph charged with {G.number_of_nodes()} nodes and "
+            f"{G.number_of_edges()} edges."
+        )
+        return G
+
+    raise ValueError("Unsupported file format. Use either .json or .npz.")

quactography/classical/utils/Astar.py

@@ -0,0 +1,90 @@
+import time
+import heapq
+from travel_related import heuristic, get_neighbors_diagonal
+
+
+def astar_stepwise(G, start, target, diagonal_mode="nondiagonal"):
+    """
+    Perform the A* pathfinding algorithm in a stepwise manner on a given graph.
+
+    Parameters
+    ----------
+    G : networkx.Graph
+        The input graph where each node is typically a tuple (e.g., (x, y)).
+        Edges may have a 'weight' attribute indicating traversal cost
+        (default is 1).
+    start : hashable
+        The starting node for the pathfinding algorithm.
+    target : hashable
+        The target (goal) node to reach.
+    diagonal_mode : str, optional
+        Neighbor retrieval mode. Must be either:
+        - "nondiagonal": only 4-directional neighbors (up, down, left, right)
+        - "diagonal": include diagonal neighbors (8 directions total)
+        Default is "nondiagonal".
+
+    Returns
+    -------
+    evaluated_nodes : list
+        The list of nodes in the order they were evaluated by the algorithm.
+    path_to_current : list of list
+        A list containing the reconstructed path from the start node to each
+        node evaluated so far, in evaluation order.
+    current_f_score : float
+        The final f-score (g + h) associated with the target node if found,
+        or with the last evaluated node if the target was not reached.
+    """
+    start_time = time.time()
+
+    g_scores = {node: float('inf') for node in G.nodes()}
+    g_scores[start] = 0
+
+    f_scores = {node: float('inf') for node in G.nodes()}
+    f_scores[start] = heuristic(start, target)
+
+    previous_nodes = {node: None for node in G.nodes()}
+    evaluated_nodes = []
+    path_to_current = []
+    priority_queue = [(f_scores[start], start)]
+    heapq.heapify(priority_queue)
+
+    while priority_queue:
+        current_f_score, current_node = heapq.heappop(priority_queue)
+
+        if current_node not in evaluated_nodes:
+            evaluated_nodes.append(current_node)
+
+        temp_path = []
+        node = current_node
+        while node is not None:
+            temp_path.append(node)
+            node = previous_nodes[node]
+        temp_path.reverse()
+        path_to_current.append(temp_path)
+
+        if current_node == target:
+            break
+
+        if diagonal_mode == "diagonal":
+            neighbors = list(get_neighbors_diagonal(current_node, G))
+        else:
+            neighbors = list(G.neighbors(current_node))
+
+        for neighbor in neighbors:
+            edge_weight = G[current_node][neighbor].get("weight", 1)
+            tentative_g_score = g_scores[current_node] + edge_weight
+            if tentative_g_score < g_scores[neighbor]:
+                previous_nodes[neighbor] = current_node
+                g_scores[neighbor] = tentative_g_score
+                f_scores[neighbor] = tentative_g_score
+                + heuristic(neighbor, target)
+                heapq.heappush(
+                    priority_queue,
+                    (f_scores[neighbor], neighbor)
+                )
+
+    end_time = time.time()
+    execution_time = end_time - start_time
+    print(f"Execution time of A* : {execution_time:.4f} seconds")
+
+    return evaluated_nodes, path_to_current, current_f_score