Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/networkx/utils/misc.py: 18%

1"""

2Miscellaneous Helpers for NetworkX.

4These are not imported into the base networkx namespace but

5can be accessed, for example, as

7>>> import networkx as nx

8>>> nx.utils.make_list_of_ints({1, 2, 3})

9[1, 2, 3]

10>>> nx.utils.arbitrary_element({5, 1, 7}) # doctest: +SKIP

111

12"""

14import itertools

15import random

16import warnings

17from collections import defaultdict

18from collections.abc import Iterable, Iterator, Sized

19from itertools import chain, tee, zip_longest

21import networkx as nx

23__all__ = [

24 "flatten",

25 "make_list_of_ints",

26 "dict_to_numpy_array",

27 "arbitrary_element",

28 "pairwise",

29 "groups",

30 "create_random_state",

31 "create_py_random_state",

32 "PythonRandomInterface",

33 "PythonRandomViaNumpyBits",

34 "nodes_equal",

35 "edges_equal",

36 "graphs_equal",

37 "_clear_cache",

38]

41# some cookbook stuff

42# used in deciding whether something is a bunch of nodes, edges, etc.

43# see G.add_nodes and others in Graph Class in networkx/base.py

46def flatten(obj, result=None):

47 """Return flattened version of (possibly nested) iterable object."""

48 if not isinstance(obj, Iterable | Sized) or isinstance(obj, str):

49 return obj

50 if result is None:

51 result = []

52 for item in obj:

53 if not isinstance(item, Iterable | Sized) or isinstance(item, str):

54 result.append(item)

55 else:

56 flatten(item, result)

57 return tuple(result)

60def make_list_of_ints(sequence):

61 """Return list of ints from sequence of integral numbers.

63 All elements of the sequence must satisfy int(element) == element

64 or a ValueError is raised. Sequence is iterated through once.

66 If sequence is a list, the non-int values are replaced with ints.

67 So, no new list is created

68 """

69 if not isinstance(sequence, list):

70 result = []

71 for i in sequence:

72 errmsg = f"sequence is not all integers: {i}"

73 try:

74 ii = int(i)

75 except ValueError:

76 raise nx.NetworkXError(errmsg) from None

77 if ii != i:

78 raise nx.NetworkXError(errmsg)

79 result.append(ii)

80 return result

81 # original sequence is a list... in-place conversion to ints

82 for indx, i in enumerate(sequence):

83 errmsg = f"sequence is not all integers: {i}"

84 if isinstance(i, int):

85 continue

86 try:

87 ii = int(i)

88 except ValueError:

89 raise nx.NetworkXError(errmsg) from None

90 if ii != i:

91 raise nx.NetworkXError(errmsg)

92 sequence[indx] = ii

93 return sequence

96def dict_to_numpy_array(d, mapping=None):

97 """Convert a dictionary of dictionaries to a numpy array

98 with optional mapping."""

99 try:

100 return _dict_to_numpy_array2(d, mapping)

101 except (AttributeError, TypeError):

102 # AttributeError is when no mapping was provided and v.keys() fails.

103 # TypeError is when a mapping was provided and d[k1][k2] fails.

104 return _dict_to_numpy_array1(d, mapping)

105

106

107def _dict_to_numpy_array2(d, mapping=None):

108 """Convert a dictionary of dictionaries to a 2d numpy array

109 with optional mapping.

110

111 """

112 import numpy as np

113

114 if mapping is None:

115 s = set(d.keys())

116 for k, v in d.items():

117 s.update(v.keys())

118 mapping = dict(zip(s, range(len(s))))

119 n = len(mapping)

120 a = np.zeros((n, n))

121 for k1, i in mapping.items():

122 for k2, j in mapping.items():

123 try:

124 a[i, j] = d[k1][k2]

125 except KeyError:

126 pass

127 return a

128

129

130def _dict_to_numpy_array1(d, mapping=None):

131 """Convert a dictionary of numbers to a 1d numpy array with optional mapping."""

132 import numpy as np

133

