# -*- coding: utf-8 -*- """revolutions_exploration.ipynb Automatically generated by Colaboratory. Original file is located at https://colab.research.google.com/drive/1omNn2hrbDL_s1qwCOr7ViaIjrRW61YDt """ # Commented out IPython magic to ensure Python compatibility. # %%capture # !pip install gradio # # !pip install gradio==3.50.2 # Commented out IPython magic to ensure Python compatibility. # %%capture # # !pip install cmocean # !pip install mesa # # !pip install opinionated import random import pandas as pd from mesa import Agent, Model from mesa.space import MultiGrid import networkx as nx from mesa.time import RandomActivation from mesa.datacollection import DataCollector import numpy as np import seaborn as sns import matplotlib.pyplot as plt import matplotlib as mpl import cmocean import tqdm import scipy as sp # from compress_pickle import dump, load from scipy.stats import beta # # %%capture # !pip install git+https://github.com/MNoichl/opinionated.git#egg=opinionated # # import opinionated import opinionated import matplotlib.pyplot as plt plt.style.use("opinionated_rc") #from opinionated.core import download_googlefont #download_googlefont('Quicksand', add_to_cache=True) #plt.rc('font', family='Quicksand') experiences = { 'dissident_experiences': [1,0,0], 'supporter_experiences': [1,1,1], } def apply_half_life_decay(data_list, half_life, decay_factors=None): steps = len(data_list) # Check if decay_factors are provided and are of the correct length if decay_factors is None or len(decay_factors) < steps: decay_factors = [0.5 ** (i / half_life) for i in range(steps)] decayed_list = [data_list[i] * decay_factors[steps - 1 - i] for i in range(steps)] return decayed_list half_life=20 decay_factors = [0.5 ** (i / half_life) for i in range(200)] def get_beta_mean_from_experience_dict(experiences, half_life=20,decay_factors=None): #note: precomputed decay supersedes halflife! eta = 1e-10 return beta.mean(sum(apply_half_life_decay(experiences['dissident_experiences'], half_life,decay_factors))+eta, sum(apply_half_life_decay(experiences['supporter_experiences'], half_life,decay_factors))+eta) def get_beta_sample_from_experience_dict(experiences, half_life=20,decay_factors=None): eta = 1e-10 # print(sum(apply_half_life_decay(experiences['dissident_experiences'], half_life))) # print(sum(apply_half_life_decay(experiences['supporter_experiences'], half_life))) return beta.rvs(sum(apply_half_life_decay(experiences['dissident_experiences'], half_life,decay_factors))+eta, sum(apply_half_life_decay(experiences['supporter_experiences'], half_life,decay_factors))+eta, size=1)[0] print(get_beta_mean_from_experience_dict(experiences,half_life,decay_factors)) print(get_beta_sample_from_experience_dict(experiences,half_life)) #@title Load network functionality def generate_community_points(num_communities, total_nodes, powerlaw_exponent=2.0, sigma=0.05, plot=False): """ This function generates points in 2D space, where points are grouped into communities. Each community is represented by a Gaussian distribution. Args: num_communities (int): Number of communities (gaussian distributions). total_nodes (int): Total number of points to be generated. powerlaw_exponent (float): The power law exponent for the powerlaw sequence. sigma (float): The standard deviation for the gaussian distributions. plot (bool): If True, the function plots the generated points. Returns: numpy.ndarray: An array of generated points. """ # Sample from a powerlaw distribution sequence = nx.utils.powerlaw_sequence(num_communities, powerlaw_exponent) # Normalize sequence to represent probabilities probabilities = sequence / np.sum(sequence) # Assign nodes to communities based on probabilities community_assignments = np.random.choice(num_communities, size=total_nodes, p=probabilities) # Calculate community_sizes from community_assignments community_sizes = np.bincount(community_assignments) # Ensure community_sizes has length equal to num_communities if len(community_sizes) < num_communities: community_sizes = np.pad(community_sizes, (0, num_communities - len(community_sizes)), 'constant') points = [] community_centers = [] # For each community for i in range(num_communities): # Create a random center for this community center = np.random.rand(2) community_centers.append(center) # Sample from Gaussian distributions with the center and sigma community_points = np.random.normal(center, sigma, (community_sizes[i], 2)) points.append(community_points) points = np.concatenate(points) # Optional plotting if plot: plt.figure(figsize=(8,8)) plt.scatter(points[:, 0], points[:, 1], alpha=0.5) # for center in community_centers: sns.kdeplot(x=points[:, 0], y=points[:, 1], levels=5, color="k", linewidths=1) # plt.xlim(0, 1) # plt.ylim(0, 1) plt.show() return points def graph_from_coordinates(coords, radius): """ This function creates a random geometric graph from an array of coordinates. Args: coords (numpy.ndarray): An array of coordinates. radius (float): A radius of circles or spheres. Returns: networkx.Graph: The created graph. """ # Create a KDTree for efficient query kdtree = sp.spatial.cKDTree(coords) edge_indexes = kdtree.query_pairs(radius) g = nx.Graph() g.add_nodes_from(list(range(len(coords)))) g.add_edges_from(edge_indexes) return g def plot_graph(graph, positions): """ This function plots a graph with the given positions. Args: graph (networkx.Graph): The graph to be plotted. positions (dict): A dictionary of positions for the nodes. """ plt.figure(figsize=(8,8)) pos_dict = {i: positions[i] for i in range(len(positions))} nx.draw_networkx_nodes(graph, pos_dict, node_size=30, node_color="#1a2340", alpha=0.7) nx.draw_networkx_edges(graph, pos_dict, edge_color="grey", width=1, alpha=1) plt.show() def ensure_neighbors(graph): """ Ensure that all nodes in a NetworkX graph have at least one neighbor. Parameters: graph (networkx.Graph): The NetworkX graph to check. Returns: networkx.Graph: The updated NetworkX graph where all nodes have at least one neighbor. """ nodes = list(graph.nodes()) for node in nodes: if len(list(graph.neighbors(node))) == 0: # The node has no neighbors, so select another node to connect it with other_node = random.choice(nodes) while other_node == node: # Make sure we don't connect the node to itself other_node = random.choice(nodes) graph.add_edge(node, other_node) return graph def compute_homophily(G,attr_name='attr'): same_attribute_edges = sum(G.nodes[n1][attr_name] == G.nodes[n2][attr_name] for n1, n2 in G.edges()) total_edges = G.number_of_edges() return same_attribute_edges / total_edges if total_edges > 0 else 0 def assign_initial_attributes(G, ratio,attr_name='attr'): nodes = list(G.nodes) random.shuffle(nodes) attr_boundary = int(ratio * len(nodes)) for i, node in enumerate(nodes): G.nodes[node][attr_name] = 0 if i < attr_boundary else 1 return G def distribute_attributes(G, target_homophily, seed=None, max_iter=10000, cooling_factor=0.9995,attr_name='attr'): random.seed(seed) current_homophily = compute_homophily(G,attr_name) temp = 1.0 for i in range(max_iter): # pick two random nodes with different attributes and swap their attributes nodes = list(G.nodes) random.shuffle(nodes) for node1, node2 in zip(nodes[::2], nodes[1::2]): if G.nodes[node1][attr_name] != G.nodes[node2][attr_name]: G.nodes[node1][attr_name], G.nodes[node2][attr_name] = G.nodes[node2][attr_name], G.nodes[node1][attr_name] break new_homophily = compute_homophily(G,attr_name) delta_homophily = new_homophily - current_homophily dir_factor = np.sign(target_homophily - current_homophily) # if the new homophily is closer to the target, or if the simulated annealing condition is met, accept the swap if abs(new_homophily - target_homophily) < abs(current_homophily - target_homophily) or \ (delta_homophily / temp < 700 and random.random() < np.exp(dir_factor * delta_homophily / temp)): current_homophily = new_homophily else: # else, undo the swap G.nodes[node1][attr_name], G.nodes[node2][attr_name] = G.nodes[node2][attr_name], G.nodes[node1][attr_name] temp *= cooling_factor # cool down return G def reindex_graph_to_match_attributes(G1, G2, attr_name): # Get a sorted list of nodes in G1 based on the attribute G1_sorted_nodes = sorted(G1.nodes(data=True), key=lambda x: x[1][attr_name]) # Get a sorted list of nodes in G2 based on the attribute G2_sorted_nodes = sorted(G2.nodes(data=True), key=lambda x: x[1][attr_name]) # Create a mapping from the G2 node IDs to the G1 node IDs mapping = {G2_node[0]: G1_node[0] for G2_node, G1_node in zip(G2_sorted_nodes, G1_sorted_nodes)} # Generate the new graph with the updated nodes G2_updated = nx.relabel_nodes(G2, mapping) return G2_updated ########################## def compute_mean(model): agent_estimations = [agent.estimation for agent in model.schedule.agents] return np.mean(agent_estimations) def compute_median(model): agent_estimations = [agent.estimation for agent in model.schedule.agents] return np.median(agent_estimations) def compute_std(model): agent_estimations = [agent.estimation for agent in model.schedule.agents] return np.std(agent_estimations) class PoliticalAgent(Agent): """An agent in the political model. Attributes: estimation (float): Agent's current expectation of political change. dissident (bool): True if the agent supports a regime change, False otherwise. networks_estimations (dict): A dictionary storing the estimations of the agent for each network. """ def __init__(self, unique_id, model, dissident): super().__init__(unique_id, model) self.experiences = { 'dissident_experiences': [1], 'supporter_experiences': [1], } # self.estimation = estimation self.estimations = [] self.estimation = .5 #hardcoded_mean, will change in first step if agent interacts. self.experiments = [] self.dissident = dissident # self.historical_estimations = [] def update_estimation(self, network_id): """Update the agent's estimation for a given network.""" # Get the neighbors from the network potential_partners = [self.model.schedule.agents[n] for n in self.model.networks[network_id]['network'].neighbors(self.unique_id)] current_estimate =get_beta_mean_from_experience_dict(self.experiences,half_life=self.model.half_life,decay_factors=self.model.decay_factors) self.estimations.append(current_estimate) self.estimation =current_estimate current_experiment = get_beta_sample_from_experience_dict(self.experiences,half_life=self.model.half_life, decay_factors=self.model.decay_factors) self.experiments.append(current_experiment) if potential_partners: partner = random.choice(potential_partners) if self.model.networks[network_id]['type'] == 'physical': if current_experiment >= self.model.threshold: if partner.dissident: # removed division by 100? self.experiences['dissident_experiences'].append(1) self.experiences['supporter_experiences'].append(0) else: self.experiences['dissident_experiences'].append(0) self.experiences['supporter_experiences'].append(1) partner.experiences['dissident_experiences'].append(1 * self.model.social_learning_factor) partner.experiences['supporter_experiences'].append(0) else: partner.experiences['dissident_experiences'].append(0) partner.experiences['supporter_experiences'].append(1 * self.model.social_learning_factor) # else: # pass # Only one network for the moment! elif self.model.networks[network_id]['type'] == 'social_media': if partner.dissident: # removed division by 100? self.experiences['dissident_experiences'].append(1 * self.model.social_media_factor) self.experiences['supporter_experiences'].append(0) else: self.experiences['dissident_experiences'].append(0) self.experiences['supporter_experiences'].append(1 * self.model.social_media_factor) # self.networks_estimations[network_id] = self.estimation def combine_estimations(self): # """Combine the estimations from all networks using a bounded confidence model.""" values = [list(d.values())[0] for d in self.current_estimations] if len(values) > 0: # Filter the network estimations based on the bounded confidence range within_range = [value for value in values if abs(self.estimation - value) <= self.model.bounded_confidence_range] # If there are any estimations within the range, update the estimation if len(within_range) > 0: self.estimation = np.mean(within_range) def step(self): """Agent step function which updates the estimation for each network and then combines the estimations.""" if not hasattr(self, 'current_estimations'): # agents might already have this attribute because they were partnered up in the past. self.current_estimations = [] for network_id in self.model.networks.keys(): self.update_estimation(network_id) self.combine_estimations() # self.historical_estimations.append(self.current_estimations) del self.current_estimations class PoliticalModel(Model): """A model of a political system with multiple interacting agents. Attributes: networks (dict): A dictionary of networks with network IDs as keys and NetworkX Graph objects as values. """ def __init__(self, n_agents, networks, share_regime_supporters, # initial_expectation_of_change, threshold, social_learning_factor=1,social_media_factor=1, # one for equal learning, lower gets discounted half_life=20, print_agents=False, print_frequency=30, early_stopping_steps=20, early_stopping_range=0.01, agent_reporters=True,intervention_list=[],randomID=False): self.num_agents = n_agents self.threshold = threshold self.social_learning_factor = social_learning_factor self.social_media_factor = social_media_factor self.print_agents_state = print_agents self.half_life = half_life self.intervention_list = intervention_list self.model_id = randomID self.print_frequency = print_frequency self.early_stopping_steps = early_stopping_steps self.early_stopping_range = early_stopping_range self.mean_estimations = [] self.decay_factors = [0.5 ** (i / self.half_life) for i in range(500)] # Nte this should be larger than # we could use this for early stopping! self.running = True self.share_regime_supporters = share_regime_supporters self.schedule = RandomActivation(self) self.networks = networks # Assign dissident as argument to networks, compute homophilies, and match up the networks so that the same id leads to the same atrribute for i, this_network in enumerate(self.networks): self.networks[this_network]["network"] = assign_initial_attributes(self.networks[this_network]["network"],self.share_regime_supporters,attr_name='dissident') if 'homophily' in self.networks[this_network]: self.networks[this_network]["network"] = distribute_attributes(self.networks[this_network]["network"], self.networks[this_network]['homophily'], max_iter=5000, cooling_factor=0.995,attr_name='dissident') self.networks[this_network]['network_data_to_keep']['actual_homophily'] = compute_homophily(self.networks[this_network]["network"],attr_name='dissident') if i>0: self.networks[this_network]["network"] = reindex_graph_to_match_attributes(self.networks[next(iter(self.networks))]["network"], self.networks[this_network]["network"], 'dissident') # print(self.networks) for i in range(self.num_agents): # estimation = random.normalvariate(initial_expectation_of_change, 0.2) We set a flat prior now agent = PoliticalAgent(i, self, self.networks[next(iter(self.networks))]["network"].nodes(data=True)[i]['dissident']) self.schedule.add(agent) # Should we update to the real share here?! #################### # Keep the attributes in the model and define model reporters model_reporters = { "Mean": compute_mean, "Median": compute_median, "STD": compute_std } for this_network in self.networks: if 'network_data_to_keep' in self.networks[this_network]: for key, value in self.networks[this_network]['network_data_to_keep'].items(): attr_name = this_network + '_' + key setattr(self, attr_name, value) # Define a reporter function for this attribute def reporter(model, attr_name=attr_name): return getattr(model, attr_name) # Add the reporter function to the dictionary model_reporters[attr_name] = reporter # Initialize DataCollector with the dynamic model reporters if agent_reporters: self.datacollector = DataCollector( model_reporters=model_reporters, agent_reporters={"Estimation": "estimation", "Dissident": "dissident"}#, "Historical Estimations": "historical_estimations"} ) else: self.datacollector = DataCollector( model_reporters=model_reporters ) def step(self): """Model step function which activates the step function of each agent.""" self.datacollector.collect(self) # Collect data # do interventions, if present: for this_intervention in self.intervention_list: # print(this_intervention) if this_intervention['time'] == len(self.mean_estimations): if this_intervention['type'] == 'threshold_adjustment': self.threshold = max(0, min(1, self.threshold + this_intervention['strength'])) if this_intervention['type'] == 'share_adjustment': target_supporter_share = max(0, min(1, self.share_regime_supporters + this_intervention['strength'])) agents = [self.schedule._agents[i] for i in self.schedule._agents] current_supporters = sum(not agent.dissident for agent in agents) total_agents = len(agents) current_share = current_supporters / total_agents # Calculate the number of agents to change required_supporters = int(target_supporter_share * total_agents) agents_to_change = abs(required_supporters - current_supporters) if current_share < target_supporter_share: # Not enough supporters, need to increase dissidents = [agent for agent in agents if agent.dissident] for agent in random.sample(dissidents, agents_to_change): agent.dissident = False elif current_share > target_supporter_share: # Too many supporters, need to reduce supporters = [agent for agent in agents if not agent.dissident] for agent in random.sample(supporters, agents_to_change): agent.dissident = True # print(self.threshold) if this_intervention['type'] == 'social_media_adjustment': self.social_media_factor = max(0, min(1, self.social_media_factor + this_intervention['strength'])) self.schedule.step() current_mean_estimation = compute_mean(self) self.mean_estimations.append(current_mean_estimation) # Implement the early stopping criteria if len(self.mean_estimations) >= self.early_stopping_steps: recent_means = self.mean_estimations[-self.early_stopping_steps:] if max(recent_means) - min(recent_means) < self.early_stopping_range: # if self.print_agents_state: # print('Early stopping at: ', self.schedule.steps) # self.print_agents() self.running = False # if self.print_agents_state and (self.schedule.steps % self.print_frequency == 0 or self.schedule.steps == 1): # print(self.schedule.steps) # self.print_agents() # def run_simulation(n_agents=300, share_regime_supporters=0.4, threshold=0.5, social_learning_factor=1, simulation_steps=400, half_life=20): # # Helper functions like graph_from_coordinates, ensure_neighbors should be defined outside this function # # Complete graph # G = nx.complete_graph(n_agents) # # Networks dictionary # networks = { # "physical": {"network": G, "type": "physical", "positions": nx.circular_layout(G)}#kamada_kawai # } # # Intervention list # intervention_list = [ ] # # Initialize the model # model = PoliticalModel(n_agents, networks, share_regime_supporters, threshold, # social_learning_factor, half_life=half_life, print_agents=False, print_frequency=50, agent_reporters=True, intervention_list=intervention_list) # # Run the model # for _ in tqdm.tqdm_notebook(range(simulation_steps)): # Run for specified number of steps # model.step() # return model # # Example usage # radius=.09 # physical_graph_points = np.random.rand(100, 2) # physical_graph = graph_from_coordinates(physical_graph_points, radius) # physical_graph = nx.convert_node_labels_to_integers(ensure_neighbors(physical_graph)) # # unconnected nodes: link or drop? # networks = { # "physical": {"network": physical_graph, "type": "physical", "positions": physical_graph_points, 'network_data_to_keep':{'radius':radius},'homophily':0. }} # model = PoliticalModel(100, networks, .5, .5,.5, half_life=20, print_agents=False, print_frequency=50, agent_reporters=True, intervention_list=[]) # for _ in tqdm.tqdm_notebook(range(40)): # Run for specified number of steps # model.step() # import matplotlib.pyplot as plt # import pandas as pd # # Assuming 'model' is defined and has a datacollector with the necessary data # agent_df = model.datacollector.get_agent_vars_dataframe().reset_index() # # Pivot the dataframe for Estimation # agent_df_pivot = agent_df.pivot(index='Step', columns='AgentID', values='Estimation') # # Create the result plot # run_plot, ax = plt.subplots(figsize=(12, 8)) # # Define colors for Dissident and Supporter # colors = {1: '#d6a44b', 0: '#1b4968'} # 1 for Dissident, 0 for Supporter # labels = {1: 'Dissident', 0: 'Supporter'} # legend_handles = [] # # Plot each agent's data # for agent_id in agent_df_pivot.columns: # # Get the agent type (Dissident or Supporter) # agent_type = agent_df[agent_df['AgentID'] == agent_id]['Dissident'].iloc[0] # # Plot # line, = plt.plot(agent_df_pivot.index, agent_df_pivot[agent_id], color=colors[agent_type], alpha=0.1) # # Compute and plot the mean estimation for each group # for agent_type, color in colors.items(): # mean_estimation = agent_df_pivot.loc[:, agent_df[agent_df['Dissident'] == agent_type]['AgentID']].mean(axis=1) # plt.plot(mean_estimation.index, mean_estimation, color=color, linewidth=2, label=f'{labels[agent_type]}') # # Set the plot title and labels # plt.title('Agent Estimation Over Time', loc='right') # plt.xlabel('Time step') # plt.ylabel('Estimation') # # Add legend # plt.legend(loc='lower right') # plt.show() import PIL def run_and_plot_simulation(separate_agent_types=False,n_agents=300, share_regime_supporters=0.4, threshold=0.5, social_learning_factor=1, simulation_steps=40, half_life=20, phys_network_radius=.06, powerlaw_exponent=3,physical_network_type='physical_network_type_fully_connected', introduce_physical_homophily_true_false=False,physical_homophily=.5, introduce_social_media_homophily_true_false=False,social_media_homophily=5,social_media_network_type_random_geometric_radius=.07,social_media_network_type_powerlaw_exponent=3, social_media_network_type='Powerlaw',use_social_media_network=False): print(physical_network_type) networks = {} # Set up physical network: if physical_network_type == 'Fully Connected': G = nx.complete_graph(n_agents) networks['physical'] = {"network": G, "type": "physical", "positions": nx.circular_layout(G)} elif physical_network_type == "Powerlaw": s = nx.utils.powerlaw_sequence(n_agents, powerlaw_exponent) #100 nodes, power-law exponent 2.5 G = nx.expected_degree_graph(s, selfloops=False) G = nx.convert_node_labels_to_integers(ensure_neighbors(G)) networks['physical'] = {"network": G, "type": "physical", "positions": nx.kamada_kawai_layout(G)} elif physical_network_type == "Random Geometric": physical_graph_points = np.random.rand(n_agents, 2) G = graph_from_coordinates(physical_graph_points, phys_network_radius) G = nx.convert_node_labels_to_integers(ensure_neighbors(G)) networks['physical'] = {"network": G, "type": "physical", "positions": physical_graph_points} if introduce_physical_homophily_true_false: networks['physical']['homophily'] = physical_homophily networks['physical']['network_data_to_keep'] = {} # Set up social media network: if use_social_media_network: if social_media_network_type == 'Fully Connected': G = nx.complete_graph(n_agents) networks['social_media'] = {"network": G, "type": "social_media", "positions": nx.circular_layout(G)} elif social_media_network_type == "Powerlaw": s = nx.utils.powerlaw_sequence(n_agents, social_media_network_type_powerlaw_exponent) # 100 nodes, power-law exponent adjusted for social media G = nx.expected_degree_graph(s, selfloops=False) G = nx.convert_node_labels_to_integers(ensure_neighbors(G)) networks['social_media'] = {"network": G, "type": "social_media", "positions": nx.kamada_kawai_layout(G)} elif social_media_network_type == "Random Geometric": social_media_graph_points = np.random.rand(n_agents, 2) G = graph_from_coordinates(social_media_graph_points, social_media_network_type_random_geometric_radius) G = nx.convert_node_labels_to_integers(ensure_neighbors(G)) networks['social_media'] = {"network": G, "type": "social_media", "positions": social_media_graph_points} if introduce_social_media_homophily_true_false: networks['social_media']['homophily'] = social_media_homophily networks['social_media']['network_data_to_keep'] = {} intervention_list = [ ] # Initialize the model model = PoliticalModel(n_agents, networks, share_regime_supporters, threshold, social_learning_factor, half_life=half_life, print_agents=False, print_frequency=50, agent_reporters=True, intervention_list=intervention_list) # Run the model for _ in tqdm.tqdm_notebook(range(simulation_steps)): # Run for specified number of steps model.step() agent_df = model.datacollector.get_agent_vars_dataframe().reset_index() # Pivot the dataframe agent_df_pivot = agent_df.pivot(index='Step', columns='AgentID', values='Estimation') # Create the esult-plot run_plot, ax = plt.subplots(figsize=(12, 8)) if not separate_agent_types: for column in agent_df_pivot.columns: plt.plot(agent_df_pivot.index, agent_df_pivot[column], color='gray', alpha=0.1) # Compute and plot the mean estimation mean_estimation = agent_df_pivot.mean(axis=1) plt.plot(mean_estimation.index, mean_estimation, color='black', linewidth=2) else: # Define colors for Dissident and Supporter colors = {1: '#d6a44b', 0: '#1b4968'} # 1 for Dissident, 0 for Supporter labels = {1: 'Dissident', 0: 'Supporter'} legend_handles = [] # Plot each agent's data for agent_id in agent_df_pivot.columns: # Get the agent type (Dissident or Supporter) agent_type = agent_df[agent_df['AgentID'] == agent_id]['Dissident'].iloc[0] # Plot line, = plt.plot(agent_df_pivot.index, agent_df_pivot[agent_id], color=colors[agent_type], alpha=0.1) # Compute and plot the mean estimation for each group for agent_type, color in colors.items(): mean_estimation = agent_df_pivot.loc[:, agent_df[agent_df['Dissident'] == agent_type]['AgentID']].mean(axis=1) plt.plot(mean_estimation.index, mean_estimation, color=color, linewidth=2, label=f'{labels[agent_type]}') plt.legend(loc='lower right') # Set the plot title and labels plt.title('Agent Estimation Over Time', loc='right') plt.xlabel('Time step') plt.ylabel('Estimation') plt.savefig('run_plot.png' ,bbox_inches='tight', dpi =400, transparent=True) run_plot = PIL.Image.open('run_plot.png').convert('RGBA') # Create the network-plot n_networks = len(networks) network_plot, axs = plt.subplots(1, n_networks, figsize=( 9.5 * n_networks,8)) if n_networks == 1: axs = [axs] estimations = {} for agent in model.schedule.agents: estimations[agent.unique_id] = agent.estimation for idx, (network_id, network_dict) in enumerate(networks.items()): network = network_dict['network'] # Collect estimations and set the node attributes nx.set_node_attributes(network, estimations, 'estimation') # Use the positions provided in the network dict if available if 'positions' in network_dict: pos = network_dict['positions'] else: pos = nx.kamada_kawai_layout(network) # Draw the network with nodes colored by their estimation values node_colors = [estimations[node] for node in network.nodes] axs[idx].set_title(f'Network: {network_id}', loc='right') # nx.draw(network, pos, node_size=50, node_color=node_colors, # cmap=cmocean.tools.crop_by_percent(cmocean.cm.curl, 20, which='both', N=None), # with_labels=False,vmin=0, vmax=1, ax=axs[idx]) # Drawing nodes nx.draw_networkx_nodes(network, pos, node_size=50, node_color=node_colors, cmap=cmocean.tools.crop_by_percent(cmocean.cm.curl, 20, which='both', N=None), vmin=0, vmax=1, ax=axs[idx]) # Drawing edges with semi-transparency nx.draw_networkx_edges(network, pos, alpha=0.3, ax=axs[idx]) # alpha value for semi-transparency # Create a dummy ScalarMappable with the same colormap sm = mpl.cm.ScalarMappable(cmap=cmocean.tools.crop_by_percent(cmocean.cm.curl, 20, which='both', N=None), norm=plt.Normalize(vmin=0, vmax=1)) sm.set_array([]) network_plot.colorbar(sm, ax=axs[idx]) plt.savefig('network_plot.png' ,bbox_inches='tight', dpi =400, transparent=True) network_plot = PIL.Image.open('network_plot.png').convert('RGBA') return run_plot, network_plot # run_and_plot_simulation(n_agents=300, share_regime_supporters=0.4, threshold=0.5, social_learning_factor=1, simulation_steps=40, half_life=20) import gradio as gr import matplotlib.pyplot as plt # Gradio interface with gr.Blocks(theme=gr.themes.Monochrome()) as demo: with gr.Column(): gr.Markdown("""# Simulate the emergence of social movements Vary the parameters below, and click 'Run Simulation' to run. """) with gr.Row(): with gr.Column(): with gr.Group(): separate_agent_types = gr.Checkbox(value=False, label="Separate agent types in plot") # Sliders for each parameter n_agents_slider = gr.Slider(minimum=100, maximum=500, step=10, label="Number of Agents", value=150) share_regime_slider = gr.Slider(minimum=0.0, maximum=1.0, step=0.01, label="Share of Regime Supporters", value=0.4) threshold_slider = gr.Slider(minimum=0.0, maximum=1.0, step=0.01, label="Threshold", value=0.5) social_learning_slider = gr.Slider(minimum=0.0, maximum=2.0, step=0.1, label="Social Learning Factor", value=1.0) steps_slider = gr.Slider(minimum=10, maximum=100, step=5, label="Simulation Steps", value=40) half_life_slider = gr.Slider(minimum=5, maximum=50, step=5, label="Half-Life", value=20) # physical network settings with gr.Group(): # with gr.Group(): gr.Markdown("""**Physical Network Settings:**""") # Define the checkbox introduce_physical_homophily_true_false = gr.Checkbox(value=False, label="Stipulate Homophily") # Define a group to hold the slider with gr.Group(visible=False) as homophily_group: physical_homophily = gr.Slider(0, 1, label="Homophily", info='How much homophily to stipulate.') # Function to update the visibility of the group based on the checkbox def update_homophily_group_visibility(checkbox_state): return { homophily_group: gr.Group(visible=checkbox_state) # The group visibility depends on the checkbox } # Bind the function to the checkbox introduce_physical_homophily_true_false.change( update_homophily_group_visibility, inputs=introduce_physical_homophily_true_false, outputs=homophily_group ) physical_network_type = gr.Dropdown(label="Physical Network Type", value="Fully Connected",choices=["Fully Connected", "Random Geometric","Powerlaw"])#value ="Fully Connected" with gr.Group(visible=True) as physical_network_type_fully_connected_group: gr.Markdown("""""") with gr.Group(visible=False) as physical_network_type_random_geometric_group: physical_network_type_random_geometric_radius = gr.Slider(minimum=.0, maximum=.5,label="Radius") with gr.Group(visible=False) as physical_network_type_powerlaw_group: physical_network_type_random_geometric_powerlaw_exponent = gr.Slider(minimum=.0, maximum=5.2,label="Powerlaw Exponent") def update_sliders(option): return { physical_network_type_fully_connected_group: gr.Group(visible=option == "Fully Connected"), physical_network_type_random_geometric_group: gr.Group(visible=option == "Random Geometric"), physical_network_type_powerlaw_group: gr.Group(visible=option == "Powerlaw") } physical_network_type.change(update_sliders, inputs=physical_network_type, outputs=[physical_network_type_fully_connected_group, physical_network_type_random_geometric_group, physical_network_type_powerlaw_group]) # social media settings: use_social_media_network = gr.Checkbox(value=False, label="Use social media network") with gr.Group(visible=False) as social_media_group: gr.Markdown("""**Social Media Network Settings:**""") # Define the checkbox for social media network social_media_factor = gr.Slider(0, 2, label="Social Media Factor", info='How strongly to weigh the social media network against learning in the real world.') introduce_social_media_homophily_true_false = gr.Checkbox(value=False, label="Stipulate Homophily") # Define a group to hold the slider for social media network with gr.Group(visible=False) as social_media_homophily_group: social_media_homophily = gr.Slider(0, 1, label="Homophily", info='How much homophily to stipulate in social media network.') # Function to update the visibility of the group based on the checkbox for social media network def update_social_media_homophily_group_visibility(checkbox_state): return { social_media_homophily_group: gr.Group(visible=checkbox_state) # The group visibility depends on the checkbox for social media network } # Bind the function to the checkbox for social media network introduce_social_media_homophily_true_false.change( update_social_media_homophily_group_visibility, inputs=introduce_social_media_homophily_true_false, outputs=social_media_homophily_group ) social_media_network_type = gr.Dropdown(label="Social Media Network Type", value="Fully Connected", choices=["Fully Connected", "Random Geometric", "Powerlaw"]) with gr.Group(visible=True) as social_media_network_type_fully_connected_group: gr.Markdown("""""") with gr.Group(visible=False) as social_media_network_type_random_geometric_group: social_media_network_type_random_geometric_radius = gr.Slider(minimum=0.0, maximum=0.5, label="Radius") with gr.Group(visible=False) as social_media_network_type_powerlaw_group: social_media_network_type_powerlaw_exponent = gr.Slider(minimum=0.0, maximum=5.2, label="Powerlaw Exponent") def update_social_media_network_sliders(option): return { social_media_network_type_fully_connected_group: gr.Group(visible=option == "Fully Connected"), social_media_network_type_random_geometric_group: gr.Group(visible=option == "Random Geometric"), social_media_network_type_powerlaw_group: gr.Group(visible=option == "Powerlaw") } social_media_network_type.change(update_social_media_network_sliders, inputs=social_media_network_type, outputs=[social_media_network_type_fully_connected_group, social_media_network_type_random_geometric_group, social_media_network_type_powerlaw_group]) def update_social_media_group_visibility(checkbox_state): return {social_media_group: gr.Group(visible=checkbox_state) } use_social_media_network.change(update_social_media_group_visibility,inputs=use_social_media_network,outputs=social_media_group) with gr.Column(): # Button to trigger the simulation button = gr.Button("Run Simulation") plot_output = gr.Image(label="Simulation Result") network_output = gr.Image(label="Networks") # gr.Button(value="Download Results",link="/file=network_plot.png") # Function to call when button is clicked def run_simulation_and_plot(*args): fig = run_and_plot_simulation(*args) return fig # Setting up the button click event button.click( run_simulation_and_plot, inputs=[separate_agent_types,n_agents_slider, share_regime_slider, threshold_slider, social_learning_slider, steps_slider, half_life_slider, physical_network_type_random_geometric_radius,physical_network_type_random_geometric_powerlaw_exponent,physical_network_type, introduce_physical_homophily_true_false,physical_homophily, introduce_social_media_homophily_true_false,social_media_homophily,social_media_network_type_random_geometric_radius,social_media_network_type_powerlaw_exponent,social_media_network_type,use_social_media_network], outputs=[plot_output,network_output] ) # Launch the interface if __name__ == "__main__": demo.launch(debug=True)