Source code for networkx.utils.misc

"""
Miscellaneous Helpers for NetworkX.

These are not imported into the base networkx namespace but
can be accessed, for example, as

>>> import networkx
>>> networkx.utils.make_list_of_ints({1, 2, 3})
[1, 2, 3]
>>> networkx.utils.arbitrary_element({5, 1, 7})  # doctest: +SKIP
1
"""

import random
import sys
import uuid
import warnings
from collections import defaultdict, deque
from collections.abc import Iterable, Iterator, Sized
from itertools import chain, tee

import networkx as nx

__all__ = [
    "flatten",
    "make_list_of_ints",
    "dict_to_numpy_array",
    "arbitrary_element",
    "pairwise",
    "groups",
    "create_random_state",
    "create_py_random_state",
    "PythonRandomInterface",
    "PythonRandomViaNumpyBits",
    "nodes_equal",
    "edges_equal",
    "graphs_equal",
    "_clear_cache",
]


# some cookbook stuff
# used in deciding whether something is a bunch of nodes, edges, etc.
# see G.add_nodes and others in Graph Class in networkx/base.py


[docs] def flatten(obj, result=None): """Return flattened version of (possibly nested) iterable object.""" if not isinstance(obj, Iterable | Sized) or isinstance(obj, str): return obj if result is None: result = [] for item in obj: if not isinstance(item, Iterable | Sized) or isinstance(item, str): result.append(item) else: flatten(item, result) return tuple(result)
[docs] def make_list_of_ints(sequence): """Return list of ints from sequence of integral numbers. All elements of the sequence must satisfy int(element) == element or a ValueError is raised. Sequence is iterated through once. If sequence is a list, the non-int values are replaced with ints. So, no new list is created """ if not isinstance(sequence, list): result = [] for i in sequence: errmsg = f"sequence is not all integers: {i}" try: ii = int(i) except ValueError: raise nx.NetworkXError(errmsg) from None if ii != i: raise nx.NetworkXError(errmsg) result.append(ii) return result # original sequence is a list... in-place conversion to ints for indx, i in enumerate(sequence): errmsg = f"sequence is not all integers: {i}" if isinstance(i, int): continue try: ii = int(i) except ValueError: raise nx.NetworkXError(errmsg) from None if ii != i: raise nx.NetworkXError(errmsg) sequence[indx] = ii return sequence
[docs] def dict_to_numpy_array(d, mapping=None): """Convert a dictionary of dictionaries to a numpy array with optional mapping.""" try: return _dict_to_numpy_array2(d, mapping) except (AttributeError, TypeError): # AttributeError is when no mapping was provided and v.keys() fails. # TypeError is when a mapping was provided and d[k1][k2] fails. return _dict_to_numpy_array1(d, mapping)
def _dict_to_numpy_array2(d, mapping=None): """Convert a dictionary of dictionaries to a 2d numpy array with optional mapping. """ import numpy as np if mapping is None: s = set(d.keys()) for k, v in d.items(): s.update(v.keys()) mapping = dict(zip(s, range(len(s)))) n = len(mapping) a = np.zeros((n, n)) for k1, i in mapping.items(): for k2, j in mapping.items(): try: a[i, j] = d[k1][k2] except KeyError: pass return a def _dict_to_numpy_array1(d, mapping=None): """Convert a dictionary of numbers to a 1d numpy array with optional mapping.""" import numpy as np if mapping is None: s = set(d.keys()) mapping = dict(zip(s, range(len(s)))) n = len(mapping) a = np.zeros(n) for k1, i in mapping.items(): i = mapping[k1] a[i] = d[k1] return a
[docs] def arbitrary_element(iterable): """Returns an arbitrary element of `iterable` without removing it. This is most useful for "peeking" at an arbitrary element of a set, but can be used for any list, dictionary, etc., as well. Parameters ---------- iterable : `abc.collections.Iterable` instance Any object that implements ``__iter__``, e.g. set, dict, list, tuple, etc. Returns ------- The object that results from ``next(iter(iterable))`` Raises ------ ValueError If `iterable` is an iterator (because the current implementation of this function would consume an element from the iterator). Examples -------- Arbitrary elements from common Iterable objects: >>> nx.utils.arbitrary_element([1, 2, 3]) # list 1 >>> nx.utils.arbitrary_element((1, 2, 3)) # tuple 1 >>> nx.utils.arbitrary_element({1, 2, 3}) # set 1 >>> d = {k: v for k, v in zip([1, 2, 3], [3, 2, 1])} >>> nx.utils.arbitrary_element(d) # dict_keys 1 >>> nx.utils.arbitrary_element(d.values()) # dict values 3 `str` is also an Iterable: >>> nx.utils.arbitrary_element("hello") 'h' :exc:`ValueError` is raised if `iterable` is an iterator: >>> iterator = iter([1, 2, 3]) # Iterator, *not* Iterable >>> nx.utils.arbitrary_element(iterator) Traceback (most recent call last): ... ValueError: cannot return an arbitrary item from an iterator Notes ----- This function does not return a *random* element. If `iterable` is ordered, sequential calls will return the same value:: >>> l = [1, 2, 3] >>> nx.utils.arbitrary_element(l) 1 >>> nx.utils.arbitrary_element(l) 1 """ if isinstance(iterable, Iterator): raise ValueError("cannot return an arbitrary item from an iterator") # Another possible implementation is ``for x in iterable: return x``. return next(iter(iterable))
# Recipe from the itertools documentation.
[docs] def pairwise(iterable, cyclic=False): "s -> (s0, s1), (s1, s2), (s2, s3), ..." a, b = tee(iterable) first = next(b, None) if cyclic is True: return zip(a, chain(b, (first,))) return zip(a, b)
[docs] def groups(many_to_one): """Converts a many-to-one mapping into a one-to-many mapping. `many_to_one` must be a dictionary whose keys and values are all :term:`hashable`. The return value is a dictionary mapping values from `many_to_one` to sets of keys from `many_to_one` that have that value. Examples -------- >>> from networkx.utils import groups >>> many_to_one = {"a": 1, "b": 1, "c": 2, "d": 3, "e": 3} >>> groups(many_to_one) # doctest: +SKIP {1: {'a', 'b'}, 2: {'c'}, 3: {'e', 'd'}} """ one_to_many = defaultdict(set) for v, k in many_to_one.items(): one_to_many[k].add(v) return dict(one_to_many)
[docs] def create_random_state(random_state=None): """Returns a numpy.random.RandomState or numpy.random.Generator instance depending on input. Parameters ---------- random_state : int or NumPy RandomState or Generator instance, optional (default=None) If int, return a numpy.random.RandomState instance set with seed=int. if `numpy.random.RandomState` instance, return it. if `numpy.random.Generator` instance, return it. if None or numpy.random, return the global random number generator used by numpy.random. """ import numpy as np if random_state is None or random_state is np.random: return np.random.mtrand._rand if isinstance(random_state, np.random.RandomState): return random_state if isinstance(random_state, int): return np.random.RandomState(random_state) if isinstance(random_state, np.random.Generator): return random_state msg = ( f"{random_state} cannot be used to create a numpy.random.RandomState or\n" "numpy.random.Generator instance" ) raise ValueError(msg)
class PythonRandomViaNumpyBits(random.Random): """Provide the random.random algorithms using a numpy.random bit generator The intent is to allow people to contribute code that uses Python's random library, but still allow users to provide a single easily controlled random bit-stream for all work with NetworkX. This implementation is based on helpful comments and code from Robert Kern on NumPy's GitHub Issue #24458. This implementation supersedes that of `PythonRandomInterface` which rewrote methods to account for subtle differences in API between `random` and `numpy.random`. Instead this subclasses `random.Random` and overwrites the methods `random`, `getrandbits`, `getstate`, `setstate` and `seed`. It makes them use the rng values from an input numpy `RandomState` or `Generator`. Those few methods allow the rest of the `random.Random` methods to provide the API interface of `random.random` while using randomness generated by a numpy generator. """ def __init__(self, rng=None): try: import numpy as np except ImportError: msg = "numpy not found, only random.random available." warnings.warn(msg, ImportWarning) if rng is None: self._rng = np.random.mtrand._rand else: self._rng = rng # Not necessary, given our overriding of gauss() below, but it's # in the superclass and nominally public, so initialize it here. self.gauss_next = None def random(self): """Get the next random number in the range 0.0 <= X < 1.0.""" return self._rng.random() def getrandbits(self, k): """getrandbits(k) -> x. Generates an int with k random bits.""" if k < 0: raise ValueError("number of bits must be non-negative") numbytes = (k + 7) // 8 # bits / 8 and rounded up x = int.from_bytes(self._rng.bytes(numbytes), "big") return x >> (numbytes * 8 - k) # trim excess bits def getstate(self): return self._rng.__getstate__() def setstate(self, state): self._rng.__setstate__(state) def seed(self, *args, **kwds): "Do nothing override method." raise NotImplementedError("seed() not implemented in PythonRandomViaNumpyBits") ################################################################## class PythonRandomInterface: """PythonRandomInterface is included for backward compatibility New code should use PythonRandomViaNumpyBits instead. """ def __init__(self, rng=None): try: import numpy as np except ImportError: msg = "numpy not found, only random.random available." warnings.warn(msg, ImportWarning) if rng is None: self._rng = np.random.mtrand._rand else: self._rng = rng def random(self): return self._rng.random() def uniform(self, a, b): return a + (b - a) * self._rng.random() def randrange(self, a, b=None): import numpy as np if b is None: a, b = 0, a if b > 9223372036854775807: # from np.iinfo(np.int64).max tmp_rng = PythonRandomViaNumpyBits(self._rng) return tmp_rng.randrange(a, b) if isinstance(self._rng, np.random.Generator): return self._rng.integers(a, b) return self._rng.randint(a, b) # NOTE: the numpy implementations of `choice` don't support strings, so # this cannot be replaced with self._rng.choice def choice(self, seq): import numpy as np if isinstance(self._rng, np.random.Generator): idx = self._rng.integers(0, len(seq)) else: idx = self._rng.randint(0, len(seq)) return seq[idx] def gauss(self, mu, sigma): return self._rng.normal(mu, sigma) def shuffle(self, seq): return self._rng.shuffle(seq) # Some methods don't match API for numpy RandomState. # Commented out versions are not used by NetworkX def sample(self, seq, k): return self._rng.choice(list(seq), size=(k,), replace=False) def randint(self, a, b): import numpy as np if b > 9223372036854775807: # from np.iinfo(np.int64).max tmp_rng = PythonRandomViaNumpyBits(self._rng) return tmp_rng.randint(a, b) if isinstance(self._rng, np.random.Generator): return self._rng.integers(a, b + 1) return self._rng.randint(a, b + 1) # exponential as expovariate with 1/argument, def expovariate(self, scale): return self._rng.exponential(1 / scale) # pareto as paretovariate with 1/argument, def paretovariate(self, shape): return self._rng.pareto(shape) # weibull as weibullvariate multiplied by beta, # def weibullvariate(self, alpha, beta): # return self._rng.weibull(alpha) * beta # # def triangular(self, low, high, mode): # return self._rng.triangular(low, mode, high) # # def choices(self, seq, weights=None, cum_weights=None, k=1): # return self._rng.choice(seq
[docs] def create_py_random_state(random_state=None): """Returns a random.Random instance depending on input. Parameters ---------- random_state : int or random number generator or None (default=None) - If int, return a `random.Random` instance set with seed=int. - If `random.Random` instance, return it. - If None or the `np.random` package, return the global random number generator used by `np.random`. - If an `np.random.Generator` instance, or the `np.random` package, or the global numpy random number generator, then return it. wrapped in a `PythonRandomViaNumpyBits` class. - If a `PythonRandomViaNumpyBits` instance, return it. - If a `PythonRandomInterface` instance, return it. - If a `np.random.RandomState` instance and not the global numpy default, return it wrapped in `PythonRandomInterface` for backward bit-stream matching with legacy code. Notes ----- - A diagram intending to illustrate the relationships behind our support for numpy random numbers is called `NetworkX Numpy Random Numbers <https://excalidraw.com/#room=b5303f2b03d3af7ccc6a,e5ZDIWdWWCTTsg8OqoRvPA>`_. - More discussion about this support also appears in `gh-6869#comment <https://github.com/networkx/networkx/pull/6869#issuecomment-1944799534>`_. - Wrappers of numpy.random number generators allow them to mimic the Python random number generation algorithms. For example, Python can create arbitrarily large random ints, and the wrappers use Numpy bit-streams with CPython's random module to choose arbitrarily large random integers too. - We provide two wrapper classes: `PythonRandomViaNumpyBits` is usually what you want and is always used for `np.Generator` instances. But for users who need to recreate random numbers produced in NetworkX 3.2 or earlier, we maintain the `PythonRandomInterface` wrapper as well. We use it only used if passed a (non-default) `np.RandomState` instance pre-initialized from a seed. Otherwise the newer wrapper is used. """ if random_state is None or random_state is random: return random._inst if isinstance(random_state, random.Random): return random_state if isinstance(random_state, int): return random.Random(random_state) try: import numpy as np except ImportError: pass else: if isinstance(random_state, PythonRandomInterface | PythonRandomViaNumpyBits): return random_state if isinstance(random_state, np.random.Generator): return PythonRandomViaNumpyBits(random_state) if random_state is np.random: return PythonRandomViaNumpyBits(np.random.mtrand._rand) if isinstance(random_state, np.random.RandomState): if random_state is np.random.mtrand._rand: return PythonRandomViaNumpyBits(random_state) # Only need older interface if specially constructed RandomState used return PythonRandomInterface(random_state) msg = f"{random_state} cannot be used to generate a random.Random instance" raise ValueError(msg)
[docs] def nodes_equal(nodes1, nodes2): """Check if nodes are equal. Equality here means equal as Python objects. Node data must match if included. The order of nodes is not relevant. Parameters ---------- nodes1, nodes2 : iterables of nodes, or (node, datadict) tuples Returns ------- bool True if nodes are equal, False otherwise. """ nlist1 = list(nodes1) nlist2 = list(nodes2) try: d1 = dict(nlist1) d2 = dict(nlist2) except (ValueError, TypeError): d1 = dict.fromkeys(nlist1) d2 = dict.fromkeys(nlist2) return d1 == d2
[docs] def edges_equal(edges1, edges2): """Check if edges are equal. Equality here means equal as Python objects. Edge data must match if included. The order of the edges is not relevant. Parameters ---------- edges1, edges2 : iterables of with u, v nodes as edge tuples (u, v), or edge tuples with data dicts (u, v, d), or edge tuples with keys and data dicts (u, v, k, d) Returns ------- bool True if edges are equal, False otherwise. """ from collections import defaultdict d1 = defaultdict(dict) d2 = defaultdict(dict) c1 = 0 for c1, e in enumerate(edges1): u, v = e[0], e[1] data = [e[2:]] if v in d1[u]: data = d1[u][v] + data d1[u][v] = data d1[v][u] = data c2 = 0 for c2, e in enumerate(edges2): u, v = e[0], e[1] data = [e[2:]] if v in d2[u]: data = d2[u][v] + data d2[u][v] = data d2[v][u] = data if c1 != c2: return False # can check one direction because lengths are the same. for n, nbrdict in d1.items(): for nbr, datalist in nbrdict.items(): if n not in d2: return False if nbr not in d2[n]: return False d2datalist = d2[n][nbr] for data in datalist: if datalist.count(data) != d2datalist.count(data): return False return True
[docs] def graphs_equal(graph1, graph2): """Check if graphs are equal. Equality here means equal as Python objects (not isomorphism). Node, edge and graph data must match. Parameters ---------- graph1, graph2 : graph Returns ------- bool True if graphs are equal, False otherwise. """ return ( graph1.adj == graph2.adj and graph1.nodes == graph2.nodes and graph1.graph == graph2.graph )
def _clear_cache(G): """Clear the cache of a graph (currently stores converted graphs). Caching is controlled via ``nx.config.cache_converted_graphs`` configuration. """ if cache := getattr(G, "__networkx_cache__", None): cache.clear() def check_create_using(create_using, *, directed=None, multigraph=None, default=None): """Assert that create_using has good properties This checks for desired directedness and multi-edge properties. It returns `create_using` unless that is `None` when it returns the optionally specified default value. Parameters ---------- create_using : None, graph class or instance The input value of create_using for a function. directed : None or bool Whether to check `create_using.is_directed() == directed`. If None, do not assert directedness. multigraph : None or bool Whether to check `create_using.is_multigraph() == multigraph`. If None, do not assert multi-edge property. default : None or graph class The graph class to return if create_using is None. Returns ------- create_using : graph class or instance The provided graph class or instance, or if None, the `default` value. Raises ------ NetworkXError When `create_using` doesn't match the properties specified by `directed` or `multigraph` parameters. """ if default is None: default = nx.Graph G = create_using if create_using is not None else default G_directed = G.is_directed(None) if isinstance(G, type) else G.is_directed() G_multigraph = G.is_multigraph(None) if isinstance(G, type) else G.is_multigraph() if directed is not None: if directed and not G_directed: raise nx.NetworkXError("create_using must be directed") if not directed and G_directed: raise nx.NetworkXError("create_using must not be directed") if multigraph is not None: if multigraph and not G_multigraph: raise nx.NetworkXError("create_using must be a multi-graph") if not multigraph and G_multigraph: raise nx.NetworkXError("create_using must not be a multi-graph") return G