134 if mapping is None:

135 s = set(d.keys())

136 mapping = dict(zip(s, range(len(s))))

137 n = len(mapping)

138 a = np.zeros(n)

139 for k1, i in mapping.items():

140 i = mapping[k1]

141 a[i] = d[k1]

142 return a

143

144

145def arbitrary_element(iterable):

146 """Returns an arbitrary element of `iterable` without removing it.

147

148 This is most useful for "peeking" at an arbitrary element of a set,

149 but can be used for any list, dictionary, etc., as well.

150

151 Parameters

152 ----------

153 iterable : `abc.collections.Iterable` instance

154 Any object that implements ``__iter__``, e.g. set, dict, list, tuple,

155 etc.

156

157 Returns

158 -------

159 The object that results from ``next(iter(iterable))``

160

161 Raises

162 ------

163 ValueError

164 If `iterable` is an iterator (because the current implementation of

165 this function would consume an element from the iterator).

166

167 Examples

168 --------

169 Arbitrary elements from common Iterable objects:

170

171 >>> nx.utils.arbitrary_element([1, 2, 3]) # list

172 1

173 >>> nx.utils.arbitrary_element((1, 2, 3)) # tuple

174 1

175 >>> nx.utils.arbitrary_element({1, 2, 3}) # set

176 1

177 >>> d = {k: v for k, v in zip([1, 2, 3], [3, 2, 1])}

178 >>> nx.utils.arbitrary_element(d) # dict_keys

179 1

180 >>> nx.utils.arbitrary_element(d.values()) # dict values

181 3

182

183 `str` is also an Iterable:

184

185 >>> nx.utils.arbitrary_element("hello")

186 'h'

187

188 :exc:`ValueError` is raised if `iterable` is an iterator:

189

190 >>> iterator = iter([1, 2, 3]) # Iterator, *not* Iterable

191 >>> nx.utils.arbitrary_element(iterator)

192 Traceback (most recent call last):

193 ...

194 ValueError: cannot return an arbitrary item from an iterator

195

196 Notes

197 -----

198 This function does not return a *random* element. If `iterable` is

199 ordered, sequential calls will return the same value::

200

201 >>> l = [1, 2, 3]

202 >>> nx.utils.arbitrary_element(l)

203 1

204 >>> nx.utils.arbitrary_element(l)

205 1

206

207 """

208 if isinstance(iterable, Iterator):

209 raise ValueError("cannot return an arbitrary item from an iterator")

210 # Another possible implementation is ``for x in iterable: return x``.

211 return next(iter(iterable))

212

213

214def pairwise(iterable, cyclic=False):

215 """Return successive overlapping pairs taken from an input iterable.

216

217 Parameters

218 ----------

219 iterable : iterable

220 An iterable from which to generate pairs.

221

222 cyclic : bool, optional (default=False)

223 If `True`, a pair with the last and first items is included at the end.

224

225 Returns

226 -------

227 iterator

228 An iterator over successive overlapping pairs from the `iterable`.

229

230 See Also

231 --------

232 itertools.pairwise

233

234 Examples

235 --------

236 >>> list(nx.utils.pairwise([1, 2, 3, 4]))

237 [(1, 2), (2, 3), (3, 4)]

238

239 >>> list(nx.utils.pairwise([1, 2, 3, 4], cyclic=True))

240 [(1, 2), (2, 3), (3, 4), (4, 1)]

241 """

242 if not cyclic:

243 return itertools.pairwise(iterable)

244 a, b = tee(iterable)

245 first = next(b, None)

246 return zip(a, chain(b, (first,)))

247

248

249def groups(many_to_one):

250 """Converts a many-to-one mapping into a one-to-many mapping.

251

252 `many_to_one` must be a dictionary whose keys and values are all

253 :term:`hashable`.

254

255 The return value is a dictionary mapping values from `many_to_one`

256 to sets of keys from `many_to_one` that have that value.

257

258 Examples

259 --------

260 >>> from networkx.utils import groups

261 >>> many_to_one = {"a": 1, "b": 1, "c": 2, "d": 3, "e": 3}

262 >>> groups(many_to_one) # doctest: +SKIP

263 {1: {'a', 'b'}, 2: {'c'}, 3: {'e', 'd'}}

264 """

