netechos.core¶

def compute_attitude_difference_matrix(self: Self) -> Self:
    """Construct matrix of differences between node attitudes."""
    self.attitude_diffs = np.subtract(
        self.attitudes,
        self.attitudes.transpose(),
    )
    return self

def create_network(
    self: Self,
    network_type: Literal[
        "complete",
        "erdos_renyi",
        "watts_strogatz",
        "newman_watts_strogatz",
        "barabasi_albert",
    ],
    p: float = 0.1,
    k: int = 2,
    m: float = 1.0,
) -> Self:
    """Create initial social network.

    Parameters
    ----------
    network_type: Literal["complete", "erdos_renyi", "watts_strogatz", "newman_watts_strogatz", "barabasi_albert"]
        Type of network to create.
    p : float, optional, default: 0.1
        Required if network_type is one of 'erdos_renyi',
        'watts_strogatz', 'newman_watts_strogatz'. Probability for
        edge creation.
    k : int, optional, default: 2
        Required if network_type is one of 'watts_strogatz',
        'newman_watts_strogatz'. Each node is joined with its `k`
        nearest neighbors in a ring topology.
    m : float, optional, default: 1.0
        Required if network_type is 'barabasi_albert'. Number
        of edges to attach from a new node to existing nodes.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """  # noqa: E501
    if network_type == "complete":
        g_initial = nx.complete_graph(n=self.number_of_nodes)
    if network_type == "erdos_renyi":
        g_initial = nx.erdos_renyi_graph(n=self.number_of_nodes, p=p)
        self.p_edge = p
    if network_type == "watts_strogatz":
        g_initial = nx.watts_strogatz_graph(
            n=self.number_of_nodes,
            k=k,
            p=p,
        )
        self.k_neighbors = k
        self.p_edge = p
    if network_type == "newman_watts_strogatz":
        g_initial = nx.newman_watts_strogatz_graph(
            n=self.number_of_nodes,
            k=k,
            p=p,
        )
        self.k_neighbors = k
        self.p_edge = p
    if network_type == "barabasi_albert":
        g_initial = nx.barabasi_albert_graph(n=self.number_of_nodes, m=m)
        self.m_new_node_edges = m
    self.social_network = g_initial
    self.network_type = network_type
    return self

def initialize_attitude_tracker(self: Self) -> Self:
    """Initialize attitude tracking table."""
    self.attitude_tracker = pl.DataFrame(
        schema={"time": int, "node": int, "attitude": float},
    )
    return self

def initialize_attitudes(
    self: Self,
    distribution: Literal[
        "normal",
        "uniform",
        "laplace",
        "vonmises",
    ] = "vonmises",
    a: float = 0.0,
    b: float = 5,
) -> Self:
    """Set initial node attitudes.

    Parameters
    ----------
    distribution : Literal['normal', 'uniform', 'laplace', 'vonmises'], default: 'vonmises'
        Type of probability distribution to sample attitudes from.
    a : float, default: 0.0
        First parameter for `distribution`. If `distribution` is
        'normal' or 'laplace' this is the `loc` parameter. If
        `distribution` is 'uniform', this is the lower bound. If
        `distribution` is 'vonmises' this is the `mu` parameter.
    b : float, default: 5
        Second parameter for `distribution`. If `distribution` is
        'normal' or 'laplace' this is the `scale` parameter. If
        `distribution` is 'uniform', this is the upper bound. If
        `distribution` is 'vonmises' this is the `kappa` parameter.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """  # noqa: E501
    rng = np.random.default_rng()
    if distribution == "normal":
        attitudes = rng.normal(
            loc=a,
            scale=b,
            size=(1, self.number_of_nodes),
        )
        self.attitude_distribution_loc = a
        self.attitude_distribution_scale = b
    if distribution == "laplace":
        attitudes = rng.laplace(
            loc=a,
            scale=b,
            size=(1, self.number_of_nodes),
        )
        self.attitude_distribution_loc = a
        self.attitude_distribution_scale = b
    if distribution == "vonmises":
        attitudes = rng.vonmises(
            mu=a,
            kappa=b,
            size=(1, self.number_of_nodes),
        )
        self.attitude_distribution_mu = a
        self.attitude_distribution_kappa = b
    if distribution == "uniform":
        attitudes = rng.uniform(
            low=a,
            high=b,
            size=(1, self.number_of_nodes),
        )
        self.attitude_distribution_low = a
        self.attitude_distribution_high = b
    attitudes = np.clip(attitudes, a_min=-np.pi / 2, a_max=np.pi / 2)
    self.attitudes = attitudes
    self.attitude_distribution = distribution
    return self

