import random
import networkx as nx
import networkit as nk
from typing import Union
from littleballoffur.sampler import Sampler
NKGraph = type(nk.graph.Graph())
NXGraph = nx.classes.graph.Graph
[docs]class MetropolisHastingsRandomWalkSampler(Sampler):
r"""An implementation of node sampling by Metropolis Hastings random walks.
The random walker has a probabilistic acceptance condition for adding new nodes
to the sampled node set. This constraint can be parametrized by the rejection
constraint exponent. The sampled graph is always connected. `"For details about the algorithm see this paper." <http://mlcb.is.tuebingen.mpg.de/Veroeffentlichungen/papers/HueBorKriGha08.pdf>`_
Args:
number_of_nodes (int): Number of nodes. Default is 100.
seed (int): Random seed. Default is 42.
alpha (float): Rejection constraint exponent. Default is 1.0.
"""
def __init__(self, number_of_nodes: int = 100, seed: int = 42, alpha: float = 1.0):
self.number_of_nodes = number_of_nodes
self.seed = seed
self.alpha = alpha
self._set_seed()
def _create_initial_node_set(self, graph, start_node):
"""
Choosing an initial node.
"""
if start_node is not None:
if start_node >= 0 and start_node < self.backend.get_number_of_nodes(graph):
self._current_node = start_node
self._sampled_nodes = set([self._current_node])
else:
raise ValueError("Starting node index is out of range.")
else:
self._current_node = random.choice(
range(self.backend.get_number_of_nodes(graph))
)
self._sampled_nodes = set([self._current_node])
def _do_a_step(self, graph):
"""
Doing a single random walk step.
"""
score = random.uniform(0, 1)
new_node = self.backend.get_random_neighbor(graph, self._current_node)
ratio = float(self.backend.get_degree(graph, self._current_node)) / float(
self.backend.get_degree(graph, new_node)
)
ratio = ratio ** self.alpha
if score < ratio:
self._current_node = new_node
self._sampled_nodes.add(self._current_node)
[docs] def sample(
self, graph: Union[NXGraph, NKGraph], start_node: int = None
) -> Union[NXGraph, NKGraph]:
"""
Sampling nodes with a Metropolis Hastings single random walk.
Arg types:
* **graph** *(NetworkX or NetworKit graph)* - The graph to be sampled from.
* **start_node** *(int, optional)* - The start node.
Return types:
* **new_graph** *(NetworkX or NetworKit graph)* - The graph of sampled edges.
"""
self._deploy_backend(graph)
self._check_number_of_nodes(graph)
self._create_initial_node_set(graph, start_node)
while len(self._sampled_nodes) < self.number_of_nodes:
self._do_a_step(graph)
new_graph = self.backend.get_subgraph(graph, self._sampled_nodes)
return new_graph