265 one_to_many = defaultdict(set)

266 for v, k in many_to_one.items():

267 one_to_many[k].add(v)

268 return dict(one_to_many)

269

270

271def create_random_state(random_state=None):

272 """Returns a numpy.random.RandomState or numpy.random.Generator instance

273 depending on input.

274

275 Parameters

276 ----------

277 random_state : int or NumPy RandomState or Generator instance, optional (default=None)

278 If int, return a numpy.random.RandomState instance set with seed=int.

279 if `numpy.random.RandomState` instance, return it.

280 if `numpy.random.Generator` instance, return it.

281 if None or numpy.random, return the global random number generator used

282 by numpy.random.

283 """

284 import numpy as np

285

286 if random_state is None or random_state is np.random:

287 return np.random.mtrand._rand

288 if isinstance(random_state, np.random.RandomState):

289 return random_state

290 if isinstance(random_state, int):

291 return np.random.RandomState(random_state)

292 if isinstance(random_state, np.random.Generator):

293 return random_state

294 msg = (

295 f"{random_state} cannot be used to create a numpy.random.RandomState or\n"

296 "numpy.random.Generator instance"

297 )

298 raise ValueError(msg)

299

300

301class PythonRandomViaNumpyBits(random.Random):

302 """Provide the random.random algorithms using a numpy.random bit generator

303

304 The intent is to allow people to contribute code that uses Python's random

305 library, but still allow users to provide a single easily controlled random

306 bit-stream for all work with NetworkX. This implementation is based on helpful

307 comments and code from Robert Kern on NumPy's GitHub Issue #24458.

308

309 This implementation supersedes that of `PythonRandomInterface` which rewrote

310 methods to account for subtle differences in API between `random` and

311 `numpy.random`. Instead this subclasses `random.Random` and overwrites

312 the methods `random`, `getrandbits`, `getstate`, `setstate` and `seed`.

313 It makes them use the rng values from an input numpy `RandomState` or `Generator`.

314 Those few methods allow the rest of the `random.Random` methods to provide

315 the API interface of `random.random` while using randomness generated by

316 a numpy generator.

317 """

318

319 def __init__(self, rng=None):

320 try:

321 import numpy as np

322 except ImportError:

323 msg = "numpy not found, only random.random available."

324 warnings.warn(msg, ImportWarning)

325

326 if rng is None:

327 self._rng = np.random.mtrand._rand

328 else:

329 self._rng = rng

330

331 # Not necessary, given our overriding of gauss() below, but it's

332 # in the superclass and nominally public, so initialize it here.

333 self.gauss_next = None

334

335 def random(self):

336 """Get the next random number in the range 0.0 <= X < 1.0."""

337 return self._rng.random()

338

339 def getrandbits(self, k):

340 """getrandbits(k) -> x. Generates an int with k random bits."""

341 if k < 0:

342 raise ValueError("number of bits must be non-negative")

343 numbytes = (k + 7) // 8 # bits / 8 and rounded up

344 x = int.from_bytes(self._rng.bytes(numbytes), "big")

345 return x >> (numbytes * 8 - k) # trim excess bits

346

347 def getstate(self):

348 return self._rng.__getstate__()

349

350 def setstate(self, state):

351 self._rng.__setstate__(state)

352

353 def seed(self, *args, **kwds):

354 "Do nothing override method."

355 raise NotImplementedError("seed() not implemented in PythonRandomViaNumpyBits")

356

357

358##################################################################

359class PythonRandomInterface:

360 """PythonRandomInterface is included for backward compatibility

361 New code should use PythonRandomViaNumpyBits instead.

362 """

363

364 def __init__(self, rng=None):

365 try:

366 import numpy as np

367 except ImportError:

368 msg = "numpy not found, only random.random available."

