1
0
mirror of https://github.com/gsi-upm/soil synced 2025-08-24 03:52:20 +00:00

black formatting

This commit is contained in:
J. Fernando Sánchez
2022-10-17 20:23:57 +02:00
parent 227fdf050e
commit 880a9f2a1c
19 changed files with 431 additions and 329 deletions

View File

@@ -20,56 +20,83 @@ class TerroristSpreadModel(FSM, Geo):
def __init__(self, model=None, unique_id=0, state=()):
super().__init__(model=model, unique_id=unique_id, state=state)
self.information_spread_intensity = model.environment_params['information_spread_intensity']
self.terrorist_additional_influence = model.environment_params['terrorist_additional_influence']
self.prob_interaction = model.environment_params['prob_interaction']
self.information_spread_intensity = model.environment_params[
"information_spread_intensity"
]
self.terrorist_additional_influence = model.environment_params[
"terrorist_additional_influence"
]
self.prob_interaction = model.environment_params["prob_interaction"]
if self['id'] == self.civilian.id: # Civilian
if self["id"] == self.civilian.id: # Civilian
self.mean_belief = self.random.uniform(0.00, 0.5)
elif self['id'] == self.terrorist.id: # Terrorist
elif self["id"] == self.terrorist.id: # Terrorist
self.mean_belief = self.random.uniform(0.8, 1.00)
elif self['id'] == self.leader.id: # Leader
elif self["id"] == self.leader.id: # Leader
self.mean_belief = 1.00
else:
raise Exception('Invalid state id: {}'.format(self['id']))
if 'min_vulnerability' in model.environment_params:
self.vulnerability = self.random.uniform( model.environment_params['min_vulnerability'], model.environment_params['max_vulnerability'] )
else :
self.vulnerability = self.random.uniform( 0, model.environment_params['max_vulnerability'] )
raise Exception("Invalid state id: {}".format(self["id"]))
if "min_vulnerability" in model.environment_params:
self.vulnerability = self.random.uniform(
model.environment_params["min_vulnerability"],
model.environment_params["max_vulnerability"],
)
else:
self.vulnerability = self.random.uniform(
0, model.environment_params["max_vulnerability"]
)
@state
def civilian(self):
neighbours = list(self.get_neighboring_agents(agent_class=TerroristSpreadModel))
if len(neighbours) > 0:
# Only interact with some of the neighbors
interactions = list(n for n in neighbours if self.random.random() <= self.prob_interaction)
influence = sum( self.degree(i) for i in interactions )
mean_belief = sum( i.mean_belief * self.degree(i) / influence for i in interactions )
mean_belief = mean_belief * self.information_spread_intensity + self.mean_belief * ( 1 - self.information_spread_intensity )
self.mean_belief = mean_belief * self.vulnerability + self.mean_belief * ( 1 - self.vulnerability )
interactions = list(
n for n in neighbours if self.random.random() <= self.prob_interaction
)
influence = sum(self.degree(i) for i in interactions)
mean_belief = sum(
i.mean_belief * self.degree(i) / influence for i in interactions
)
mean_belief = (
mean_belief * self.information_spread_intensity
+ self.mean_belief * (1 - self.information_spread_intensity)
)
self.mean_belief = mean_belief * self.vulnerability + self.mean_belief * (
1 - self.vulnerability
)
if self.mean_belief >= 0.8:
return self.terrorist
@state
def leader(self):
self.mean_belief = self.mean_belief ** ( 1 - self.terrorist_additional_influence )
for neighbour in self.get_neighboring_agents(state_id=[self.terrorist.id, self.leader.id]):
self.mean_belief = self.mean_belief ** (1 - self.terrorist_additional_influence)
for neighbour in self.get_neighboring_agents(
state_id=[self.terrorist.id, self.leader.id]
):
if self.betweenness(neighbour) > self.betweenness(self):
return self.terrorist
@state
def terrorist(self):
neighbours = self.get_agents(state_id=[self.terrorist.id, self.leader.id],
agent_class=TerroristSpreadModel,
limit_neighbors=True)
neighbours = self.get_agents(
state_id=[self.terrorist.id, self.leader.id],
agent_class=TerroristSpreadModel,
limit_neighbors=True,
)
if len(neighbours) > 0:
influence = sum( self.degree(n) for n in neighbours )
mean_belief = sum( n.mean_belief * self.degree(n) / influence for n in neighbours )
mean_belief = mean_belief * self.vulnerability + self.mean_belief * ( 1 - self.vulnerability )
self.mean_belief = self.mean_belief ** ( 1 - self.terrorist_additional_influence )
influence = sum(self.degree(n) for n in neighbours)
mean_belief = sum(
n.mean_belief * self.degree(n) / influence for n in neighbours
)
mean_belief = mean_belief * self.vulnerability + self.mean_belief * (
1 - self.vulnerability
)
self.mean_belief = self.mean_belief ** (
1 - self.terrorist_additional_influence
)
# Check if there are any leaders in the group
leaders = list(filter(lambda x: x.state.id == self.leader.id, neighbours))
@@ -82,21 +109,29 @@ class TerroristSpreadModel(FSM, Geo):
return self.leader
def ego_search(self, steps=1, center=False, node=None, **kwargs):
'''Get a list of nodes in the ego network of *node* of radius *steps*'''
"""Get a list of nodes in the ego network of *node* of radius *steps*"""
node = as_node(node if node is not None else self)
G = self.subgraph(**kwargs)
return nx.ego_graph(G, node, center=center, radius=steps).nodes()
def degree(self, node, force=False):
node = as_node(node)
if force or (not hasattr(self.model, '_degree')) or getattr(self.model, '_last_step', 0) < self.now:
if (
force
or (not hasattr(self.model, "_degree"))
or getattr(self.model, "_last_step", 0) < self.now
):
self.model._degree = nx.degree_centrality(self.G)
self.model._last_step = self.now
return self.model._degree[node]
def betweenness(self, node, force=False):
node = as_node(node)
if force or (not hasattr(self.model, '_betweenness')) or getattr(self.model, '_last_step', 0) < self.now:
if (
force
or (not hasattr(self.model, "_betweenness"))
or getattr(self.model, "_last_step", 0) < self.now
):
self.model._betweenness = nx.betweenness_centrality(self.G)
self.model._last_step = self.now
return self.model._betweenness[node]
@@ -114,17 +149,20 @@ class TrainingAreaModel(FSM, Geo):
def __init__(self, model=None, unique_id=0, state=()):
super().__init__(model=model, unique_id=unique_id, state=state)
self.training_influence = model.environment_params['training_influence']
if 'min_vulnerability' in model.environment_params:
self.min_vulnerability = model.environment_params['min_vulnerability']
else: self.min_vulnerability = 0
self.training_influence = model.environment_params["training_influence"]
if "min_vulnerability" in model.environment_params:
self.min_vulnerability = model.environment_params["min_vulnerability"]
else:
self.min_vulnerability = 0
@default_state
@state
def terrorist(self):
for neighbour in self.get_neighboring_agents(agent_class=TerroristSpreadModel):
if neighbour.vulnerability > self.min_vulnerability:
neighbour.vulnerability = neighbour.vulnerability ** ( 1 - self.training_influence )
neighbour.vulnerability = neighbour.vulnerability ** (
1 - self.training_influence
)
class HavenModel(FSM, Geo):
@@ -141,11 +179,12 @@ class HavenModel(FSM, Geo):
def __init__(self, model=None, unique_id=0, state=()):
super().__init__(model=model, unique_id=unique_id, state=state)
self.haven_influence = model.environment_params['haven_influence']
if 'min_vulnerability' in model.environment_params:
self.min_vulnerability = model.environment_params['min_vulnerability']
else: self.min_vulnerability = 0
self.max_vulnerability = model.environment_params['max_vulnerability']
self.haven_influence = model.environment_params["haven_influence"]
if "min_vulnerability" in model.environment_params:
self.min_vulnerability = model.environment_params["min_vulnerability"]
else:
self.min_vulnerability = 0
self.max_vulnerability = model.environment_params["max_vulnerability"]
def get_occupants(self, **kwargs):
return self.get_neighboring_agents(agent_class=TerroristSpreadModel, **kwargs)
@@ -158,14 +197,18 @@ class HavenModel(FSM, Geo):
for neighbour in self.get_occupants():
if neighbour.vulnerability > self.min_vulnerability:
neighbour.vulnerability = neighbour.vulnerability * ( 1 - self.haven_influence )
neighbour.vulnerability = neighbour.vulnerability * (
1 - self.haven_influence
)
return self.civilian
@state
def terrorist(self):
for neighbour in self.get_occupants():
if neighbour.vulnerability < self.max_vulnerability:
neighbour.vulnerability = neighbour.vulnerability ** ( 1 - self.haven_influence )
neighbour.vulnerability = neighbour.vulnerability ** (
1 - self.haven_influence
)
return self.terrorist
@@ -184,10 +227,10 @@ class TerroristNetworkModel(TerroristSpreadModel):
def __init__(self, model=None, unique_id=0, state=()):
super().__init__(model=model, unique_id=unique_id, state=state)
self.vision_range = model.environment_params['vision_range']
self.sphere_influence = model.environment_params['sphere_influence']
self.weight_social_distance = model.environment_params['weight_social_distance']
self.weight_link_distance = model.environment_params['weight_link_distance']
self.vision_range = model.environment_params["vision_range"]
self.sphere_influence = model.environment_params["sphere_influence"]
self.weight_social_distance = model.environment_params["weight_social_distance"]
self.weight_link_distance = model.environment_params["weight_link_distance"]
@state
def terrorist(self):
@@ -201,27 +244,48 @@ class TerroristNetworkModel(TerroristSpreadModel):
def update_relationships(self):
if self.count_neighboring_agents(state_id=self.civilian.id) == 0:
close_ups = set(self.geo_search(radius=self.vision_range, agent_class=TerroristNetworkModel))
step_neighbours = set(self.ego_search(self.sphere_influence, agent_class=TerroristNetworkModel, center=False))
neighbours = set(agent.id for agent in self.get_neighboring_agents(agent_class=TerroristNetworkModel))
close_ups = set(
self.geo_search(
radius=self.vision_range, agent_class=TerroristNetworkModel
)
)
step_neighbours = set(
self.ego_search(
self.sphere_influence,
agent_class=TerroristNetworkModel,
center=False,
)
)
neighbours = set(
agent.id
for agent in self.get_neighboring_agents(
agent_class=TerroristNetworkModel
)
)
search = (close_ups | step_neighbours) - neighbours
for agent in self.get_agents(search):
social_distance = 1 / self.shortest_path_length(agent.id)
spatial_proximity = ( 1 - self.get_distance(agent.id) )
prob_new_interaction = self.weight_social_distance * social_distance + self.weight_link_distance * spatial_proximity
if agent['id'] == agent.civilian.id and self.random.random() < prob_new_interaction:
spatial_proximity = 1 - self.get_distance(agent.id)
prob_new_interaction = (
self.weight_social_distance * social_distance
+ self.weight_link_distance * spatial_proximity
)
if (
agent["id"] == agent.civilian.id
and self.random.random() < prob_new_interaction
):
self.add_edge(agent)
break
def get_distance(self, target):
source_x, source_y = nx.get_node_attributes(self.G, 'pos')[self.id]
target_x, target_y = nx.get_node_attributes(self.G, 'pos')[target]
dx = abs( source_x - target_x )
dy = abs( source_y - target_y )
return ( dx ** 2 + dy ** 2 ) ** ( 1 / 2 )
source_x, source_y = nx.get_node_attributes(self.G, "pos")[self.id]
target_x, target_y = nx.get_node_attributes(self.G, "pos")[target]
dx = abs(source_x - target_x)
dy = abs(source_y - target_y)
return (dx**2 + dy**2) ** (1 / 2)
def shortest_path_length(self, target):
try:
return nx.shortest_path_length(self.G, self.id, target)
except nx.NetworkXNoPath:
return float('inf')
return float("inf")