netechos.core

Core functionality for the network echos model.

NetworkModel(number_of_nodes, interaction_type, alpha=1, beta=0.001)

Object containing network model parameters and methods.

Parameters:

Name	Type	Description	Default
number_of_nodes	int	Number of nodes to add to the network model.	required
interaction_type	Literal['symmetric', 'asymmetric']	Type of interaction between nodes. If "symmetric" interactions will be reciprocated. If "asymmetric", interactions may be one-directional.	required
alpha	float	Exponent of the adjacency matrix during attitude reinforcement.	1
beta	float	Factor governing the rate of attitude change. Smaller values slow attitude change.	0.001

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

def __init__(
    self: Self,
    number_of_nodes: int,
    interaction_type: Literal["symmetric", "asymmetric"],
    alpha: float = 1,
    beta: float = 1e-3,
) -> Self:
    """Object storing network model data and methods.

    Parameters
    ----------
    number_of_nodes : int
        Number of nodes to add to the network model.
    interaction_type : Literal["symmetric", "asymmetric"]
        Type of interaction between nodes. If "symmetric"
        interactions will be reciprocated. If "asymmetric",
        interactions may be one-directional.
    alpha : float
        Exponent of the adjacency matrix during attitude
        reinforcement.
    beta : float
        Factor governing the rate of attitude change. Smaller
        values slow attitude change.

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

    """
    allowed_types = ["symmetric", "asymmetric"]
    if interaction_type not in allowed_types:
        msg = f"interaction_type must be one of: {allowed_types}."
        raise ValueError(msg)
    if number_of_nodes < 1:
        msg = "number_of_nodes must be greater than one."
        raise ValueError(msg)
    self.number_of_nodes = number_of_nodes
    self.interaction_type = interaction_type
    self.adjacency_exponent = alpha
    self.attitude_change_speed = beta

compute_attitude_difference_matrix()

Construct matrix of differences between node attitudes.

Source code in src/netechos/core.py

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

create_network(network_type, p=0.1, k=2, m=1.0)

Create initial social network.

Parameters:

Name	Type	Description	Default
network_type	Literal['complete', 'erdos_renyi', 'watts_strogatz', 'newman_watts_strogatz', 'barabasi_albert']	Type of network to create.	required
p	float	Required if network_type is one of 'erdos_renyi', 'watts_strogatz', 'newman_watts_strogatz'. Probability for edge creation.	0.1
k	int	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.	2
m	float	Required if network_type is 'barabasi_albert'. Number of edges to attach from a new node to existing nodes.	1.0

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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

initialize_attitude_tracker()

Initialize attitude tracking table.

Source code in src/netechos/core.py

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

initialize_attitudes(distribution='vonmises', a=0.0, b=5)

Set initial node attitudes.

Parameters:

Name	Type	Description	Default
distribution	Literal['normal', 'uniform', 'laplace', 'vonmises']	Type of probability distribution to sample attitudes from.	'vonmises'
a	float	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.	0.0
b	float	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.	5

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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

initialize_connections(neighbor_weight=1.0, non_neighbor_weight=0.0)

Set initial network edge weights.

Parameters:

Name	Type	Description	Default
neighbor_weight	float	The weight to assign to edges between nodes which have an existing edge in the intial network.	1.0
non_neighbor_weight	float	The weight to assign to edges between nodes which don't have an existing edge in the intial network. If non-zero the network will technically become a complete network.	0.0

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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 intial network.
    non_neighbor_weight : float
        The weight to assign to edges between nodes which don't
        have an existing edge in the intial 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

initialize_summary_table()

Initialize summary table.

Source code in src/netechos/core.py

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

make_asymmetric_interactions()

Construct a random asymmetric adjacency matrix.

Source code in src/netechos/core.py

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

make_interactions()

Construct interaction matrix.

Source code in src/netechos/core.py

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

make_symmetric_interactions()

Construct a random symmetric adjacency matrix.

Source code in src/netechos/core.py

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

run_simulation(tmax)

Simulate interaction and attitude reinforcement over time.

Parameters:

Name	Type	Description	Default
tmax	int	Number of time steps to iterate over.	required

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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()
    for t in tqdm(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)

update_attitude_tracker(time)

Construct attitude tracking table for one time.

Parameters:

Name	Type	Description	Default
time	int	Time index. Will be assigned to all rows of the time column being added to the summary table.	required

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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

update_attitudes()

Calculate change in attitude.

Source code in src/netechos/core.py

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

update_connections()

Calculate change in connection strength.

Source code in src/netechos/core.py

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

update_summary_table(time)

Construct summary of the network at one time.

Parameters:

Name	Type	Description	Default
time	int	Time index. Will be assigned to all rows of the time column being added to the summary table.	required

Returns:

Name	Type	Description
self	Self @ NetworkModel	An instance of the NetworkModel object.

Source code in src/netechos/core.py

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