369 warnings.warn(msg, ImportWarning)

370

371 if rng is None:

372 self._rng = np.random.mtrand._rand

373 else:

374 self._rng = rng

375

376 def random(self):

377 return self._rng.random()

378

379 def uniform(self, a, b):

380 return a + (b - a) * self._rng.random()

381

382 def randrange(self, a, b=None):

383 import numpy as np

384

385 if b is None:

386 a, b = 0, a

387 if b > 9223372036854775807: # from np.iinfo(np.int64).max

388 tmp_rng = PythonRandomViaNumpyBits(self._rng)

389 return tmp_rng.randrange(a, b)

390

391 if isinstance(self._rng, np.random.Generator):

392 return self._rng.integers(a, b)

393 return self._rng.randint(a, b)

394

395 # NOTE: the numpy implementations of `choice` don't support strings, so

396 # this cannot be replaced with self._rng.choice

397 def choice(self, seq):

398 import numpy as np

399

400 if isinstance(self._rng, np.random.Generator):

401 idx = self._rng.integers(0, len(seq))

402 else:

403 idx = self._rng.randint(0, len(seq))

404 return seq[idx]

405

406 def gauss(self, mu, sigma):

407 return self._rng.normal(mu, sigma)

408

409 def shuffle(self, seq):

410 return self._rng.shuffle(seq)

411

412 # Some methods don't match API for numpy RandomState.

413 # Commented out versions are not used by NetworkX

414

415 def sample(self, seq, k):

416 return self._rng.choice(list(seq), size=(k,), replace=False)

417

418 def randint(self, a, b):

419 import numpy as np

420

421 if b > 9223372036854775807: # from np.iinfo(np.int64).max

422 tmp_rng = PythonRandomViaNumpyBits(self._rng)

423 return tmp_rng.randint(a, b)

424

425 if isinstance(self._rng, np.random.Generator):

426 return self._rng.integers(a, b + 1)

427 return self._rng.randint(a, b + 1)

428

429 # exponential as expovariate with 1/argument,

430 def expovariate(self, scale):

431 return self._rng.exponential(1 / scale)

432

433 # pareto as paretovariate with argument,

434 def paretovariate(self, shape):

435 return self._rng.pareto(shape)

436

437

438# weibull as weibullvariate multiplied by beta,

439# def weibullvariate(self, alpha, beta):

440# return self._rng.weibull(alpha) * beta

441#

442# def triangular(self, low, high, mode):

443# return self._rng.triangular(low, mode, high)

444#

445# def choices(self, seq, weights=None, cum_weights=None, k=1):

446# return self._rng.choice(seq

447

448

449def create_py_random_state(random_state=None):

450 """Returns a random.Random instance depending on input.

451

452 Parameters

453 ----------

454 random_state : int or random number generator or None (default=None)

455 - If int, return a `random.Random` instance set with seed=int.

456 - If `random.Random` instance, return it.

457 - If None or the `np.random` package, return the global random number

458 generator used by `np.random`.

459 - If an `np.random.Generator` instance, or the `np.random` package, or

460 the global numpy random number generator, then return it.

461 wrapped in a `PythonRandomViaNumpyBits` class.

462 - If a `PythonRandomViaNumpyBits` instance, return it.

463 - If a `PythonRandomInterface` instance, return it.

464 - If a `np.random.RandomState` instance and not the global numpy default,

465 return it wrapped in `PythonRandomInterface` for backward bit-stream

466 matching with legacy code.

467

468 Notes

469 -----

470 - A diagram intending to illustrate the relationships behind our support

471 for numpy random numbers is called

472 `NetworkX Numpy Random Numbers <https://excalidraw.com/#room=b5303f2b03d3af7ccc6a,e5ZDIWdWWCTTsg8OqoRvPA>`_.

473 - More discussion about this support also appears in

474 `gh-6869#comment <https://github.com/networkx/networkx/pull/6869#issuecomment-1944799534>`_.

475 - Wrappers of numpy.random number generators allow them to mimic the Python random

476 number generation algorithms. For example, Python can create arbitrarily large

477 random ints, and the wrappers use Numpy bit-streams with CPython's random module

478 to choose arbitrarily large random integers too.

479 - We provide two wrapper classes:

480 `PythonRandomViaNumpyBits` is usually what you want and is always used for

481 `np.Generator` instances. But for users who need to recreate random numbers

482 produced in NetworkX 3.2 or earlier, we maintain the `PythonRandomInterface`

483 wrapper as well. We use it only used if passed a (non-default) `np.RandomState`

484 instance pre-initialized from a seed. Otherwise the newer wrapper is used.

485 """