def initialize_connections(
    self: Self,
    neighbor_weight: float = 1.0,
    non_neighbor_weight: float = 0.0,
) -> Self:
    """Set initial network edge weights.

    Parameters
    ----------
    neighbor_weight : float
        The weight to assign to edges between nodes which have an
        existing edge in the initial network.
    non_neighbor_weight : float
        The weight to assign to edges between nodes which don't
        have an existing edge in the initial network. If non-zero
        the network will technically become a complete network.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """
    adjacency_mat = nx.to_numpy_array(self.social_network)
    adjacency_mat = neighbor_weight * adjacency_mat
    adjacency_mat[adjacency_mat == 0] = non_neighbor_weight
    self.connections = adjacency_mat
    self.neighbor_weight = neighbor_weight
    self.non_neighbor_weight = non_neighbor_weight
    return self

def initialize_summary_table(self: Self) -> Self:
    """Initialize summary table."""
    self.summary_table = pl.DataFrame(
        schema={
            "time": int,
            "attitude_mean": float,
            "attitude_sd": float,
            "connection_mean": float,
            "connection_sd": float,
        },
    )
    return self

def make_asymmetric_interactions(self: Self) -> Self:
    """Construct a random asymmetric adjacency matrix."""
    rng = np.random.default_rng()
    rand_mat = rng.random(size=(self.number_of_nodes, self.number_of_nodes))
    interactions = np.where(rand_mat <= self.connections, 1, 0)
    self.interactions = interactions
    return self

def make_interactions(self: Self) -> Self:
    """Construct interaction matrix."""
    if self.interaction_type == "symmetric":
        self.make_symmetric_interactions()
    elif self.interaction_type == "asymmetric":
        self.make_asymmetric_interactions()
    else:
        msg = f"""self.interaction_type must be one of ["symmetric",
        "asymmetric"], got: {self.interaction_type}"""
        raise ValueError(msg)
    return self

def make_symmetric_interactions(self: Self) -> Self:
    """Construct a random symmetric adjacency matrix."""
    rng = np.random.default_rng()
    rand_mat = rng.random(size=(self.number_of_nodes, self.number_of_nodes))
    interactions = np.where(rand_mat <= self.connections, 1, 0)
    interactions = np.maximum(interactions, interactions.transpose())
    self.interactions = interactions
    return self

def run_simulation(self: Self, tmax: int) -> Self:
    """Simulate interaction and attitude reinforcement over time.

    Parameters
    ----------
    tmax : int
        Number of time steps to iterate over.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """
    if tmax < 1:
        msg = "tmax must be greater than 1"
        raise ValueError(msg)
    self.initialize_summary_table()
    self.initialize_attitude_tracker()
    with self.progress as p:
        for t in p.track(range(tmax)):
            self.compute_attitude_difference_matrix()
            self.make_interactions()
            self.update_connections()
            self.update_attitudes()
            self.update_summary_table(time=t)
            self.update_attitude_tracker(time=t)

def update_attitude_tracker(self: Self, time: int) -> Self:
    """Construct attitude tracking table for one time.

    Parameters
    ----------
    time : int
        Time index. Will be assigned to all rows of the `time`
        column being added to the summary table.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """
    attitudes_flat = self.attitudes.flatten()
    step_attitude_tracker = pl.DataFrame(
        {
            "time": np.full(
                shape=attitudes_flat.shape,
                fill_value=time,
                dtype="int64",
            ),
            "node": np.arange(self.number_of_nodes, dtype="int64").reshape(
                attitudes_flat.shape,
            ),
            "attitude": attitudes_flat,
        },
    )
    self.attitude_tracker = pl.concat(
        [self.attitude_tracker, step_attitude_tracker],
    )
    return self

def update_attitudes(self: Self) -> Self:
    """Calculate change in attitude."""
    connections_to_power = np.where(
        self.connections != 0,
        np.power(self.connections, self.adjacency_exponent),
        0,
    )
    d_attitudes = np.multiply(
        np.multiply(connections_to_power, self.interactions),
        np.sin(self.attitude_diffs),
    ).sum(axis=0)
    self.attitudes = self.attitudes + (self.attitude_change_speed * d_attitudes)
    self.attitudes = np.clip(
        self.attitudes,
        a_min=-np.pi / 2,
        a_max=np.pi / 2,
    )
    return self

def update_connections(self: Self) -> Self:
    """Calculate change in connection strength."""
    d_connections = np.multiply(
        self.interactions,
        np.sin(self.attitude_diffs),
    )
    self.connections = self.connections + d_connections
    self.connections = np.clip(self.connections, a_min=0, a_max=1)
    return self

def update_summary_table(self: Self, time: int) -> Self:
    """Construct summary of the network at one time.

    Parameters
    ----------
    time : int
        Time index. Will be assigned to all rows of the `time`
        column being added to the summary table.

    Returns
    -------
    self : Self@NetworkModel
        An instance of the NetworkModel object.

    """
    step_summary_table = pl.DataFrame(
        {
            "time": time,
            "attitude_mean": np.mean(np.sin(self.attitudes)),
            "attitude_sd": np.std(np.sin(self.attitudes)),
            "connection_mean": np.mean(self.connections),
            "connection_sd": np.std(self.connections),
        },
    )
    self.summary_table = pl.concat([self.summary_table, step_summary_table])
    return self