# Source code for nngt.generation.rewiring

```
# -*- coding: utf-8 -*-
# SPDX-FileCopyrightText: 2015-2023 Tanguy Fardet
# SPDX-License-Identifier: GPL-3.0-or-later
# nngt/generation/rewiring.py
""" Rewiring functions """
from copy import deepcopy
import numpy as np
import nngt
from nngt.generation import graph_connectivity as gc
from nngt.lib import nonstring_container
__all__ = [
"lattice_rewire",
"random_rewire"
]
[docs]def lattice_rewire(g, target_reciprocity=1., node_attr_constraints=None,
edge_attr_constraints=None, weight=None,
weight_constraint="distance", distance_sort="inverse"):
r'''
Build a (generally irregular) lattice by rewiring the edges of a graph.
.. versionadded:: 2.0
The lattice is based on a circular graph, meaning that the nodes are placed
on a circle and connected based on the topological distance between them,
the distance being defined through the positive modulo:
.. math::
d_{ij} = (i - j) \% N
with :math:`N` the number of nodes in the graph.
Parameters
----------
g : :class:`~nngt.Graph`
Graph based on which the lattice will be generated.
target_reciprocity : float, optional (default: 1.)
Value of reciprocity that should be aimed at. Depending on the number
of edges, it may not be possible to reach this value exactly.
node_attr_constraints : str, optional (default: randomize all attributes)
Whether attribute randomization is constrained: either "preserve",
where all nodes keep their attributes, or "together", where attributes
are randomized by groups (all attributes of a given node are sent to
the same new node). By default, attributes are completely and
separately randomized.
edge_attr_constraints : str, optional (default: randomize all but `weight`)
Whether attribute randomization is constrained.
If "distance" is used, then all number attributes (float or int) are
sorted and are first associated to the shortest or longest edges
depending on the value of `distance_sort`. Note that, for directed
graphs, if a reciprocal edge exists, it is immediately assigned the
next highest (respectively lowest) attribute after that of its directed
couterpart.
If "together" is used, edges attributes are randomized by groups (all
attributes of a given edge are sent to the same new edge) either
randomly if `weight` is None, or following the constrained `weight`
attribute. By default, attributes are completely and separately
randomized (except for `weight` if it has been provided).
weight : str, optional (default: None)
Whether a specific edge attribute should play the role of weight and
have special constraints.
weight_constraint : str, optional (default: "distance")
Same as `edge_attr_constraints`` but only applies to `weight` and can
only be "distance" or None since "together" was related to `weight`.
distance_sort : str, optional (default: "inverse")
How attributes are sorted with edge distance: either "inverse", with
the shortest edges being assigned the largest weights, or with a
"linear" sort, where shortest edges are assigned the lowest weights.
'''
directed = g.is_directed()
num_nodes = g.node_nb()
num_edges = g.edge_nb()
# check that requested lattice is possible
if directed:
if num_edges < int(num_nodes*(1 + target_reciprocity)):
raise ValueError("The number of edges in the graph is not "
"sufficient to make a lattice with requested "
"reciprocity.")
else:
if num_edges < num_nodes:
raise ValueError("The number of edges in the graph is not"
"sufficient to make a lattice.")
# check arguments
if node_attr_constraints not in (None, "preserve", "together"):
raise ValueError("`node_attr_constraints` must be either None, "
"'preserve', or 'together'.")
if edge_attr_constraints not in (None, "distance", "together"):
raise ValueError("`edge_attr_constraints` must be either None, "
"'distance', or 'together'.")
if weight_constraint not in ("distance", None):
raise ValueError("`weight_constraint` can only be 'distance' or None.")
if distance_sort not in ("linear", "inverse"):
raise ValueError("`distance_sort` must be either 'linear' or "
"'inverse'.")
if not directed and target_reciprocity != 1:
raise ValueError("Reciprocity is always 1 for undirected graphs.")
# init graph and edges
new_graph = nngt.Graph(nodes=num_nodes, directed=directed,
name=g.name + "_latticized")
ia_edges = np.full((num_edges, 2), -1, dtype=np.int64)
# compute the coodination number of the closest regular lattice
coord_nb = None
if directed:
# coordination number must be even
coord_nb = 2 * int(num_edges * (1 - 0.5 * target_reciprocity)
/ num_nodes)
else:
# coordination number must be even and resulting edges are half
coord_nb = 2*int(num_edges / num_nodes)
e_reglat = int(0.5*num_nodes*coord_nb)
# generate the edges of the regular lattice (setting 0 reciprocity for
# directed case, this is ignored if graph is undirected)
ids = range(num_nodes)
ia_edges[:e_reglat] = gc._circular(
ids, ids, coord_nb, directed=False,
reciprocity_choice="closest-ordered")
# add the remaining edges (remaining edges strictly smaller than num_nodes)
e_remaining = num_edges - e_reglat
if e_remaining:
last_edges = np.full((e_remaining, 2), -1, dtype=np.int64)
if directed:
# make reciprocal edges first
num_recip = int(0.5 * target_reciprocity * num_edges)
# check if recip are more numerous that regular lattice edges
first_recip = num_recip if num_recip <= e_reglat else e_reglat
if first_recip:
last_edges[:first_recip] = ia_edges[:first_recip, ::-1]
e_remaining -= first_recip
if e_remaining:
# new connections are one step above the max regular lattice
# distance
dist = int(0.5*coord_nb) + 1
# make reciprocal edges
num_recip -= first_recip
if num_recip:
last_edges[first_recip:first_recip + num_recip] = \
[(i, (i + dist) % num_nodes) for i in range(num_recip)]
start = first_recip + num_recip
stop = first_recip + 2*num_recip
last_edges[start:stop] = \
last_edges[first_recip:start, ::-1]
# make remaning non-reciprocal edges
e_final = e_remaining - 2*num_recip
if e_final:
last_edges[first_recip + 2*num_recip:] = \
[(i, (i + dist) % num_nodes)
for i in range(num_recip, num_recip + e_final)]
else:
# new connections are one step above the max regular lattice
# distance
dist = int(0.5*coord_nb) + 1
last_edges[:] = [(i, i + dist) for i in range(e_remaining)]
# put nodes back into [0, num_nodes[
last_edges[last_edges >= num_nodes] -= num_nodes
ia_edges[e_reglat:] = last_edges
# add the edges
new_graph.new_edges(ia_edges, check_duplicates=False,
check_self_loops=False, check_existing=False)
# set the node attributes
_set_node_attributes(g, new_graph, node_attr_constraints, num_nodes)
# edge attributes
order = None
# start with the weight
if weight is not None:
order = _lattice_shuffle_eattr(
weight, g, new_graph, coord_nb, target_reciprocity,
weight_constraint, distance_sort)
for eattr in g.edge_attributes:
if eattr != weight:
ordering = (order if edge_attr_constraints == "together"
else edge_attr_constraints)
order = _lattice_shuffle_eattr(
eattr, g, new_graph, coord_nb, target_reciprocity,
ordering, distance_sort)
return new_graph
[docs]def random_rewire(g, constraints=None, node_attr_constraints=None,
edge_attr_constraints=None, **kwargs):
'''
Generate a new rewired graph from `g`.
.. versionadded:: 2.0
Parameters
----------
g : :class:`~nngt.Graph`
Base graph based on which a new rewired graph will be generated.
constraints : str, optional (default: no constraints)
Defines which properties of `g` will be maintained in the rewired
graph. By default, the graph is completely rewired into an Erdos-Renyi
model. Available constraints are "in-degree", "out-degree",
"total-degree", "all-degrees", and "clustering".
node_attr_constraints : str, optional (default: randomize all attributes)
Whether attribute randomization is constrained: either "preserve",
where all nodes keep their attributes, or "together", where attributes
are randomized by groups (all attributes of a given node are sent to
the same new node). By default, attributes are completely and
separately randomized.
edge_attr_constraints : str, optional (default: randomize all attributes)
Whether attribute randomization is constrained.
If `constraints` is "in-degree" (respectively "out-degree") or
"degrees", this can be "preserve_in" (respectively "preserve_out"),
in which case all attributes of a given edge are moved together to a
new incoming (respectively outgoing) edge of the same node.
Regardless of `constraints`, "together" can be used so that edges
attributes are randomized by groups (all attributes of a given edge are
sent to the same new edge). By default, attributes are completely and
separately randomized.
**kwargs : optional keyword arguments
These are optional arguments in the case `constraints` is "clustering".
In that case, the user can provide both:
* `rtol` : float, optional (default: 5%)
The tolerance on the relative error to the average clustering for the
rewired graph.
* `connected` : bool, optional (default: False)
Whether the generated graph should be connected (this reduces the
precision of the final clustering).
* `method` : str, optional (default: "star-component")
Defines how the initially disconnected components of the generated
graph should be connected among themselves.
Available methods are "sequential", where the components are
connected sequentially, forming a long thread and increasing the
graph's diameter, "star-component", where all disconnected components
are connected to random nodes in the largest component,
"central-node" , where all disconnected components are connected to
the same node in the largest component, and "random", where
components are connected randomly.
'''
directed = g.is_directed()
num_nodes = g.node_nb()
num_edges = g.edge_nb()
new_graph = None
if node_attr_constraints not in (None, "preserve", "together"):
raise ValueError("`node_attr_constraints` must be either None, "
"'preserve', or 'together'.")
# check compatibility between `constraints` and `edge_attr_constraints`
valid_e = (None, "preserve_in", "preserve_out", "together")
if edge_attr_constraints not in valid_e:
raise ValueError(
"`edge_attr_constraints` must be in {}.".format(valid_e))
elif edge_attr_constraints == "preserve_in":
assert constraints in ("in-degree", "all-degrees"), \
"Can only use 'preserve_in' if `constraints` is 'in-degree' or " \
"'all-degrees'."
elif edge_attr_constraints == "preserve_out":
assert constraints in ("out-degree", "all-degrees"), \
"Can only use 'preserve_out' if `constraints` is 'out-degree' " \
"or 'all-degrees'."
# generate rewired graph
if constraints is None:
new_graph = gc.erdos_renyi(edges=num_edges, nodes=num_nodes,
directed=directed)
elif constraints == "all-degrees":
raise NotImplementedError("Full degrees constraints is not yet "
"implemented.")
elif "degree" in constraints:
degrees = g.get_degrees(constraints)
new_graph = gc.from_degree_list(degrees, constraints,
directed=directed)
elif constraints == "clustering":
rtol = kwargs.get("rtol", 0.05)
connected = kwargs.get("connected", g.is_connected())
method = kwargs.get("method", "star-component")
c = nngt.analysis.local_clustering(g).mean()
new_graph = gc.sparse_clustered(
c, edges=g.edge_nb(), nodes=num_nodes, connected=connected,
method=method, exact_edge_nb=True, rtol=rtol)
rng = nngt._rng
# node attributes
_set_node_attributes(g, new_graph, node_attr_constraints, num_nodes)
# edge attributes
order = np.arange(num_edges, dtype=int)
if edge_attr_constraints == "together":
rng.shuffle(order)
elif edge_attr_constraints == "preserve_in":
for i in range(num_nodes):
old_edges = g.get_edges(target_node=i)
new_edges = new_graph.get_edges(target_node=i)
if len(new_edges):
old_ids = g.edge_id(old_edges)
new_ids = new_graph.edge_id(new_edges)
order[new_ids] = old_ids
elif edge_attr_constraints == "preserve_out":
for i in range(num_nodes):
old_edges = g.get_edges(source_node=i)
new_edges = new_graph.get_edges(source_node=i)
if len(new_edges):
old_ids = g.edge_id(old_edges)
new_ids = new_graph.edge_id(new_edges)
order[new_ids] = old_ids
for k in g.edge_attributes:
v = deepcopy(g.get_edge_attributes(name=k))
if edge_attr_constraints is None:
rng.shuffle(v)
else:
v = v[order]
dtype = g.get_attribute_type(k, attribute_class="edge")
new_graph.new_edge_attribute(k, dtype, values=v)
# set spatial/network properties
if g.is_spatial():
nngt.Graph.make_spatial(new_graph, shape=g.shape.copy(),
positions=g.get_positions().copy())
if g.is_network():
nngt.Graph.make_network(new_graph, neural_pop=g.population.copy())
new_graph._name = g.name + "_rewired"
return new_graph
# ----- #
# Tools #
# ----- #
def _set_node_attributes(old_graph, new_graph, constraints, num_nodes):
''' Reassign node attributes '''
order = None
if constraints == "together":
order = [i for i in range(num_nodes)]
nngt._rng.shuffle(order) # shuffled order for "together"
for k, v in old_graph.node_attributes.items():
values = v.copy()
if constraints is None:
nngt._rng.shuffle(values)
elif constraints == "together":
values = v[order]
dtype = old_graph.get_attribute_type(k, attribute_class="node")
new_graph.new_node_attribute(k, dtype, values=values)
def _lattice_shuffle_eattr(name, old_graph, new_graph, coord_nb,
target_recip, order, distance_sort):
'''
Reassign edge attributes based on a constraint or a pre-defined
order for the lattice rewiring.
Parameters
----------
name : str
Name of the edge attribute.
old_graph : :class:`~nngt.Graph`
The old graph.
new_graph : :class:`~nngt.Graph`
The new graph.
coord_nb : int
Coordination number of the lattice.
target_recip : float
Target reciprocity of the lattice.
order : array of indices, distance", or None
Constraint on edge reassignment: either a precomputed order, "distance"
if we perform a distance-based shuffle, or None if we randomly
shuffle the attributes.
distance_sort : str
How the attributes should correlate with the distance (linearly or
inversely) if `order` is "distance".
Returns
-------
order : array of indices
The order in which the edge attributes have been shuffled.
'''
num_nodes = new_graph.node_nb()
num_edges = new_graph.edge_nb()
# old attribute
value_type = old_graph.get_attribute_type(name, "edge")
values = old_graph.edge_attributes[name].copy()
# compute order and reassign values
if order is None:
order = np.arange(num_edges, dtype=int)
nngt._rng.shuffle(order)
values = values[order]
elif nonstring_container(order):
# use precomputed order
values = values[order]
else:
# distance sort
directed = new_graph.is_directed()
if directed:
# we need to find the reciprocal edges for the attribute
# assignment (this relies on the precise implementation of the
# function _circular_directed_recip that the closest distances
# come first, then the reciprocal edges are at the end in the
# same order)
init_edges = int(0.5*num_nodes*coord_nb)
num_recip = int(0.5 * target_recip * num_edges)
first_recip = \
num_recip if num_recip <= init_edges else init_edges
second_recip = num_recip - first_recip
# fill the order list in the following order
order = np.zeros(num_edges, dtype=int)
# the first entries are the initial edges that got a reciprocal
# connection, we order them with every other first indices
# since the reciprocal edges will come in between
if first_recip:
order[:2*first_recip - 1:2] = np.arange(first_recip)
# we enter the index of the reciprocal connection
order[1:2*first_recip:2] = \
np.arange(init_edges, init_edges + first_recip)
# then (if needed) we fill the 2nd wave of reciprocal edges
if second_recip:
start = init_edges + first_recip
order[2*first_recip:2*num_recip - 1:2] = \
np.arange(start, start + second_recip)
order[2*first_recip + 1:2*num_recip:2] = \
np.arange(start + second_recip, start + 2*second_recip)
# then we fill the last entries with the initial edges that did
# not get a reciprocal connection
end = 2*num_recip + init_edges - first_recip
order[2*num_recip:end] = np.arange(num_recip, init_edges)
order[end:] = \
np.arange(init_edges + num_recip + second_recip, num_edges)
order = np.argsort(order)
else:
# we don't need to sort the new edges because they are ordered by
# distance by default in the circular algorithm
order = slice(num_edges)
# sort the attribute
if distance_sort == "linear":
# order for other attributes if "together" is used
order = np.argsort(values)[order]
# sorted values
values = values[order]
else:
# order for other attributes if "together" is used
order = np.argsort(values)[::-1][order]
# sorted values
values = values[order]
# set the new attributes
new_graph.new_edge_attribute(name, value_type, values=values)
return order
```