486 if random_state is None or random_state is random:

487 return random._inst

488 if isinstance(random_state, random.Random):

489 return random_state

490 if isinstance(random_state, int):

491 return random.Random(random_state)

492

493 try:

494 import numpy as np

495 except ImportError:

496 pass

497 else:

498 if isinstance(random_state, PythonRandomInterface | PythonRandomViaNumpyBits):

499 return random_state

500 if isinstance(random_state, np.random.Generator):

501 return PythonRandomViaNumpyBits(random_state)

502 if random_state is np.random:

503 return PythonRandomViaNumpyBits(np.random.mtrand._rand)

504

505 if isinstance(random_state, np.random.RandomState):

506 if random_state is np.random.mtrand._rand:

507 return PythonRandomViaNumpyBits(random_state)

508 # Only need older interface if specially constructed RandomState used

509 return PythonRandomInterface(random_state)

510

511 msg = f"{random_state} cannot be used to generate a random.Random instance"

512 raise ValueError(msg)

513

514

515def nodes_equal(nodes1, nodes2):

516 """Check if nodes are equal.

517

518 Equality here means equal as Python objects.

519 Node data must match if included.

520 The order of nodes is not relevant.

521

522 Parameters

523 ----------

524 nodes1, nodes2 : iterables of nodes, or (node, datadict) tuples

525

526 Returns

527 -------

528 bool

529 True if nodes are equal, False otherwise.

530 """

531 nlist1 = list(nodes1)

532 nlist2 = list(nodes2)

533 try:

534 d1 = dict(nlist1)

535 d2 = dict(nlist2)

536 except (ValueError, TypeError):

537 d1 = dict.fromkeys(nlist1)

538 d2 = dict.fromkeys(nlist2)

539 return d1 == d2

540

541

542def edges_equal(edges1, edges2, *, directed=False):

543 """Return whether edgelists are equal.

544

545 Equality here means equal as Python objects. Edge data must match

546 if included. Ordering of edges in an edgelist is not relevant;

547 ordering of nodes in an edge is only relevant if ``directed == True``.

548

549 Parameters

550 ----------

551 edges1, edges2 : iterables of tuples

552 Each tuple can be

553 an edge tuple ``(u, v)``, or

554 an edge tuple with data `dict` s ``(u, v, d)``, or

555 an edge tuple with keys and data `dict` s ``(u, v, k, d)``.

556

557 directed : bool, optional (default=False)

558 If `True`, edgelists are treated as coming from directed

559 graphs.

560

561 Returns

562 -------

563 bool

564 `True` if edgelists are equal, `False` otherwise.

565

566 Examples

567 --------

568 >>> G1 = nx.complete_graph(3)

569 >>> G2 = nx.cycle_graph(3)

570 >>> edges_equal(G1.edges, G2.edges)

571 True

572

573 Edge order is not taken into account:

574

575 >>> G1 = nx.Graph([(0, 1), (1, 2)])

576 >>> G2 = nx.Graph([(1, 2), (0, 1)])

577 >>> edges_equal(G1.edges, G2.edges)

578 True

579

580 The `directed` parameter controls whether edges are treated as

581 coming from directed graphs.

582

583 >>> DG1 = nx.DiGraph([(0, 1)])

584 >>> DG2 = nx.DiGraph([(1, 0)])

585 >>> edges_equal(DG1.edges, DG2.edges, directed=False) # Not recommended.

586 True

587 >>> edges_equal(DG1.edges, DG2.edges, directed=True)

588 False

589

590 This function is meant to be used on edgelists (i.e. the output of a

591 ``G.edges()`` call), and can give unexpected results on unprocessed

592 lists of edges:

593

594 >>> l1 = [(0, 1)]

595 >>> l2 = [(0, 1), (1, 0)]

596 >>> edges_equal(l1, l2) # Not recommended.

597 False

598 >>> G1 = nx.Graph(l1)

599 >>> G2 = nx.Graph(l2)

600 >>> edges_equal(G1.edges, G2.edges)

601 True

602 >>> DG1 = nx.DiGraph(l1)

603 >>> DG2 = nx.DiGraph(l2)

604 >>> edges_equal(DG1.edges, DG2.edges, directed=True)

605 False

606 """

