API d'écriture d'attaques

scripts/Attack.py

@@ -0,0 +1,62 @@
+import osmnx as ox
+import networkx as nx
+import random
+
+from CityGraph import CityGraph
+
+# Juste pour test pour l'instant
+ville="Saints"
+
+class Attack :
+	"""
+	Represents an attack on a graph
+	--- Attributes ---
+	graph : the graph to attack (READ-ONLY)
+	attacks : the attacks to perform
+	"""
+
+	graph : CityGraph
+	attacks : list[int] # Format : [time, edge_idx]
+
+	def __init__(self, graph : CityGraph) :
+		self.graph = graph
+		self.attacks = []
+
+	def add_attack(self, time : int, edge_idx : int) -> None :
+		"""Adds an attack to the list"""
+		self.attacks.append((time, edge_idx))
+
+	def write_to_file(self, file_path : str) -> None :
+		"""Writes the attacks to a file"""
+
+		# Conversion
+		edges_deleted : dict[int, set[int]] = {}
+		for time, edge_idx in self.attacks :
+			node0 = self.graph.get_node_index(self.graph.get_edge_index(edge_idx)[0])
+			node1 = self.graph.get_node_index(self.graph.get_edge_index(edge_idx)[1])
+			if (node0, node1) in edges_deleted :
+				edges_deleted[(node0, node1)].add(time)
+			else :
+				edges_deleted[(node0, node1)] = {time}
+
+		# Sorting for easier parsing in the future	
+		for edge in edges_deleted :
+			edges_deleted[edge] = list(edges_deleted[edge])
+			edges_deleted[edge].sort()
+
+		edges_deleted : list[tuple[tuple[int, int], list[int]]] = [((edge[0], edge[1]), edges_deleted[edge]) for edge in edges_deleted]
+		edges_deleted.sort(key = lambda x : (x[0][0], x[0][1]))
+
+		# Writing
+		with open(file_path, "w") as file :
+			file.write(f"{len(edges_deleted)}\n")
+			for edge, times in edges_deleted :
+				file.write(f"{edge[0]} {edge[1]} {len(times)} {' '.join(map(str, times))}\n")
+
+		print(f"Attacks have been written to {file_path}")
+
+def print_suppression_edge(graph : CityGraph, edge_idx : int, time : int):
+	edge = graph.get_edge_index(edge_idx)
+	noeud0 = graph.get_node_index(edge[0])
+	noeud1 = graph.get_node_index(edge[1])
+	print(f"T={time}, suppression de l'arete {edge_idx} entre les noeuds {noeud0} et {noeud1}")

scripts/CityGraph.py

@@ -5,47 +5,70 @@ class CityGraph :
 	"""
 	Represents a graph of a city, with some useful methods
 	--- Attributes ---
-	city : name of the city
+	city_name : name of the city
 	graph : the graph itself
-	nodes : a dict that maps a node to its index in the graph
+	nodes_idx_map : a dict that maps a node to its index in the graph
 	projected_graph : the graph projected on a plane
 	"""
-	city : str
+	city_name : str
 	graph : nx.Graph
-	nodes : dict[int, int]
+	nodes_idx_map : dict[int, int]
 	projected_graph : nx.Graph = None
 
-	def __init__(self, city : str, network_type : str = "drive") :
+	def __init__(self, city_name : str, network_type : str = "drive") :
 		"""Loads a graph from OSmnx through its name"""
-		self.city = city
-		self.graph = ox.graph_from_place(city, network_type=network_type)
-		self.nodes = { edge : i  for i, edge in enumerate(self.graph.nodes())}
+		ox.settings.osm_xml_way_tags=["highway", "lanes"]
+		self.city_name = city_name
+		self.graph = ox.graph_from_place(city_name, network_type=network_type)
+		#Load the graph with the number of lanes too
+		self.graph = ox.graph_from_place(city_name, network_type=network_type)
 
-	def get_node_index(self, node : int) :
-		return self.nodes[node]
+		self.nodes_idx_map = { edge : i  for i, edge in enumerate(self.graph.nodes())}
 
-	def get_node(self, index : int) :
-		return self.nodes[index]
+	def get_node_index(self, node : int) :
+		"""Returns the index in the matrix of the node through it's OSmnx numero"""
+		return self.nodes_idx_map[node]
 
 	def show(self) :
+		"""Opens an image of the graph"""
 		ox.plot_graph(self.graph)
 
 	def print_stats(self) :
+		"""Prints some stats about the graph"""
 		print(ox.basic_stats(self.graph))
 
 	def get_projected_graph(self) :
+		"""Returns the projected graph"""
 		if self.projected_graph is None :
 			self.projected_graph = ox.project_graph(self.graph)
 		return self.projected_graph
 
 	def edges(self) :
+		"""Returns the edges of the graph"""
 		return self.graph.edges()
 
-	def size(self) :
+	def nb_nodes(self) :
+		"""Returns the number of nodes in the graph"""
 		return self.graph.size()
 
+	def nb_edges(self) :
+		"""Returns the number of edges in the graph"""
+		return self.graph.number_of_edges()
+
 	def get_edge_index(self, edge_idx : int) :
+		"""Accesses an edge through its index"""
 		return list(self.graph.edges())[edge_idx]
+
+	# TODO : ca marche pas
+	def get_nb_lanes_of_edge(self, edge_idx : int) -> int:
+		"""Returns the number of lanes of an edge"""
+		#edges = ox.graph_to_gdfs(self.graph, nodes=False, edges=True)
+		#print(edges.columns)
+		#lanes = edges["lanes"]
+		#for (u, v, d) in self.graph.edges(data=True):
+			#print(d)
+		#print(lanes)
+		return self.graph.get_edge_data(*self.get_edge_index(edge_idx), key="lanes")
 
 	@staticmethod
 	def read_from_file(file_path : str) :

scripts/Strategies.py

@@ -0,0 +1,37 @@
+import osmnx as ox
+import networkx as nx
+import random
+
+from CityGraph import CityGraph
+from Attack import Attack, print_suppression_edge
+
+# Juste pour test pour l'instant
+ville="Saints"
+
+def random_attack(graph : CityGraph, nbTemps : int, budget : int) -> Attack :
+	"""A chaque temps, on supprime budget liens choisis aleatoirement"""
+
+	attacks = Attack(graph)
+	edges = list(graph.edges())
+
+	for i in range(nbTemps) :
+		edges_t = edges.copy()
+		budget_used : int = 0
+		nb_retires : int = 0
+		while budget_used < budget :
+			edge_idx = random.randint(0, graph.nb_edges() - 1 - nb_retires)
+			# print_suppression_edge(graph, edge_idx, i)
+			attacks.add_attack(i, edge_idx)
+			edges_t.pop(edge_idx)
+			nb_retires += 1
+			budget_used += 1 #graph.get_nb_lanes_of_edge(edge_idx)
+
+	return attacks
+
+
+graph = CityGraph(ville)
+graph.print_stats()
+graph.show()
+
+a = random_attack(graph, 50, 50)
+a.write_to_file("test_attack.txt")