s444409/Network_attack_propagation
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

Compare commits

base: s444409:main
Branches Tags
s444409:main
s444409:stars
..
compare: s444409:stars
Branches Tags
s444409:main
s444409:stars

No commits in common. "main" and "stars" have entirely different histories.
main ... stars

3 changed files with 56 additions and 725 deletions
Show Stats Expand all files Collapse all files

6
.gitignore vendored
View File

@ -9,12 +9,6 @@
profile_default/ profile_default/
ipython_config.py ipython_config.py
*.gif
*.png
__pycache__/
.idea/
# Remove previous ipynb_checkpoints # Remove previous ipynb_checkpoints
# git rm -r .ipynb_checkpoints/ # git rm -r .ipynb_checkpoints/

546
network_attack_propagation.ipynb
View File

File diff suppressed because one or more lines are too long

229
network_attack_propagation.py
View File

@ -10,7 +10,6 @@ class Node:
def __init__(self, is_infected=False): def __init__(self, is_infected=False):
self.id = random.randint(1, 2000000) self.id = random.randint(1, 2000000)
self.is_infected = is_infected self.is_infected = is_infected
self.under_attack = False
def as_tuple(self): def as_tuple(self):
return self.id, self.is_infected return self.id, self.is_infected
@ -28,15 +27,10 @@ class Edge:
def as_tuple(self): def as_tuple(self):
return self.node_a, self.node_b, {'weight': self.weight} return self.node_a, self.node_b, {'weight': self.weight}
def has_node(self, node: Node) -> bool:
return self.node_a is node or self.node_b is node
class Graph: class Graph:
def __init__(self, use_weights=False): def __init__(self):
self.edges = [] self.edges = []
self.use_weights = use_weights
self.rounds_survived = 0
def add_edge(self, edge: Edge): def add_edge(self, edge: Edge):
self.edges.append(edge) self.edges.append(edge)
@ -51,230 +45,87 @@ class Graph:
nodes.add(edge.node_b) nodes.add(edge.node_b)
return nodes return nodes
def get_edges_with_node(self, node: Node):
return filter(lambda ed: ed.has_node(node), self.edges)
def clear_attacked(self) -> None:
for edge in self.edges:
edge.node_a.under_attack = False
edge.node_b.under_attack = False
def get_adjacent_nodes(self, node: Node) -> [(Node, int)]:
"""
:param node: Node to search for
:return: An array of tuples (node, weight)
"""
edges_with_node = self.get_edges_with_node(node)
nodes = set()
for e in edges_with_node:
if e.node_a is node:
nodes.add((e.node_b, e.weight))
else:
nodes.add((e.node_a, e.weight))
return nodes
def is_alive(self):
nodes_alive = list(filter(lambda x: not x.is_infected, self.get_nodes()))
return len(nodes_alive) > 0
def update_survived(self):
if not self.is_alive():
return
self.rounds_survived += 1
def infect_step(self):
infected_nodes = list(filter(lambda n: n.is_infected, self.get_nodes()))
for node in infected_nodes:
adjacent_nodes = self.get_adjacent_nodes(node)
if self.use_weights:
to_be_infected = random.choices([n[0] for n in adjacent_nodes], weights=[n[1] for n in adjacent_nodes])[0]
else:
to_be_infected = random.choice([n[0] for n in adjacent_nodes])
to_be_infected.is_infected = True
to_be_infected.under_attack = True
self.update_survived()
def update(num, layout, g_repr, ax, our_graph: Graph): def update(num, layout, g_repr, ax, our_graph: Graph):
""" """
This function is called every 'step', so if you wish to update the graph, do it here This function is called every 'step', so if you wish to update the graph, do it here
""" """
if not our_graph.is_alive():
return
if num != 0:
our_graph.infect_step()
ax.clear() ax.clear()
ax.set_title(f'Step: {num}', loc='right', fontsize=30) for n in our_graph.get_nodes():
n.is_infected = bool(random.getrandbits(1))
colors = ['red' if n.is_infected else 'blue' for n in g_repr] colors = ['red' if n.is_infected else 'blue' for n in g_repr]
edgecolors = ['black' if n.under_attack else 'none' for n in g_repr] nx.draw_networkx(g_repr, ax=ax, pos=layout, node_color=colors, with_labels=False)
linewidths = [3 if c == 'black' else 0 for c in edgecolors]
sizes = [300 if n.is_infected else 150 for n in g_repr]
nx.draw(
g_repr,
ax=ax,
pos=layout,
node_color=colors,
linewidths=linewidths,
edgecolors=edgecolors,
with_labels=False,
node_size=sizes,
alpha=0.7,
)
our_graph.clear_attacked()
def do_graph_animation(output_file_name: str, in_graph: Graph, frame_count: int, layout): def do_graph_animation(output_file_name: str, in_graph: Graph, frame_count: int):
g_repr = nx.Graph() g_repr = nx.Graph()
# Convert our graph class into tuples understood by networkx # Convert our graph class into tuples understood by networkx
g_repr.add_edges_from([e.as_tuple() for e in in_graph.edges]) g_repr.add_edges_from([e.as_tuple() for e in in_graph.edges])
layout = layout(g_repr) layout = nx.spring_layout(g_repr)
fig, ax = plt.subplots() fig, ax = plt.subplots()
fig.set_figwidth(8) anim = animation.FuncAnimation(fig, update, frames=frame_count, fargs=(layout, g_repr, ax, in_graph))
fig.set_figheight(8)
anim = animation.FuncAnimation(
fig, update, frames=frame_count, interval=500, fargs=(layout, g_repr, ax, in_graph)
)
anim.save(output_file_name) anim.save(output_file_name)
plt.style.use('seaborn')
plt.show() plt.show()
def degree_avg(edges, digits=2): def bus_network(n=30) -> Graph:
degrees = {}
for e in edges:
degrees[e.node_a] = degrees.get(e.node_a, 0) + 1
degrees[e.node_b] = degrees.get(e.node_b, 0) + 1
return round(mean(degrees.values()), digits)
def bus_network(n=30, infected_idx=0) -> tuple[Graph, float, int]:
network = Graph() network = Graph()
nodes = [Node() for _ in range(n)] nodes = [Node() for _ in range(n)]
nodes[infected_idx].is_infected = True
edges = [Edge(nodes[i], nodes[i + 1], 1.0) for i in range(n - 1)] edges = [Edge(nodes[i], nodes[i + 1], 1.0) for i in range(n - 1)]
network.add_edges(edges) network.add_edges(edges)
return network, degree_avg(edges), n return network
def star_network(cluster_count=5, starsize=6, use_weights=False) -> tuple[Graph, float, int]: def rank_avg(edges, digits=2):
ranks = {}
for e in edges:
ranks[e.node_a] = ranks.get(e.node_a, 0) + 1
ranks[e.node_b] = ranks.get(e.node_b, 0) + 1
return round(mean(ranks.values()), digits)
def star_network(cluster_count=5, starsize=6) -> tuple[Graph, float]:
node_count = cluster_count + cluster_count * starsize + 1 node_count = cluster_count + cluster_count * starsize + 1
nodes = [Node() for _ in range(node_count)] nodes = [Node() for _ in range(node_count)]
nodes[starsize-1].is_infected = True
edges = [] edges = []
for x in range(cluster_count): for x in range(cluster_count):
center_node = x * starsize + x center_node = x * starsize + x
vulnerability = 1.0 if not use_weights else max(1.0, starsize) edges += [Edge(nodes[center_node], nodes[i], 1.0) for i in range(center_node + 1, center_node + starsize + 1)]
edges += [Edge(nodes[center_node], nodes[i], vulnerability) for i in range(center_node + 1, center_node + starsize + 1)]
edges.append(Edge(nodes[-1], nodes[center_node], 1.0)) edges.append(Edge(nodes[-1], nodes[center_node], 1.0))
network = Graph() network = Graph()
network.add_edges(edges) network.add_edges(edges)
return network, degree_avg(edges), node_count return network, rank_avg(edges)
def ring_network(n=30) -> tuple[Graph, float, int]:
network = Graph()
nodes = [Node() for _ in range(n)]
nodes[0].is_infected = True
edges = [Edge(nodes[i], nodes[i + 1], 1.0) for i in range(n - 1)]
end_edge = Edge(nodes[n - 1], nodes[0], 1.0)
edges.append(end_edge)
network.add_edges(edges)
return network, degree_avg(edges), n
def summary(average_degrees: list[float], propagation_speeds: list[float]) -> None:
fig, ax = plt.subplots()
ax.plot(average_degrees, propagation_speeds)
ax.set(xlabel='Average degree', ylabel='Propagation speed', title='Summary')
fig.savefig("summary.png")
plt.show()
def bus_experiment():
degrees = []
speeds = []
sizes = [0, 8, 15]
for i in sizes:
bus, bus_avg_degree, node_count = bus_network(20 + i)
do_graph_animation(f'bus{i}.gif', bus, 90, nx.spring_layout)
speeds.append(bus.rounds_survived / node_count)
degrees.append(bus_avg_degree)
print(f"\n{node_count} NODE bus")
print(f"average degree = {bus_avg_degree}")
print(f"propagation speed = {round(speeds[-1], 2)}")
print(f"bus{i} rounds survived = {bus.rounds_survived + 1}")
summary(degrees, speeds)
def ring_experiment():
degrees = []
speeds = []
sizes = [0, 8, 15]
for i in sizes:
ring, ring_avg_degree, node_count = ring_network(20 + i)
do_graph_animation(f'ring{i}.gif', ring, 90, nx.circular_layout)
speeds.append(ring.rounds_survived / node_count)
degrees.append(ring_avg_degree)
print(f"\n{node_count} NODE ring")
print(f"average degree = {ring_avg_degree}")
print(f"propagation speed = {round(speeds[-1], 2)}")
print(f"ring{i} rounds survived = {ring.rounds_survived + 1}")
def star_experiment():
degrees = []
speeds = []
sizes = range(0, 8, 2)
for i in sizes:
star, star_avg_degree, node_count = star_network(cluster_count=3 + i, starsize=3 + i)
do_graph_animation(f'star{i}.gif', star, 120, nx.kamada_kawai_layout)
speeds.append(star.rounds_survived / node_count)
degrees.append(star_avg_degree)
print(f"\n{node_count} NODE STAR")
print(f"average degree = {star_avg_degree}")
print(f"propagation speed = {round(speeds[-1], 2)}")
print(f"star{i} rounds survived = {star.rounds_survived + 1}")
summary(degrees, speeds)
def weighted_star_experiment():
degrees = []
speeds = []
sizes = range(0, 8, 2)
for i in sizes:
star, star_avg_degree, node_count = star_network(cluster_count=3 + i, starsize=3 + i, use_weights=True)
do_graph_animation(f'star_weighted{i}.gif', star, 120, nx.kamada_kawai_layout)
speeds.append(star.rounds_survived / node_count)
degrees.append(star_avg_degree)
print(f"\n{node_count} NODE STAR")
print(f"average degree = {star_avg_degree}")
print(f"propagation speed = {round(speeds[-1], 2)}")
print(f"star{i} rounds survived = {star.rounds_survived + 1}")
summary(degrees, speeds)
def main(): def main():
bus_experiment() network = Graph()
ring_experiment() nodes = [Node(True), Node(), Node(), Node(True), Node()]
star_experiment()
weighted_star_experiment() network.add_edges([
Edge(nodes[1], nodes[0], 0.02),
Edge(nodes[1], nodes[2], 0.2),
Edge(nodes[2], nodes[0], 0.7),
Edge(nodes[3], nodes[2], 0.2),
Edge(nodes[3], nodes[1], 0.2),
Edge(nodes[4], nodes[3], 0.2)
])
do_graph_animation('test.gif', network, 5)
bus = bus_network()
do_graph_animation('bus.gif', bus, 5)
star, star_avg_rank = star_network()
do_graph_animation('star.gif', star, 5)
if __name__ == "__main__": if __name__ == "__main__":
main() main()