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.

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

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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
    self.progress = Progress(
        SpinnerColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        BarColumn(),
        TimeRemainingColumn(),
        TimeElapsedColumn(),
    )

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.

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

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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.

PARAMETER	DESCRIPTION
distribution	Type of probability distribution to sample attitudes from. TYPE: Literal['normal', 'uniform', 'laplace', 'vonmises'] DEFAULT: 'vonmises'
a	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. TYPE: float DEFAULT: 0.0
b	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. TYPE: float DEFAULT: 5

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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.

PARAMETER	DESCRIPTION
neighbor_weight	The weight to assign to edges between nodes which have an existing edge in the initial network. TYPE: float DEFAULT: 1.0
non_neighbor_weight	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. TYPE: float DEFAULT: 0.0

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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 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

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.

PARAMETER	DESCRIPTION
tmax	Number of time steps to iterate over. TYPE: int

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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()
    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)

update_attitude_tracker(time)

Construct attitude tracking table for one time.

PARAMETER	DESCRIPTION
time	Time index. Will be assigned to all rows of the time column being added to the summary table. TYPE: int

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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.

PARAMETER	DESCRIPTION
time	Time index. Will be assigned to all rows of the time column being added to the summary table. TYPE: int

RETURNS	DESCRIPTION
self	An instance of the NetworkModel object. TYPE: Self @ NetworkModel

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