607 d1 = defaultdict(list)

608 d2 = defaultdict(list)

609

610 for e1, e2 in zip_longest(edges1, edges2, fillvalue=None):

611 if e1 is None or e2 is None:

612 return False # One is longer.

613 for e, d in [(e1, d1), (e2, d2)]:

614 u, v, *data = e

615 d[u, v].append(data)

616 if not directed:

617 d[v, u].append(data)

618

619 # Can check one direction because lengths are the same.

620 return all(d1[e].count(data) == d2[e].count(data) for e in d1 for data in d1[e])

621

622

623def graphs_equal(graph1, graph2):

624 """Check if graphs are equal.

625

626 Equality here means equal as Python objects (not isomorphism).

627 Node, edge and graph data must match.

628

629 Parameters

630 ----------

631 graph1, graph2 : graph

632

633 Returns

634 -------

635 bool

636 True if graphs are equal, False otherwise.

637 """

638 return (

639 graph1.adj == graph2.adj

640 and graph1.nodes == graph2.nodes

641 and graph1.graph == graph2.graph

642 )

643

644

645def _clear_cache(G):

646 """Clear the cache of a graph (currently stores converted graphs).

647

648 Caching is controlled via ``nx.config.cache_converted_graphs`` configuration.

649 """

650 if cache := getattr(G, "__networkx_cache__", None):

651 cache.clear()

652

653

654def check_create_using(create_using, *, directed=None, multigraph=None, default=None):

655 """Assert that create_using has good properties

656

657 This checks for desired directedness and multi-edge properties.

658 It returns `create_using` unless that is `None` when it returns

659 the optionally specified default value.

660

661 Parameters

662 ----------

663 create_using : None, graph class or instance

664 The input value of create_using for a function.

665 directed : None or bool

666 Whether to check `create_using.is_directed() == directed`.

667 If None, do not assert directedness.

668 multigraph : None or bool

669 Whether to check `create_using.is_multigraph() == multigraph`.

670 If None, do not assert multi-edge property.

671 default : None or graph class

672 The graph class to return if create_using is None.

673

674 Returns

675 -------

676 create_using : graph class or instance

677 The provided graph class or instance, or if None, the `default` value.

678

679 Raises

680 ------

681 NetworkXError

682 When `create_using` doesn't match the properties specified by `directed`

683 or `multigraph` parameters.

684 """

685 if default is None:

686 default = nx.Graph

687 G = create_using if create_using is not None else default

688

689 G_directed = G.is_directed(None) if isinstance(G, type) else G.is_directed()

690 G_multigraph = G.is_multigraph(None) if isinstance(G, type) else G.is_multigraph()

691

692 if directed is not None:

693 if directed and not G_directed:

694 raise nx.NetworkXError("create_using must be directed")

695 if not directed and G_directed:

696 raise nx.NetworkXError("create_using must not be directed")

697

698 if multigraph is not None:

699 if multigraph and not G_multigraph:

700 raise nx.NetworkXError("create_using must be a multi-graph")

701 if not multigraph and G_multigraph:

702 raise nx.NetworkXError("create_using must not be a multi-graph")

703 return G