import json import math import random from collections import Counter import networkx as nx import numpy as np def calculate_pws(c, d): """ Calculates the rewiring probability pws for WS networks based on the desired clustering coefficient c and the average degree d. """ if d <= 2: raise ValueError("The average degree d must be greater than 2.") zähler = 4 * c * (d - 1) nenner = 3 * (d - 2) inner_term = zähler / nenner pws = 1 - math.pow(inner_term, 1 / 3) return pws def bonacich_centrality(G, eta): """ Calculate the Bonacich centrality for nodes in a graph. Args: G (networkx.Graph): The input graph to analyze. eta (float): The attenuation factor, typically scaled relative to the largest eigenvalue. Returns: numpy.ndarray | str: An array of centrality scores for each node, or an error message if the matrix inversion fails. """ A = nx.to_numpy_array(G) n = A.shape[0] I = np.eye(n) ones = np.ones(n) eigenvalues = np.linalg.eigvals(A) lambda_max = np.max(np.real(eigenvalues)) beta_eff = eta / lambda_max try: # Term (I - (eta/lambda_max) * A) term_to_invert = I - beta_eff * A # Formula (8) from Gao et al. (2024): (I - beta_eff * A)^-1 * A * 1n centrality = np.linalg.solve(term_to_invert, A @ ones) return centrality except np.linalg.LinAlgError: return "Calculation error: Matrix is singular (eta may be too close to 1)." def get_neighbors_for_node(matrix_json, node_id): """Returns a list of node_ids that are connected to the given node_id.""" matrix = json.loads(matrix_json) neighbors = [] # In an adjacency matrix, matrix[i][j] == 1 means a connection for target_node_id, connected in enumerate(matrix[node_id]): if connected == 1: neighbors.append(target_node_id) return neighbors def calculate_bot_decisions(group): """ Calculates decisions for the 8 bots based on the 2-phase rule. """ first_player = group.get_players()[0] first_participant = first_player.participant # 1. Load context bot_nodes = json.loads(first_participant.bot_nodes) network_map = json.loads(first_participant.network_map) group_key = first_participant.code node_map = first_player.session.vars['group_maps'][group_key] round_num = group.round_number current_bot_decisions = {} # --- PHASE 2 LOGIC --- if first_participant.in_phase_2: # Pick the alternative to the established option # If Phase 1 majority was 1, bots now pick 0 (and vice versa) target_choice = 1 if first_participant.established_option == 0 else 0 for bot_id in bot_nodes: current_bot_decisions[bot_id] = target_choice # --- PHASE 1 LOGIC --- else: for bot_id in bot_nodes: if round_num == 1: # Initial round: Random current_bot_decisions[bot_id] = random.randint(0, 1) else: # Majority of neighbors from PREVIOUS round neighbors = network_map[str(bot_id)] prev_choices = [] for n_id in neighbors: # Check if neighbor is a bot or human if n_id in bot_nodes: prev_bot_data = json.loads(group.in_round(round_num - 1).bot_decisions) prev_choices.append(prev_bot_data[str(n_id)]) else: val = node_map.get(str(n_id)) if val is not None: p_id = int(val) neighbor_prev_player = group.subsession.in_round(round_num - 1).get_players()[p_id - 1] prev_choices.append(neighbor_prev_player.decision) # Determine majority counts = Counter(prev_choices) if counts[0] != counts[1]: # Pick the majority (0 or 1) current_bot_decisions[bot_id] = counts.most_common(1)[0][0] else: # Tie: Pick own choice from previous round prev_bot_data = json.loads(group.in_round(round_num - 1).bot_decisions) current_bot_decisions[bot_id] = prev_bot_data[str(bot_id)] group.bot_decisions = json.dumps(current_bot_decisions) def calculate_ghost_decision(player): decision = None if player.round_number == 1: decision = random.randint(0, 1) else: prev_player = player.in_round(player.round_number - 1) prev_neighbor_data = prev_player.get_neighbor_data() counts = Counter([n['decision'] for n in prev_neighbor_data if n['decision'] is not None]) if counts[0] != counts[1]: # Pick the majority (0 or 1) most_frequent_decision = counts.most_common(1)[0][0] decision = most_frequent_decision else: # Tie: Pick own choice from previous round decision = prev_player.decision return decision