exploration.graphs
- Authors: Peter Mawhorter
- Consulted:
- Date: 2022-3-5
- Purpose: Low-level graph helpers & types.
This file defines tools on top of the networkx package which are
lower-level than the key types used for most tasks (see core.py for
those).
1""" 2- Authors: Peter Mawhorter 3- Consulted: 4- Date: 2022-3-5 5- Purpose: Low-level graph helpers & types. 6 7This file defines tools on top of the `networkx` package which are 8lower-level than the key types used for most tasks (see `core.py` for 9those). 10""" 11 12from typing import ( 13 Optional, Hashable, Dict, Union, Iterable, Tuple, Any, NoReturn, 14 Set, Sequence, cast, List, TypeVar, Generic, Callable, Self 15) 16 17import networkx as nx # type: ignore[import] 18 19 20Node = TypeVar('Node', bound=Hashable) 21"Type variable for graph nodes." 22 23Edge = TypeVar('Edge', bound=Hashable) 24"Type variable for graph edges." 25 26 27class UniqueExitsGraph(nx.MultiDiGraph, Generic[Node, Edge]): 28 """ 29 A `networkx.MultiDiGraph` which has unique-per-source-node names for 30 each edge. On top of base functionality, this uses some extra memory 31 to store per-edge outgoing (but not incoming) by-edge dictionaries, 32 so that you can iterate over edges by their names rather than 33 iterating over neighbor nodes. This helps in some circumstances where 34 you know the edge name but not the name of the room it connects to. 35 36 This does NOT change the meaning of any of the built-in 37 `networkx.MultiDiGraph` methods, but instead adds new methods for 38 access to nodes or attributes by node -> edge name. 39 """ 40 def __init__(self) -> None: 41 super().__init__() 42 # A dictionary that maps nodes to edge names, storing neighbor 43 # nodes for each edge. Those neighbor nodes can be used to look 44 # up edge attributes using the normal MultiDiGraph machinery. 45 self._byEdge: Dict[Node, Dict[Edge, Node]] = {} 46 47 # Note: not hashable 48 49 def __eq__(self, other: Any) -> bool: 50 """ 51 Compares two graphs for equality. Note that various kinds of 52 graphs can be equal to a `UniqueExitsGraph` as long as the node 53 names, edge names, and data attributes are all the same. 54 """ 55 if not isinstance(other, nx.Graph): 56 return False 57 else: 58 # Compare nodes 59 myNodes = list(self) 60 otherNodes = list(self) 61 if len(myNodes) != len(otherNodes): 62 return False 63 myNodes.sort() 64 otherNodes.sort() 65 if myNodes != otherNodes: 66 return False 67 68 # Compare edges 69 myEdges = list(self.edges) 70 otherEdges = list(other.edges) 71 if len(myEdges) != len(otherEdges): 72 return False 73 if len(myEdges) > 0 and len(myEdges[0]) != len(otherEdges[0]): 74 return False 75 myEdges.sort() 76 otherEdges.sort() 77 if myEdges != otherEdges: 78 return False 79 80 # Compare node data 81 if any( 82 self.nodes[node] != other.nodes[node] 83 for node in myNodes 84 ): 85 return False 86 87 # Compare edge data 88 if any( 89 self.edges[edge] != other.edges[edge] 90 for edge in myEdges 91 ): 92 return False 93 94 # Everything checks out... 95 return True 96 97 def new_edge_key(self, u: Node, v: Node) -> NoReturn: 98 """ 99 This method would normally be used to generate new edge keys. We 100 disable it, because we want to ensure that all edges are properly 101 labeled. 102 """ 103 raise NotImplementedError( 104 "Attempted to add an edge without specifying a key!" 105 ) 106 107 # TODO: Sort out networkx type annotations? 108 def add_node(self, node: Node, **attr: Any): # type:ignore [override] 109 """ 110 See `networkx.MultiDiGraph.add_node`. 111 """ 112 super().add_node(node, **attr) 113 self._byEdge[node] = {} # type Dict[Edge, Node] 114 115 def add_nodes_from( # type:ignore [override] 116 self, 117 nodes: Union[ 118 Iterable[Node], 119 Iterable[Tuple[Node, Dict[Any, Any]]] 120 ], 121 **attr: Any 122 ): 123 """ 124 See `networkx.MultiDiGraph.add_nodes_from`. 125 """ 126 super().add_nodes_from(nodes, **attr) 127 # Reassignment during tuple unpacking is not checkable... 128 n: Any 129 for n in nodes: 130 # Test for hashability & unpack tuple if not 131 try: 132 self._byEdge.get(n) 133 except TypeError: 134 n, _ = n # mypy can't handle this properly 135 self._byEdge[n] = {} 136 137 def remove_node(self, node: Node): 138 """ 139 See `networkx.MultiDiGraph.remove_node`. 140 """ 141 # Handle deletion from inherited structures 142 super().remove_node(node) 143 144 # Ignore if not present 145 if node not in self._byEdge: 146 return 147 148 # Remove record of outgoing edges 149 del self._byEdge[node] 150 151 # Remove incoming edge records 152 for source, edgeMap in self._byEdge.items(): 153 delete = [] 154 # Find all edges which go to the deleted node 155 # (this is not terribly efficient) 156 for edgeName, dest in edgeMap.items(): 157 if dest == node: 158 delete.append(edgeName) 159 # Delete them in a separate loop, so that we don't 160 # modify-while-iterating (not efficient and maybe 161 # unnecessary?) 162 for d in delete: 163 del edgeMap[d] 164 165 def remove_nodes_from(self, nodes: Iterable[Node]): 166 """ 167 See `networkx.MultiDiGraph.remove_nodes_from`. 168 """ 169 # First use inherited method to remove from inherited structures 170 super().remove_nodes_from(nodes) 171 # remove our custom info 172 for n in nodes: 173 if n in self._byEdge: 174 del self._byEdge[n] 175 176 for source, edgeMap in self._byEdge.items(): 177 delete = [] 178 # Find all edges that go to any deleted node 179 for edgeName, dest in edgeMap.items(): 180 if dest in nodes: 181 delete.append(edgeName) 182 183 # Remove edges in separate loop to avoid 184 # modifying-while-iterating (not efficient and maybe 185 # unnecessary?) 186 for d in delete: 187 del edgeMap[d] 188 189 def add_edge( # type:ignore [override] 190 self, 191 u_of_edge: Node, 192 v_of_edge: Node, 193 key: Edge, 194 **attr: Any 195 ) -> Edge: 196 """ 197 See `networkx.MultiDiGraph.add_edge`. 198 199 For a `UniqueExitsGraph`, an edge key must be supplied 200 explicitly. A `KeyError` will be raised if an edge using the 201 given key (i.e., name) already exists starting at the source node 202 (regardless of its destination!). 203 204 Returns the key it was given, to match the base `add_edge` API. 205 """ 206 if u_of_edge in self._byEdge and key in self._byEdge[u_of_edge]: 207 raise KeyError( 208 f"Cannot add a second edge {key!r} starting at node" 209 f" {u_of_edge!r}." 210 ) 211 super().add_edge(u_of_edge, v_of_edge, key, **attr) 212 # Note: the base add_edge function does NOT call our add_node 213 # function :( 214 if u_of_edge not in self._byEdge: 215 self._byEdge[u_of_edge] = {} 216 if v_of_edge not in self._byEdge: 217 self._byEdge[v_of_edge] = {} 218 # Add the edge to our by-edge-name structure 219 self._byEdge[u_of_edge][key] = v_of_edge 220 221 return key 222 223 def add_edges_from( 224 self, 225 ebunch_to_add: Any, 226 # Type should be this, but checker won't pass it: 227 # Union[ 228 # Iterable[Tuple[Node, Node, Edge]], 229 # Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 230 # ], 231 **attr: Any 232 ): 233 """ 234 See `networkx.MultiDiGraph.add_edges_from`. Tuples in the ebunch 235 must be 3- or 4-tuples that include a specific key (not just 236 data). Nodes will be created as necessary. 237 238 Raises a `KeyError` if adding an edge is impossible because it 239 re-uses the same edge name at a particular source node, but if an 240 attempt is made to add an existing edge with the same 241 destination, this will just update the relevant edge attributes. 242 243 Raises a `KeyError` instead of silently updating edge properties 244 if the existing edge was also added by an earlier entry in the 245 `ebunch_to_add` (i.e., if you are trying to add two edges at 246 once that go between the same pair of nodes and use the same 247 edge key). 248 249 >>> from exploration import graphs as eg 250 >>> g = eg.UniqueExitsGraph() 251 >>> g.add_edges_from([ 252 ... ('A', 'B', 'up'), 253 ... ('A', 'B', 'up2'), 254 ... ('B', 'A', 'down'), 255 ... ('B', 'B', 'self'), 256 ... ('B', 'C', 'next'), 257 ... ('C', 'B', 'prev') 258 ... ]) 259 >>> g.nodes 260 NodeView(('A', 'B', 'C')) 261 >>> for edge in g.edges: 262 ... print(edge) 263 ('A', 'B', 'up') 264 ('A', 'B', 'up2') 265 ('B', 'A', 'down') 266 ('B', 'B', 'self') 267 ('B', 'C', 'next') 268 ('C', 'B', 'prev') 269 """ 270 etuple: Any 271 for i, etuple in enumerate(ebunch_to_add): 272 if len(etuple) < 3: 273 raise ValueError( 274 f"Edges to add must contain explicit keys for a" 275 f" UniqueExitsGraph (edge #{i} had only 2 parts)." 276 ) 277 try: 278 hash(etuple[2]) 279 except TypeError: 280 raise ValueError( 281 f"Edges to add must contain explicit keys for a" 282 f" UniqueExitsGraph (edge #{i} had an unhashable 3rd" 283 f" component)." 284 ) 285 286 # Check edge name uniqueness 287 u, v, k = etuple[:3] 288 if u in self._byEdge and self._byEdge[u].get(k) != v: 289 raise KeyError( 290 f"Cannot add or update an edge named '{k}' from node" 291 f" '{u}' to node '{v}' because an edge by that name" 292 f" already exists and goes to a different" 293 f" destination." 294 ) 295 296 # Add edges to inherited structures 297 super().add_edges_from(ebunch_to_add, **attr) 298 299 # Note base implementation calls add_edge, so we don't need to 300 # add edges to our extra structure 301 302 def remove_edge( # type:ignore [override] 303 self, 304 u_of_edge: Node, 305 v_of_edge: Node, 306 key: Edge 307 ): 308 """ 309 See `networkx.MultiDiGraph.remove_edge`. A key is required in 310 this version to specify which edge we're removing. 311 312 Raises a NetworkXError if the target edge does not exist. 313 """ 314 super().remove_edge(u_of_edge, v_of_edge, key) 315 del self._byEdge[u_of_edge][key] 316 317 def remove_edges_from( 318 self, 319 ebunch: Union[ # type:ignore [override] 320 Iterable[Tuple[Node, Node, Edge]], 321 Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 322 ] 323 ): 324 """ 325 See `networkx.MultiDiGraph.remove_edges_from`. Edge tuples in 326 the ebunch must be 3- or 4-tuples that include a key. 327 328 If an edge being removed is not present, it will be ignored. 329 """ 330 if any(len(etuple) not in (3, 4) for etuple in ebunch): 331 raise ValueError( 332 "Edges to remove must be u, v, k 3-tuples or u, v, k, d" 333 " 4-tuples." 334 ) 335 # TODO: Fix networkx MultiDiGraph type stubs 336 super().remove_edges_from(ebunch) # type:ignore [arg-type] 337 # This calls self.remove_edge under the hood so we don't need 338 # extra cleanup steps for _byEdge. 339 340 def clear(self) -> None: 341 """ 342 See `networkx.MultiDiGraph.clear`. 343 """ 344 super().clear() 345 self._byEdge.clear() 346 347 def clear_edges(self) -> None: 348 """ 349 See `networkx.MultiDiGraph.clear_edges`. 350 """ 351 super().clear_edges() 352 for _, edgeMap in self._byEdge.items(): 353 edgeMap.clear() 354 355 def reverse(self) -> NoReturn: # type:ignore [override] 356 """ 357 See `networkx.MultiDiGraph.reverse`. 358 """ 359 raise NotImplementedError( 360 "Reversing a UniqueExitsGraph is not supported because" 361 " reversed edge names might not be unique." 362 ) 363 364 def removeEdgeByKey(self, uOfEdge: Node, key: Edge): 365 """ 366 Removes an edge sourced at a particular node that has a 367 particular key, without knowing what the destination is. 368 369 Raises a `KeyError` if the named edge does not exist. 370 371 ## Example 372 373 >>> g = UniqueExitsGraph() 374 >>> g.add_edges_from([ 375 ... ('A', 'B', 'up'), 376 ... ('A', 'B', 'up2'), 377 ... ('B', 'A', 'down'), 378 ... ('B', 'B', 'self'), 379 ... ('B', 'C', 'next'), 380 ... ('C', 'B', 'prev') 381 ... ]) 382 >>> g.getDestination('A', 'up') 383 'B' 384 >>> g.getDestination('A', 'up2') 385 'B' 386 >>> g.getDestination('B', 'self') 387 'B' 388 >>> g.removeEdgeByKey('A', 'up2') 389 >>> g.removeEdgeByKey('B', 'self') 390 >>> g.getDestination('A', 'up2') is None 391 True 392 >>> g.getDestination('B', 'self') is None 393 True 394 """ 395 vOfEdge = self._byEdge[uOfEdge][key] 396 super().remove_edge(uOfEdge, vOfEdge, key) 397 del self._byEdge[uOfEdge][key] 398 399 def removeEdgesByKey(self, edgeIds: Iterable[Tuple[Node, Edge]]): 400 """ 401 Removes multiple edges by source node and key, without needing 402 to know destination nodes. The `edgeIds` argument must be a list 403 of tuples containing source node, edge key pairs. 404 405 Silently ignores already-nonexistent edges. 406 407 ## Example 408 409 >>> g = UniqueExitsGraph() 410 >>> g.add_edges_from([ 411 ... ('A', 'B', 'up'), 412 ... ('A', 'B', 'up2'), 413 ... ('B', 'A', 'down'), 414 ... ('B', 'B', 'self'), 415 ... ('B', 'C', 'next'), 416 ... ('C', 'B', 'prev') 417 ... ]) 418 >>> g.getDestination('A', 'up') 419 'B' 420 >>> g.getDestination('A', 'up2') 421 'B' 422 >>> g.getDestination('B', 'self') 423 'B' 424 >>> g.removeEdgesByKey([('A', 'up2'), ('B', 'self')]) 425 >>> g.getDestination('A', 'up2') is None 426 True 427 >>> g.getDestination('B', 'self') is None 428 True 429 """ 430 for source, key in edgeIds: 431 if key in self._byEdge.get(source, {}): 432 self.removeEdgeByKey(source, key) 433 # Otherwise ignore this edge... 434 435 def connections( 436 self, 437 includeOnly: Optional[ 438 Callable[[Node, Edge, Node, Self], bool] 439 ] = None 440 ) -> nx.Graph: 441 """ 442 Returns an undirected graph with the same nodes IDs as the base 443 graph but none of the node or edge attributes. Nodes which have 444 any connection between them in either direction in the original 445 graph will be connected by a single edge in the connections 446 graph. (Note that self-edges will remain for nodes that had 447 actions.) 448 449 If an `includeOnly` function is provided, it will be given a 450 source node, an edge, a destination node, and the entire graph as 451 arguments. It should return a boolean, and for edges where it 452 returns False, these won't be included in the final graph. Note 453 that because two nodes can be connected by multiple edges in 454 either direction, filtering out a single edge may not sever the 455 connection between two nodes in the final graph. 456 """ 457 result: nx.Graph = nx.Graph() 458 result.add_nodes_from(self) 459 if includeOnly is None: 460 # TODO: Fix networkx type annotations? 461 edges = self.edges(keys=False, data=False) # type:ignore 462 result.add_edges_from(edges) 463 else: 464 # TODO: Fix networkx type annotations? 465 edges = self.edges(keys=True) # type:ignore 466 for (src, dst, edge) in edges: 467 if includeOnly(src, edge, dst, self): 468 result.add_edge(src, dst) 469 return result 470 471 def destinationsFrom(self, source: Node) -> Dict[Edge, Node]: 472 """ 473 Given a source node, returns a dictionary mapping the keys of all 474 outgoing edges from that node to their destination nodes. Raises 475 a `KeyError` if the node is not present in the graph. 476 477 Editing the dictionary returned could cause serious problems, so 478 please don't; it will be updated live as the graph is changed. 479 480 ## Example 481 482 >>> g = UniqueExitsGraph() 483 >>> g.add_edges_from([ 484 ... ('A', 'B', 'up'), 485 ... ('A', 'B', 'up2'), 486 ... ('B', 'A', 'down'), 487 ... ('B', 'B', 'self'), 488 ... ('B', 'C', 'next'), 489 ... ('C', 'B', 'prev') 490 ... ]) 491 >>> g.destinationsFrom('A') 492 {'up': 'B', 'up2': 'B'} 493 >>> g.destinationsFrom('B') 494 {'down': 'A', 'self': 'B', 'next': 'C'} 495 >>> g.destinationsFrom('C') 496 {'prev': 'B'} 497 >>> g.destinationsFrom('D') 498 Traceback (most recent call last): 499 ... 500 KeyError... 501 """ 502 return self._byEdge[source] 503 504 def destination(self, source: Node, edge: Edge) -> Node: 505 """ 506 Given a source node and an edge key, looks up and returns the 507 destination node for that edge. Raises a `KeyError` if there is no 508 edge from the specified node with the specified name (including 509 if the specified node doesn't exist). 510 511 ## Example 512 513 >>> g = UniqueExitsGraph() 514 >>> g.add_edges_from([ 515 ... ('A', 'B', 'up'), 516 ... ('A', 'B', 'up2'), 517 ... ('B', 'A', 'down'), 518 ... ('B', 'B', 'self'), 519 ... ('B', 'C', 'next'), 520 ... ('C', 'B', 'prev') 521 ... ]) 522 >>> g.destination('A', 'up') 523 'B' 524 >>> g.destination('A', 'up2') 525 'B' 526 >>> g.destination('B', 'down') 527 'A' 528 >>> g.destination('A', 'nonexistent') 529 Traceback (most recent call last): 530 ... 531 KeyError... 532 >>> g.destination('D', 'any') 533 Traceback (most recent call last): 534 ... 535 KeyError... 536 """ 537 return self._byEdge[source][edge] 538 539 def getDestination( 540 self, 541 source: Node, 542 edge: Edge, 543 default: Any = None 544 ) -> Optional[Node]: 545 """ 546 Works like `destination`, but instead of raising a `KeyError` if 547 the node or edge is missing, it returns a default value (with a 548 default default of `None`). 549 550 ## Example 551 552 >>> g = UniqueExitsGraph() 553 >>> g.add_edges_from([ 554 ... ('A', 'B', 'up'), 555 ... ('A', 'B', 'up2'), 556 ... ('B', 'A', 'down'), 557 ... ('B', 'B', 'self'), 558 ... ('B', 'C', 'next'), 559 ... ('C', 'B', 'prev') 560 ... ]) 561 >>> g.getDestination('A', 'up') 562 'B' 563 >>> g.getDestination('A', 'up2') 564 'B' 565 >>> g.getDestination('B', 'down') 566 'A' 567 >>> g.getDestination('A', 'nonexistent') is None 568 True 569 >>> g.getDestination('A', 'nonexistent', 'default') 570 'default' 571 >>> g.getDestination('D', 'any') is None 572 True 573 """ 574 return self._byEdge.get(source, {}).get(edge, default) 575 576 def allEdgesTo( 577 self, 578 destination: Node 579 ) -> List[Tuple[Node, Edge]]: 580 """ 581 Searches the entire graph for edges whose destinations are the 582 specified destination, and returns a list of (node, edge) pairs 583 indicating the source node and edge name for each of those edges. 584 Self-edges are included in this list. 585 586 ## Example 587 588 >>> g = UniqueExitsGraph() 589 >>> g.add_edges_from([ 590 ... ('A', 'B', 'up'), 591 ... ('A', 'B', 'up2'), 592 ... ('B', 'A', 'down'), 593 ... ('B', 'B', 'self'), 594 ... ('B', 'C', 'next'), 595 ... ('C', 'B', 'prev') 596 ... ]) 597 >>> g.allEdgesTo('A') 598 [('B', 'down')] 599 >>> g.allEdgesTo('B') 600 [('A', 'up'), ('A', 'up2'), ('B', 'self'), ('C', 'prev')] 601 >>> g.allEdgesTo('C') 602 [('B', 'next')] 603 >>> g.allEdgesTo('D') 604 [] 605 """ 606 results = [] 607 for node in self: 608 fromThere = self[node] 609 toHere = fromThere.get(destination, {}) 610 for edgeKey in toHere: 611 results.append((node, edgeKey)) 612 613 return results 614 615 def allEdges(self) -> List[Tuple[Node, Node, Edge]]: 616 """ 617 Returns a list of tuples containing source node, destination 618 node, and then edge node, which includes each edge in the graph 619 once. 620 """ 621 # TODO: Fix networkx type annotations 622 return self.edges(keys=True) # type: ignore 623 624 def textMapObj( 625 self, 626 edgeSep: str = '::', 627 external: Optional[Set[Node]] = None, 628 explorationOrder: Optional[Tuple[Node, Sequence[Edge]]] = None, 629 edgeOrders: Union[ 630 Dict[Node, Sequence[Edge]], 631 Dict[Node, Dict[Edge, Any]], 632 None 633 ] = None 634 ): 635 """ 636 Returns a special object which is JSON-serializable and which 637 when serialized creates a semi-human-usable text-format map of 638 the graph. 639 640 The object consists of nested dictionaries, one per node, where 641 keys are node name + edge name strings (combined using the 642 `edgeSep` argument, default is '::'). The value for each key is 643 one of: 644 645 1. Another dictionary representing the node that edge leads 646 to, which can in turn have dictionary values... 647 2. A string naming a destination node that's already represented 648 elsewhere (or naming the current node for self-edges). 649 650 Any node present in the specified `external` set will be linked 651 to instead of listed out, even if it exists in the graph. The 652 `external` set **will be modified** by this function to include 653 all visited nodes in the graph. 654 655 If an `explorationOrder` is provided, it must be a tuple 656 specifying a start node followed by a sequence of edges that 657 indicates the path taken, and the edges will be visited 658 according to that order (this only matters in Python 3.7+ where 659 dictionaries have consistent order). A `ValueError` will be 660 raised if an invalid exploration order is provided. The path 661 list will be ignored if `edgeOrders` is provided explicitly. 662 663 TODO: What about unexplorable graphs (allow node names in place 664 of edge names in exploration order?)?!? 665 666 If `edgeOrders` is provided directly, it will override the 667 path part of the `explorationOrder` to determine the ordering of 668 edges at each node. If not and `explorationOrder` is provided, it 669 will be deduced from the `explorationOrder`. If neither is 670 present, ordering will follow whatever natural order is in the 671 graph, which in most cases should be order-of-creation. 672 673 Notes: 674 - For the format to avoid ambiguity, the `edgeSep` value must be 675 a string which does not appear in any node or edge names. 676 - Nodes and edge values will be converted to strings to build the 677 map. 678 - Node and edge properties are not represented in the resulting 679 object. 680 - For a variety of reasons, the result cannot be converted back 681 to a graph object. This is not intended for use as a JSON 682 serialization route (see the `networkx.readwrite.json_graph` 683 module for some built-in options). 684 - To get a string representation, one could do: 685 `json.dumps(graph.textMapObj())` 686 687 ## Examples 688 689 >>> from exploration import graphs as eg 690 >>> import json 691 >>> g = eg.UniqueExitsGraph() 692 >>> g.add_edges_from([ 693 ... ('A', 'B', 'up'), 694 ... ('A', 'B', 'up2'), 695 ... ('B', 'A', 'down'), 696 ... ('B', 'B', 'self'), 697 ... ('B', 'C', 'next'), 698 ... ('C', 'B', 'prev') 699 ... ]) 700 >>> print(json.dumps(g.textMapObj(), indent=2)) 701 { 702 "A::up": { 703 "B::down": "A", 704 "B::self": "B", 705 "B::next": { 706 "C::prev": "B" 707 } 708 }, 709 "A::up2": "B" 710 } 711 """ 712 # We use `external` as our visited set 713 if external is None: 714 external = set() 715 716 if explorationOrder is not None: 717 here, path = explorationOrder 718 else: 719 # Find first non-external node as our starting node 720 for here in self.nodes: 721 if here not in external: 722 break 723 724 # Path is empty 725 path = [] 726 727 # Determine edge ordering for each node from exploration order 728 # or by natural ordering if no explorationOrder is available 729 if edgeOrders is None: 730 edgeOrders = cast( 731 Dict[Node, Dict[Edge, Any]], 732 {} 733 ) 734 current = here 735 for i in range(len(path)): 736 edge = path[i] 737 # Add this edge next in the ordering for this node 738 orderHere: Dict[Edge, Any] = edgeOrders.setdefault(current, {}) 739 # Note: we use a dictionary here because dictionaries do 740 # preserve insertion ordering (3.7+) and we need to both 741 # keep things in order AND do a bunch of lookups to 742 # avoid duplicates. 743 if edge not in orderHere: 744 orderHere[edge] = True 745 746 # Move to next node 747 if edge not in self._byEdge[current]: 748 raise ValueError( 749 f"Invalid edge in exploration order path: at" 750 f" step {i} we reached node {current} and were" 751 f" supposed to take edge {edge} but that edge" 752 f" does not exist." 753 ) 754 current = self._byEdge[current][edge] 755 756 # Add any unexplored nodes and/or edges in natural order 757 for node in self.nodes: 758 orderHere = edgeOrders.setdefault(node, {}) 759 for edge in self._byEdge[node]: 760 if edge not in orderHere: 761 orderHere[edge] = True 762 763 result = {} 764 external.add(here) 765 # Now loop through keys of this node 766 for key in edgeOrders[here]: 767 combined = str(here) + edgeSep + str(key) 768 dest = self._byEdge[here][key] 769 if dest in external: 770 # links, including self-links 771 result[combined] = str(dest) 772 else: 773 # Recurse 774 result[combined] = self.textMapObj( 775 edgeSep, 776 external, 777 (dest, []), # empty path since we have edgeOrders 778 edgeOrders 779 ) 780 781 return result
Type variable for graph nodes.
Type variable for graph edges.
28class UniqueExitsGraph(nx.MultiDiGraph, Generic[Node, Edge]): 29 """ 30 A `networkx.MultiDiGraph` which has unique-per-source-node names for 31 each edge. On top of base functionality, this uses some extra memory 32 to store per-edge outgoing (but not incoming) by-edge dictionaries, 33 so that you can iterate over edges by their names rather than 34 iterating over neighbor nodes. This helps in some circumstances where 35 you know the edge name but not the name of the room it connects to. 36 37 This does NOT change the meaning of any of the built-in 38 `networkx.MultiDiGraph` methods, but instead adds new methods for 39 access to nodes or attributes by node -> edge name. 40 """ 41 def __init__(self) -> None: 42 super().__init__() 43 # A dictionary that maps nodes to edge names, storing neighbor 44 # nodes for each edge. Those neighbor nodes can be used to look 45 # up edge attributes using the normal MultiDiGraph machinery. 46 self._byEdge: Dict[Node, Dict[Edge, Node]] = {} 47 48 # Note: not hashable 49 50 def __eq__(self, other: Any) -> bool: 51 """ 52 Compares two graphs for equality. Note that various kinds of 53 graphs can be equal to a `UniqueExitsGraph` as long as the node 54 names, edge names, and data attributes are all the same. 55 """ 56 if not isinstance(other, nx.Graph): 57 return False 58 else: 59 # Compare nodes 60 myNodes = list(self) 61 otherNodes = list(self) 62 if len(myNodes) != len(otherNodes): 63 return False 64 myNodes.sort() 65 otherNodes.sort() 66 if myNodes != otherNodes: 67 return False 68 69 # Compare edges 70 myEdges = list(self.edges) 71 otherEdges = list(other.edges) 72 if len(myEdges) != len(otherEdges): 73 return False 74 if len(myEdges) > 0 and len(myEdges[0]) != len(otherEdges[0]): 75 return False 76 myEdges.sort() 77 otherEdges.sort() 78 if myEdges != otherEdges: 79 return False 80 81 # Compare node data 82 if any( 83 self.nodes[node] != other.nodes[node] 84 for node in myNodes 85 ): 86 return False 87 88 # Compare edge data 89 if any( 90 self.edges[edge] != other.edges[edge] 91 for edge in myEdges 92 ): 93 return False 94 95 # Everything checks out... 96 return True 97 98 def new_edge_key(self, u: Node, v: Node) -> NoReturn: 99 """ 100 This method would normally be used to generate new edge keys. We 101 disable it, because we want to ensure that all edges are properly 102 labeled. 103 """ 104 raise NotImplementedError( 105 "Attempted to add an edge without specifying a key!" 106 ) 107 108 # TODO: Sort out networkx type annotations? 109 def add_node(self, node: Node, **attr: Any): # type:ignore [override] 110 """ 111 See `networkx.MultiDiGraph.add_node`. 112 """ 113 super().add_node(node, **attr) 114 self._byEdge[node] = {} # type Dict[Edge, Node] 115 116 def add_nodes_from( # type:ignore [override] 117 self, 118 nodes: Union[ 119 Iterable[Node], 120 Iterable[Tuple[Node, Dict[Any, Any]]] 121 ], 122 **attr: Any 123 ): 124 """ 125 See `networkx.MultiDiGraph.add_nodes_from`. 126 """ 127 super().add_nodes_from(nodes, **attr) 128 # Reassignment during tuple unpacking is not checkable... 129 n: Any 130 for n in nodes: 131 # Test for hashability & unpack tuple if not 132 try: 133 self._byEdge.get(n) 134 except TypeError: 135 n, _ = n # mypy can't handle this properly 136 self._byEdge[n] = {} 137 138 def remove_node(self, node: Node): 139 """ 140 See `networkx.MultiDiGraph.remove_node`. 141 """ 142 # Handle deletion from inherited structures 143 super().remove_node(node) 144 145 # Ignore if not present 146 if node not in self._byEdge: 147 return 148 149 # Remove record of outgoing edges 150 del self._byEdge[node] 151 152 # Remove incoming edge records 153 for source, edgeMap in self._byEdge.items(): 154 delete = [] 155 # Find all edges which go to the deleted node 156 # (this is not terribly efficient) 157 for edgeName, dest in edgeMap.items(): 158 if dest == node: 159 delete.append(edgeName) 160 # Delete them in a separate loop, so that we don't 161 # modify-while-iterating (not efficient and maybe 162 # unnecessary?) 163 for d in delete: 164 del edgeMap[d] 165 166 def remove_nodes_from(self, nodes: Iterable[Node]): 167 """ 168 See `networkx.MultiDiGraph.remove_nodes_from`. 169 """ 170 # First use inherited method to remove from inherited structures 171 super().remove_nodes_from(nodes) 172 # remove our custom info 173 for n in nodes: 174 if n in self._byEdge: 175 del self._byEdge[n] 176 177 for source, edgeMap in self._byEdge.items(): 178 delete = [] 179 # Find all edges that go to any deleted node 180 for edgeName, dest in edgeMap.items(): 181 if dest in nodes: 182 delete.append(edgeName) 183 184 # Remove edges in separate loop to avoid 185 # modifying-while-iterating (not efficient and maybe 186 # unnecessary?) 187 for d in delete: 188 del edgeMap[d] 189 190 def add_edge( # type:ignore [override] 191 self, 192 u_of_edge: Node, 193 v_of_edge: Node, 194 key: Edge, 195 **attr: Any 196 ) -> Edge: 197 """ 198 See `networkx.MultiDiGraph.add_edge`. 199 200 For a `UniqueExitsGraph`, an edge key must be supplied 201 explicitly. A `KeyError` will be raised if an edge using the 202 given key (i.e., name) already exists starting at the source node 203 (regardless of its destination!). 204 205 Returns the key it was given, to match the base `add_edge` API. 206 """ 207 if u_of_edge in self._byEdge and key in self._byEdge[u_of_edge]: 208 raise KeyError( 209 f"Cannot add a second edge {key!r} starting at node" 210 f" {u_of_edge!r}." 211 ) 212 super().add_edge(u_of_edge, v_of_edge, key, **attr) 213 # Note: the base add_edge function does NOT call our add_node 214 # function :( 215 if u_of_edge not in self._byEdge: 216 self._byEdge[u_of_edge] = {} 217 if v_of_edge not in self._byEdge: 218 self._byEdge[v_of_edge] = {} 219 # Add the edge to our by-edge-name structure 220 self._byEdge[u_of_edge][key] = v_of_edge 221 222 return key 223 224 def add_edges_from( 225 self, 226 ebunch_to_add: Any, 227 # Type should be this, but checker won't pass it: 228 # Union[ 229 # Iterable[Tuple[Node, Node, Edge]], 230 # Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 231 # ], 232 **attr: Any 233 ): 234 """ 235 See `networkx.MultiDiGraph.add_edges_from`. Tuples in the ebunch 236 must be 3- or 4-tuples that include a specific key (not just 237 data). Nodes will be created as necessary. 238 239 Raises a `KeyError` if adding an edge is impossible because it 240 re-uses the same edge name at a particular source node, but if an 241 attempt is made to add an existing edge with the same 242 destination, this will just update the relevant edge attributes. 243 244 Raises a `KeyError` instead of silently updating edge properties 245 if the existing edge was also added by an earlier entry in the 246 `ebunch_to_add` (i.e., if you are trying to add two edges at 247 once that go between the same pair of nodes and use the same 248 edge key). 249 250 >>> from exploration import graphs as eg 251 >>> g = eg.UniqueExitsGraph() 252 >>> g.add_edges_from([ 253 ... ('A', 'B', 'up'), 254 ... ('A', 'B', 'up2'), 255 ... ('B', 'A', 'down'), 256 ... ('B', 'B', 'self'), 257 ... ('B', 'C', 'next'), 258 ... ('C', 'B', 'prev') 259 ... ]) 260 >>> g.nodes 261 NodeView(('A', 'B', 'C')) 262 >>> for edge in g.edges: 263 ... print(edge) 264 ('A', 'B', 'up') 265 ('A', 'B', 'up2') 266 ('B', 'A', 'down') 267 ('B', 'B', 'self') 268 ('B', 'C', 'next') 269 ('C', 'B', 'prev') 270 """ 271 etuple: Any 272 for i, etuple in enumerate(ebunch_to_add): 273 if len(etuple) < 3: 274 raise ValueError( 275 f"Edges to add must contain explicit keys for a" 276 f" UniqueExitsGraph (edge #{i} had only 2 parts)." 277 ) 278 try: 279 hash(etuple[2]) 280 except TypeError: 281 raise ValueError( 282 f"Edges to add must contain explicit keys for a" 283 f" UniqueExitsGraph (edge #{i} had an unhashable 3rd" 284 f" component)." 285 ) 286 287 # Check edge name uniqueness 288 u, v, k = etuple[:3] 289 if u in self._byEdge and self._byEdge[u].get(k) != v: 290 raise KeyError( 291 f"Cannot add or update an edge named '{k}' from node" 292 f" '{u}' to node '{v}' because an edge by that name" 293 f" already exists and goes to a different" 294 f" destination." 295 ) 296 297 # Add edges to inherited structures 298 super().add_edges_from(ebunch_to_add, **attr) 299 300 # Note base implementation calls add_edge, so we don't need to 301 # add edges to our extra structure 302 303 def remove_edge( # type:ignore [override] 304 self, 305 u_of_edge: Node, 306 v_of_edge: Node, 307 key: Edge 308 ): 309 """ 310 See `networkx.MultiDiGraph.remove_edge`. A key is required in 311 this version to specify which edge we're removing. 312 313 Raises a NetworkXError if the target edge does not exist. 314 """ 315 super().remove_edge(u_of_edge, v_of_edge, key) 316 del self._byEdge[u_of_edge][key] 317 318 def remove_edges_from( 319 self, 320 ebunch: Union[ # type:ignore [override] 321 Iterable[Tuple[Node, Node, Edge]], 322 Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 323 ] 324 ): 325 """ 326 See `networkx.MultiDiGraph.remove_edges_from`. Edge tuples in 327 the ebunch must be 3- or 4-tuples that include a key. 328 329 If an edge being removed is not present, it will be ignored. 330 """ 331 if any(len(etuple) not in (3, 4) for etuple in ebunch): 332 raise ValueError( 333 "Edges to remove must be u, v, k 3-tuples or u, v, k, d" 334 " 4-tuples." 335 ) 336 # TODO: Fix networkx MultiDiGraph type stubs 337 super().remove_edges_from(ebunch) # type:ignore [arg-type] 338 # This calls self.remove_edge under the hood so we don't need 339 # extra cleanup steps for _byEdge. 340 341 def clear(self) -> None: 342 """ 343 See `networkx.MultiDiGraph.clear`. 344 """ 345 super().clear() 346 self._byEdge.clear() 347 348 def clear_edges(self) -> None: 349 """ 350 See `networkx.MultiDiGraph.clear_edges`. 351 """ 352 super().clear_edges() 353 for _, edgeMap in self._byEdge.items(): 354 edgeMap.clear() 355 356 def reverse(self) -> NoReturn: # type:ignore [override] 357 """ 358 See `networkx.MultiDiGraph.reverse`. 359 """ 360 raise NotImplementedError( 361 "Reversing a UniqueExitsGraph is not supported because" 362 " reversed edge names might not be unique." 363 ) 364 365 def removeEdgeByKey(self, uOfEdge: Node, key: Edge): 366 """ 367 Removes an edge sourced at a particular node that has a 368 particular key, without knowing what the destination is. 369 370 Raises a `KeyError` if the named edge does not exist. 371 372 ## Example 373 374 >>> g = UniqueExitsGraph() 375 >>> g.add_edges_from([ 376 ... ('A', 'B', 'up'), 377 ... ('A', 'B', 'up2'), 378 ... ('B', 'A', 'down'), 379 ... ('B', 'B', 'self'), 380 ... ('B', 'C', 'next'), 381 ... ('C', 'B', 'prev') 382 ... ]) 383 >>> g.getDestination('A', 'up') 384 'B' 385 >>> g.getDestination('A', 'up2') 386 'B' 387 >>> g.getDestination('B', 'self') 388 'B' 389 >>> g.removeEdgeByKey('A', 'up2') 390 >>> g.removeEdgeByKey('B', 'self') 391 >>> g.getDestination('A', 'up2') is None 392 True 393 >>> g.getDestination('B', 'self') is None 394 True 395 """ 396 vOfEdge = self._byEdge[uOfEdge][key] 397 super().remove_edge(uOfEdge, vOfEdge, key) 398 del self._byEdge[uOfEdge][key] 399 400 def removeEdgesByKey(self, edgeIds: Iterable[Tuple[Node, Edge]]): 401 """ 402 Removes multiple edges by source node and key, without needing 403 to know destination nodes. The `edgeIds` argument must be a list 404 of tuples containing source node, edge key pairs. 405 406 Silently ignores already-nonexistent edges. 407 408 ## Example 409 410 >>> g = UniqueExitsGraph() 411 >>> g.add_edges_from([ 412 ... ('A', 'B', 'up'), 413 ... ('A', 'B', 'up2'), 414 ... ('B', 'A', 'down'), 415 ... ('B', 'B', 'self'), 416 ... ('B', 'C', 'next'), 417 ... ('C', 'B', 'prev') 418 ... ]) 419 >>> g.getDestination('A', 'up') 420 'B' 421 >>> g.getDestination('A', 'up2') 422 'B' 423 >>> g.getDestination('B', 'self') 424 'B' 425 >>> g.removeEdgesByKey([('A', 'up2'), ('B', 'self')]) 426 >>> g.getDestination('A', 'up2') is None 427 True 428 >>> g.getDestination('B', 'self') is None 429 True 430 """ 431 for source, key in edgeIds: 432 if key in self._byEdge.get(source, {}): 433 self.removeEdgeByKey(source, key) 434 # Otherwise ignore this edge... 435 436 def connections( 437 self, 438 includeOnly: Optional[ 439 Callable[[Node, Edge, Node, Self], bool] 440 ] = None 441 ) -> nx.Graph: 442 """ 443 Returns an undirected graph with the same nodes IDs as the base 444 graph but none of the node or edge attributes. Nodes which have 445 any connection between them in either direction in the original 446 graph will be connected by a single edge in the connections 447 graph. (Note that self-edges will remain for nodes that had 448 actions.) 449 450 If an `includeOnly` function is provided, it will be given a 451 source node, an edge, a destination node, and the entire graph as 452 arguments. It should return a boolean, and for edges where it 453 returns False, these won't be included in the final graph. Note 454 that because two nodes can be connected by multiple edges in 455 either direction, filtering out a single edge may not sever the 456 connection between two nodes in the final graph. 457 """ 458 result: nx.Graph = nx.Graph() 459 result.add_nodes_from(self) 460 if includeOnly is None: 461 # TODO: Fix networkx type annotations? 462 edges = self.edges(keys=False, data=False) # type:ignore 463 result.add_edges_from(edges) 464 else: 465 # TODO: Fix networkx type annotations? 466 edges = self.edges(keys=True) # type:ignore 467 for (src, dst, edge) in edges: 468 if includeOnly(src, edge, dst, self): 469 result.add_edge(src, dst) 470 return result 471 472 def destinationsFrom(self, source: Node) -> Dict[Edge, Node]: 473 """ 474 Given a source node, returns a dictionary mapping the keys of all 475 outgoing edges from that node to their destination nodes. Raises 476 a `KeyError` if the node is not present in the graph. 477 478 Editing the dictionary returned could cause serious problems, so 479 please don't; it will be updated live as the graph is changed. 480 481 ## Example 482 483 >>> g = UniqueExitsGraph() 484 >>> g.add_edges_from([ 485 ... ('A', 'B', 'up'), 486 ... ('A', 'B', 'up2'), 487 ... ('B', 'A', 'down'), 488 ... ('B', 'B', 'self'), 489 ... ('B', 'C', 'next'), 490 ... ('C', 'B', 'prev') 491 ... ]) 492 >>> g.destinationsFrom('A') 493 {'up': 'B', 'up2': 'B'} 494 >>> g.destinationsFrom('B') 495 {'down': 'A', 'self': 'B', 'next': 'C'} 496 >>> g.destinationsFrom('C') 497 {'prev': 'B'} 498 >>> g.destinationsFrom('D') 499 Traceback (most recent call last): 500 ... 501 KeyError... 502 """ 503 return self._byEdge[source] 504 505 def destination(self, source: Node, edge: Edge) -> Node: 506 """ 507 Given a source node and an edge key, looks up and returns the 508 destination node for that edge. Raises a `KeyError` if there is no 509 edge from the specified node with the specified name (including 510 if the specified node doesn't exist). 511 512 ## Example 513 514 >>> g = UniqueExitsGraph() 515 >>> g.add_edges_from([ 516 ... ('A', 'B', 'up'), 517 ... ('A', 'B', 'up2'), 518 ... ('B', 'A', 'down'), 519 ... ('B', 'B', 'self'), 520 ... ('B', 'C', 'next'), 521 ... ('C', 'B', 'prev') 522 ... ]) 523 >>> g.destination('A', 'up') 524 'B' 525 >>> g.destination('A', 'up2') 526 'B' 527 >>> g.destination('B', 'down') 528 'A' 529 >>> g.destination('A', 'nonexistent') 530 Traceback (most recent call last): 531 ... 532 KeyError... 533 >>> g.destination('D', 'any') 534 Traceback (most recent call last): 535 ... 536 KeyError... 537 """ 538 return self._byEdge[source][edge] 539 540 def getDestination( 541 self, 542 source: Node, 543 edge: Edge, 544 default: Any = None 545 ) -> Optional[Node]: 546 """ 547 Works like `destination`, but instead of raising a `KeyError` if 548 the node or edge is missing, it returns a default value (with a 549 default default of `None`). 550 551 ## Example 552 553 >>> g = UniqueExitsGraph() 554 >>> g.add_edges_from([ 555 ... ('A', 'B', 'up'), 556 ... ('A', 'B', 'up2'), 557 ... ('B', 'A', 'down'), 558 ... ('B', 'B', 'self'), 559 ... ('B', 'C', 'next'), 560 ... ('C', 'B', 'prev') 561 ... ]) 562 >>> g.getDestination('A', 'up') 563 'B' 564 >>> g.getDestination('A', 'up2') 565 'B' 566 >>> g.getDestination('B', 'down') 567 'A' 568 >>> g.getDestination('A', 'nonexistent') is None 569 True 570 >>> g.getDestination('A', 'nonexistent', 'default') 571 'default' 572 >>> g.getDestination('D', 'any') is None 573 True 574 """ 575 return self._byEdge.get(source, {}).get(edge, default) 576 577 def allEdgesTo( 578 self, 579 destination: Node 580 ) -> List[Tuple[Node, Edge]]: 581 """ 582 Searches the entire graph for edges whose destinations are the 583 specified destination, and returns a list of (node, edge) pairs 584 indicating the source node and edge name for each of those edges. 585 Self-edges are included in this list. 586 587 ## Example 588 589 >>> g = UniqueExitsGraph() 590 >>> g.add_edges_from([ 591 ... ('A', 'B', 'up'), 592 ... ('A', 'B', 'up2'), 593 ... ('B', 'A', 'down'), 594 ... ('B', 'B', 'self'), 595 ... ('B', 'C', 'next'), 596 ... ('C', 'B', 'prev') 597 ... ]) 598 >>> g.allEdgesTo('A') 599 [('B', 'down')] 600 >>> g.allEdgesTo('B') 601 [('A', 'up'), ('A', 'up2'), ('B', 'self'), ('C', 'prev')] 602 >>> g.allEdgesTo('C') 603 [('B', 'next')] 604 >>> g.allEdgesTo('D') 605 [] 606 """ 607 results = [] 608 for node in self: 609 fromThere = self[node] 610 toHere = fromThere.get(destination, {}) 611 for edgeKey in toHere: 612 results.append((node, edgeKey)) 613 614 return results 615 616 def allEdges(self) -> List[Tuple[Node, Node, Edge]]: 617 """ 618 Returns a list of tuples containing source node, destination 619 node, and then edge node, which includes each edge in the graph 620 once. 621 """ 622 # TODO: Fix networkx type annotations 623 return self.edges(keys=True) # type: ignore 624 625 def textMapObj( 626 self, 627 edgeSep: str = '::', 628 external: Optional[Set[Node]] = None, 629 explorationOrder: Optional[Tuple[Node, Sequence[Edge]]] = None, 630 edgeOrders: Union[ 631 Dict[Node, Sequence[Edge]], 632 Dict[Node, Dict[Edge, Any]], 633 None 634 ] = None 635 ): 636 """ 637 Returns a special object which is JSON-serializable and which 638 when serialized creates a semi-human-usable text-format map of 639 the graph. 640 641 The object consists of nested dictionaries, one per node, where 642 keys are node name + edge name strings (combined using the 643 `edgeSep` argument, default is '::'). The value for each key is 644 one of: 645 646 1. Another dictionary representing the node that edge leads 647 to, which can in turn have dictionary values... 648 2. A string naming a destination node that's already represented 649 elsewhere (or naming the current node for self-edges). 650 651 Any node present in the specified `external` set will be linked 652 to instead of listed out, even if it exists in the graph. The 653 `external` set **will be modified** by this function to include 654 all visited nodes in the graph. 655 656 If an `explorationOrder` is provided, it must be a tuple 657 specifying a start node followed by a sequence of edges that 658 indicates the path taken, and the edges will be visited 659 according to that order (this only matters in Python 3.7+ where 660 dictionaries have consistent order). A `ValueError` will be 661 raised if an invalid exploration order is provided. The path 662 list will be ignored if `edgeOrders` is provided explicitly. 663 664 TODO: What about unexplorable graphs (allow node names in place 665 of edge names in exploration order?)?!? 666 667 If `edgeOrders` is provided directly, it will override the 668 path part of the `explorationOrder` to determine the ordering of 669 edges at each node. If not and `explorationOrder` is provided, it 670 will be deduced from the `explorationOrder`. If neither is 671 present, ordering will follow whatever natural order is in the 672 graph, which in most cases should be order-of-creation. 673 674 Notes: 675 - For the format to avoid ambiguity, the `edgeSep` value must be 676 a string which does not appear in any node or edge names. 677 - Nodes and edge values will be converted to strings to build the 678 map. 679 - Node and edge properties are not represented in the resulting 680 object. 681 - For a variety of reasons, the result cannot be converted back 682 to a graph object. This is not intended for use as a JSON 683 serialization route (see the `networkx.readwrite.json_graph` 684 module for some built-in options). 685 - To get a string representation, one could do: 686 `json.dumps(graph.textMapObj())` 687 688 ## Examples 689 690 >>> from exploration import graphs as eg 691 >>> import json 692 >>> g = eg.UniqueExitsGraph() 693 >>> g.add_edges_from([ 694 ... ('A', 'B', 'up'), 695 ... ('A', 'B', 'up2'), 696 ... ('B', 'A', 'down'), 697 ... ('B', 'B', 'self'), 698 ... ('B', 'C', 'next'), 699 ... ('C', 'B', 'prev') 700 ... ]) 701 >>> print(json.dumps(g.textMapObj(), indent=2)) 702 { 703 "A::up": { 704 "B::down": "A", 705 "B::self": "B", 706 "B::next": { 707 "C::prev": "B" 708 } 709 }, 710 "A::up2": "B" 711 } 712 """ 713 # We use `external` as our visited set 714 if external is None: 715 external = set() 716 717 if explorationOrder is not None: 718 here, path = explorationOrder 719 else: 720 # Find first non-external node as our starting node 721 for here in self.nodes: 722 if here not in external: 723 break 724 725 # Path is empty 726 path = [] 727 728 # Determine edge ordering for each node from exploration order 729 # or by natural ordering if no explorationOrder is available 730 if edgeOrders is None: 731 edgeOrders = cast( 732 Dict[Node, Dict[Edge, Any]], 733 {} 734 ) 735 current = here 736 for i in range(len(path)): 737 edge = path[i] 738 # Add this edge next in the ordering for this node 739 orderHere: Dict[Edge, Any] = edgeOrders.setdefault(current, {}) 740 # Note: we use a dictionary here because dictionaries do 741 # preserve insertion ordering (3.7+) and we need to both 742 # keep things in order AND do a bunch of lookups to 743 # avoid duplicates. 744 if edge not in orderHere: 745 orderHere[edge] = True 746 747 # Move to next node 748 if edge not in self._byEdge[current]: 749 raise ValueError( 750 f"Invalid edge in exploration order path: at" 751 f" step {i} we reached node {current} and were" 752 f" supposed to take edge {edge} but that edge" 753 f" does not exist." 754 ) 755 current = self._byEdge[current][edge] 756 757 # Add any unexplored nodes and/or edges in natural order 758 for node in self.nodes: 759 orderHere = edgeOrders.setdefault(node, {}) 760 for edge in self._byEdge[node]: 761 if edge not in orderHere: 762 orderHere[edge] = True 763 764 result = {} 765 external.add(here) 766 # Now loop through keys of this node 767 for key in edgeOrders[here]: 768 combined = str(here) + edgeSep + str(key) 769 dest = self._byEdge[here][key] 770 if dest in external: 771 # links, including self-links 772 result[combined] = str(dest) 773 else: 774 # Recurse 775 result[combined] = self.textMapObj( 776 edgeSep, 777 external, 778 (dest, []), # empty path since we have edgeOrders 779 edgeOrders 780 ) 781 782 return result
A networkx.MultiDiGraph which has unique-per-source-node names for
each edge. On top of base functionality, this uses some extra memory
to store per-edge outgoing (but not incoming) by-edge dictionaries,
so that you can iterate over edges by their names rather than
iterating over neighbor nodes. This helps in some circumstances where
you know the edge name but not the name of the room it connects to.
This does NOT change the meaning of any of the built-in
networkx.MultiDiGraph methods, but instead adds new methods for
access to nodes or attributes by node -> edge name.
98 def new_edge_key(self, u: Node, v: Node) -> NoReturn: 99 """ 100 This method would normally be used to generate new edge keys. We 101 disable it, because we want to ensure that all edges are properly 102 labeled. 103 """ 104 raise NotImplementedError( 105 "Attempted to add an edge without specifying a key!" 106 )
This method would normally be used to generate new edge keys. We disable it, because we want to ensure that all edges are properly labeled.
109 def add_node(self, node: Node, **attr: Any): # type:ignore [override] 110 """ 111 See `networkx.MultiDiGraph.add_node`. 112 """ 113 super().add_node(node, **attr) 114 self._byEdge[node] = {} # type Dict[Edge, Node]
See networkx.MultiDiGraph.add_node.
116 def add_nodes_from( # type:ignore [override] 117 self, 118 nodes: Union[ 119 Iterable[Node], 120 Iterable[Tuple[Node, Dict[Any, Any]]] 121 ], 122 **attr: Any 123 ): 124 """ 125 See `networkx.MultiDiGraph.add_nodes_from`. 126 """ 127 super().add_nodes_from(nodes, **attr) 128 # Reassignment during tuple unpacking is not checkable... 129 n: Any 130 for n in nodes: 131 # Test for hashability & unpack tuple if not 132 try: 133 self._byEdge.get(n) 134 except TypeError: 135 n, _ = n # mypy can't handle this properly 136 self._byEdge[n] = {}
See networkx.MultiDiGraph.add_nodes_from.
138 def remove_node(self, node: Node): 139 """ 140 See `networkx.MultiDiGraph.remove_node`. 141 """ 142 # Handle deletion from inherited structures 143 super().remove_node(node) 144 145 # Ignore if not present 146 if node not in self._byEdge: 147 return 148 149 # Remove record of outgoing edges 150 del self._byEdge[node] 151 152 # Remove incoming edge records 153 for source, edgeMap in self._byEdge.items(): 154 delete = [] 155 # Find all edges which go to the deleted node 156 # (this is not terribly efficient) 157 for edgeName, dest in edgeMap.items(): 158 if dest == node: 159 delete.append(edgeName) 160 # Delete them in a separate loop, so that we don't 161 # modify-while-iterating (not efficient and maybe 162 # unnecessary?) 163 for d in delete: 164 del edgeMap[d]
See networkx.MultiDiGraph.remove_node.
166 def remove_nodes_from(self, nodes: Iterable[Node]): 167 """ 168 See `networkx.MultiDiGraph.remove_nodes_from`. 169 """ 170 # First use inherited method to remove from inherited structures 171 super().remove_nodes_from(nodes) 172 # remove our custom info 173 for n in nodes: 174 if n in self._byEdge: 175 del self._byEdge[n] 176 177 for source, edgeMap in self._byEdge.items(): 178 delete = [] 179 # Find all edges that go to any deleted node 180 for edgeName, dest in edgeMap.items(): 181 if dest in nodes: 182 delete.append(edgeName) 183 184 # Remove edges in separate loop to avoid 185 # modifying-while-iterating (not efficient and maybe 186 # unnecessary?) 187 for d in delete: 188 del edgeMap[d]
See networkx.MultiDiGraph.remove_nodes_from.
224 def add_edges_from( 225 self, 226 ebunch_to_add: Any, 227 # Type should be this, but checker won't pass it: 228 # Union[ 229 # Iterable[Tuple[Node, Node, Edge]], 230 # Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 231 # ], 232 **attr: Any 233 ): 234 """ 235 See `networkx.MultiDiGraph.add_edges_from`. Tuples in the ebunch 236 must be 3- or 4-tuples that include a specific key (not just 237 data). Nodes will be created as necessary. 238 239 Raises a `KeyError` if adding an edge is impossible because it 240 re-uses the same edge name at a particular source node, but if an 241 attempt is made to add an existing edge with the same 242 destination, this will just update the relevant edge attributes. 243 244 Raises a `KeyError` instead of silently updating edge properties 245 if the existing edge was also added by an earlier entry in the 246 `ebunch_to_add` (i.e., if you are trying to add two edges at 247 once that go between the same pair of nodes and use the same 248 edge key). 249 250 >>> from exploration import graphs as eg 251 >>> g = eg.UniqueExitsGraph() 252 >>> g.add_edges_from([ 253 ... ('A', 'B', 'up'), 254 ... ('A', 'B', 'up2'), 255 ... ('B', 'A', 'down'), 256 ... ('B', 'B', 'self'), 257 ... ('B', 'C', 'next'), 258 ... ('C', 'B', 'prev') 259 ... ]) 260 >>> g.nodes 261 NodeView(('A', 'B', 'C')) 262 >>> for edge in g.edges: 263 ... print(edge) 264 ('A', 'B', 'up') 265 ('A', 'B', 'up2') 266 ('B', 'A', 'down') 267 ('B', 'B', 'self') 268 ('B', 'C', 'next') 269 ('C', 'B', 'prev') 270 """ 271 etuple: Any 272 for i, etuple in enumerate(ebunch_to_add): 273 if len(etuple) < 3: 274 raise ValueError( 275 f"Edges to add must contain explicit keys for a" 276 f" UniqueExitsGraph (edge #{i} had only 2 parts)." 277 ) 278 try: 279 hash(etuple[2]) 280 except TypeError: 281 raise ValueError( 282 f"Edges to add must contain explicit keys for a" 283 f" UniqueExitsGraph (edge #{i} had an unhashable 3rd" 284 f" component)." 285 ) 286 287 # Check edge name uniqueness 288 u, v, k = etuple[:3] 289 if u in self._byEdge and self._byEdge[u].get(k) != v: 290 raise KeyError( 291 f"Cannot add or update an edge named '{k}' from node" 292 f" '{u}' to node '{v}' because an edge by that name" 293 f" already exists and goes to a different" 294 f" destination." 295 ) 296 297 # Add edges to inherited structures 298 super().add_edges_from(ebunch_to_add, **attr) 299 300 # Note base implementation calls add_edge, so we don't need to 301 # add edges to our extra structure
See networkx.MultiDiGraph.add_edges_from. Tuples in the ebunch
must be 3- or 4-tuples that include a specific key (not just
data). Nodes will be created as necessary.
Raises a KeyError if adding an edge is impossible because it
re-uses the same edge name at a particular source node, but if an
attempt is made to add an existing edge with the same
destination, this will just update the relevant edge attributes.
Raises a KeyError instead of silently updating edge properties
if the existing edge was also added by an earlier entry in the
ebunch_to_add (i.e., if you are trying to add two edges at
once that go between the same pair of nodes and use the same
edge key).
>>> from exploration import graphs as eg
>>> g = eg.UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.nodes
NodeView(('A', 'B', 'C'))
>>> for edge in g.edges:
... print(edge)
('A', 'B', 'up')
('A', 'B', 'up2')
('B', 'A', 'down')
('B', 'B', 'self')
('B', 'C', 'next')
('C', 'B', 'prev')
318 def remove_edges_from( 319 self, 320 ebunch: Union[ # type:ignore [override] 321 Iterable[Tuple[Node, Node, Edge]], 322 Iterable[Tuple[Node, Node, Edge, Dict[Any, Any]]] 323 ] 324 ): 325 """ 326 See `networkx.MultiDiGraph.remove_edges_from`. Edge tuples in 327 the ebunch must be 3- or 4-tuples that include a key. 328 329 If an edge being removed is not present, it will be ignored. 330 """ 331 if any(len(etuple) not in (3, 4) for etuple in ebunch): 332 raise ValueError( 333 "Edges to remove must be u, v, k 3-tuples or u, v, k, d" 334 " 4-tuples." 335 ) 336 # TODO: Fix networkx MultiDiGraph type stubs 337 super().remove_edges_from(ebunch) # type:ignore [arg-type] 338 # This calls self.remove_edge under the hood so we don't need 339 # extra cleanup steps for _byEdge.
See networkx.MultiDiGraph.remove_edges_from. Edge tuples in
the ebunch must be 3- or 4-tuples that include a key.
If an edge being removed is not present, it will be ignored.
341 def clear(self) -> None: 342 """ 343 See `networkx.MultiDiGraph.clear`. 344 """ 345 super().clear() 346 self._byEdge.clear()
See networkx.MultiDiGraph.clear.
348 def clear_edges(self) -> None: 349 """ 350 See `networkx.MultiDiGraph.clear_edges`. 351 """ 352 super().clear_edges() 353 for _, edgeMap in self._byEdge.items(): 354 edgeMap.clear()
See networkx.MultiDiGraph.clear_edges.
365 def removeEdgeByKey(self, uOfEdge: Node, key: Edge): 366 """ 367 Removes an edge sourced at a particular node that has a 368 particular key, without knowing what the destination is. 369 370 Raises a `KeyError` if the named edge does not exist. 371 372 ## Example 373 374 >>> g = UniqueExitsGraph() 375 >>> g.add_edges_from([ 376 ... ('A', 'B', 'up'), 377 ... ('A', 'B', 'up2'), 378 ... ('B', 'A', 'down'), 379 ... ('B', 'B', 'self'), 380 ... ('B', 'C', 'next'), 381 ... ('C', 'B', 'prev') 382 ... ]) 383 >>> g.getDestination('A', 'up') 384 'B' 385 >>> g.getDestination('A', 'up2') 386 'B' 387 >>> g.getDestination('B', 'self') 388 'B' 389 >>> g.removeEdgeByKey('A', 'up2') 390 >>> g.removeEdgeByKey('B', 'self') 391 >>> g.getDestination('A', 'up2') is None 392 True 393 >>> g.getDestination('B', 'self') is None 394 True 395 """ 396 vOfEdge = self._byEdge[uOfEdge][key] 397 super().remove_edge(uOfEdge, vOfEdge, key) 398 del self._byEdge[uOfEdge][key]
Removes an edge sourced at a particular node that has a particular key, without knowing what the destination is.
Raises a KeyError if the named edge does not exist.
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.getDestination('A', 'up')
'B'
>>> g.getDestination('A', 'up2')
'B'
>>> g.getDestination('B', 'self')
'B'
>>> g.removeEdgeByKey('A', 'up2')
>>> g.removeEdgeByKey('B', 'self')
>>> g.getDestination('A', 'up2') is None
True
>>> g.getDestination('B', 'self') is None
True
400 def removeEdgesByKey(self, edgeIds: Iterable[Tuple[Node, Edge]]): 401 """ 402 Removes multiple edges by source node and key, without needing 403 to know destination nodes. The `edgeIds` argument must be a list 404 of tuples containing source node, edge key pairs. 405 406 Silently ignores already-nonexistent edges. 407 408 ## Example 409 410 >>> g = UniqueExitsGraph() 411 >>> g.add_edges_from([ 412 ... ('A', 'B', 'up'), 413 ... ('A', 'B', 'up2'), 414 ... ('B', 'A', 'down'), 415 ... ('B', 'B', 'self'), 416 ... ('B', 'C', 'next'), 417 ... ('C', 'B', 'prev') 418 ... ]) 419 >>> g.getDestination('A', 'up') 420 'B' 421 >>> g.getDestination('A', 'up2') 422 'B' 423 >>> g.getDestination('B', 'self') 424 'B' 425 >>> g.removeEdgesByKey([('A', 'up2'), ('B', 'self')]) 426 >>> g.getDestination('A', 'up2') is None 427 True 428 >>> g.getDestination('B', 'self') is None 429 True 430 """ 431 for source, key in edgeIds: 432 if key in self._byEdge.get(source, {}): 433 self.removeEdgeByKey(source, key) 434 # Otherwise ignore this edge...
Removes multiple edges by source node and key, without needing
to know destination nodes. The edgeIds argument must be a list
of tuples containing source node, edge key pairs.
Silently ignores already-nonexistent edges.
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.getDestination('A', 'up')
'B'
>>> g.getDestination('A', 'up2')
'B'
>>> g.getDestination('B', 'self')
'B'
>>> g.removeEdgesByKey([('A', 'up2'), ('B', 'self')])
>>> g.getDestination('A', 'up2') is None
True
>>> g.getDestination('B', 'self') is None
True
436 def connections( 437 self, 438 includeOnly: Optional[ 439 Callable[[Node, Edge, Node, Self], bool] 440 ] = None 441 ) -> nx.Graph: 442 """ 443 Returns an undirected graph with the same nodes IDs as the base 444 graph but none of the node or edge attributes. Nodes which have 445 any connection between them in either direction in the original 446 graph will be connected by a single edge in the connections 447 graph. (Note that self-edges will remain for nodes that had 448 actions.) 449 450 If an `includeOnly` function is provided, it will be given a 451 source node, an edge, a destination node, and the entire graph as 452 arguments. It should return a boolean, and for edges where it 453 returns False, these won't be included in the final graph. Note 454 that because two nodes can be connected by multiple edges in 455 either direction, filtering out a single edge may not sever the 456 connection between two nodes in the final graph. 457 """ 458 result: nx.Graph = nx.Graph() 459 result.add_nodes_from(self) 460 if includeOnly is None: 461 # TODO: Fix networkx type annotations? 462 edges = self.edges(keys=False, data=False) # type:ignore 463 result.add_edges_from(edges) 464 else: 465 # TODO: Fix networkx type annotations? 466 edges = self.edges(keys=True) # type:ignore 467 for (src, dst, edge) in edges: 468 if includeOnly(src, edge, dst, self): 469 result.add_edge(src, dst) 470 return result
Returns an undirected graph with the same nodes IDs as the base graph but none of the node or edge attributes. Nodes which have any connection between them in either direction in the original graph will be connected by a single edge in the connections graph. (Note that self-edges will remain for nodes that had actions.)
If an includeOnly function is provided, it will be given a
source node, an edge, a destination node, and the entire graph as
arguments. It should return a boolean, and for edges where it
returns False, these won't be included in the final graph. Note
that because two nodes can be connected by multiple edges in
either direction, filtering out a single edge may not sever the
connection between two nodes in the final graph.
472 def destinationsFrom(self, source: Node) -> Dict[Edge, Node]: 473 """ 474 Given a source node, returns a dictionary mapping the keys of all 475 outgoing edges from that node to their destination nodes. Raises 476 a `KeyError` if the node is not present in the graph. 477 478 Editing the dictionary returned could cause serious problems, so 479 please don't; it will be updated live as the graph is changed. 480 481 ## Example 482 483 >>> g = UniqueExitsGraph() 484 >>> g.add_edges_from([ 485 ... ('A', 'B', 'up'), 486 ... ('A', 'B', 'up2'), 487 ... ('B', 'A', 'down'), 488 ... ('B', 'B', 'self'), 489 ... ('B', 'C', 'next'), 490 ... ('C', 'B', 'prev') 491 ... ]) 492 >>> g.destinationsFrom('A') 493 {'up': 'B', 'up2': 'B'} 494 >>> g.destinationsFrom('B') 495 {'down': 'A', 'self': 'B', 'next': 'C'} 496 >>> g.destinationsFrom('C') 497 {'prev': 'B'} 498 >>> g.destinationsFrom('D') 499 Traceback (most recent call last): 500 ... 501 KeyError... 502 """ 503 return self._byEdge[source]
Given a source node, returns a dictionary mapping the keys of all
outgoing edges from that node to their destination nodes. Raises
a KeyError if the node is not present in the graph.
Editing the dictionary returned could cause serious problems, so please don't; it will be updated live as the graph is changed.
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.destinationsFrom('A')
{'up': 'B', 'up2': 'B'}
>>> g.destinationsFrom('B')
{'down': 'A', 'self': 'B', 'next': 'C'}
>>> g.destinationsFrom('C')
{'prev': 'B'}
>>> g.destinationsFrom('D')
Traceback (most recent call last):
...
KeyError...
505 def destination(self, source: Node, edge: Edge) -> Node: 506 """ 507 Given a source node and an edge key, looks up and returns the 508 destination node for that edge. Raises a `KeyError` if there is no 509 edge from the specified node with the specified name (including 510 if the specified node doesn't exist). 511 512 ## Example 513 514 >>> g = UniqueExitsGraph() 515 >>> g.add_edges_from([ 516 ... ('A', 'B', 'up'), 517 ... ('A', 'B', 'up2'), 518 ... ('B', 'A', 'down'), 519 ... ('B', 'B', 'self'), 520 ... ('B', 'C', 'next'), 521 ... ('C', 'B', 'prev') 522 ... ]) 523 >>> g.destination('A', 'up') 524 'B' 525 >>> g.destination('A', 'up2') 526 'B' 527 >>> g.destination('B', 'down') 528 'A' 529 >>> g.destination('A', 'nonexistent') 530 Traceback (most recent call last): 531 ... 532 KeyError... 533 >>> g.destination('D', 'any') 534 Traceback (most recent call last): 535 ... 536 KeyError... 537 """ 538 return self._byEdge[source][edge]
Given a source node and an edge key, looks up and returns the
destination node for that edge. Raises a KeyError if there is no
edge from the specified node with the specified name (including
if the specified node doesn't exist).
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.destination('A', 'up')
'B'
>>> g.destination('A', 'up2')
'B'
>>> g.destination('B', 'down')
'A'
>>> g.destination('A', 'nonexistent')
Traceback (most recent call last):
...
KeyError...
>>> g.destination('D', 'any')
Traceback (most recent call last):
...
KeyError...
540 def getDestination( 541 self, 542 source: Node, 543 edge: Edge, 544 default: Any = None 545 ) -> Optional[Node]: 546 """ 547 Works like `destination`, but instead of raising a `KeyError` if 548 the node or edge is missing, it returns a default value (with a 549 default default of `None`). 550 551 ## Example 552 553 >>> g = UniqueExitsGraph() 554 >>> g.add_edges_from([ 555 ... ('A', 'B', 'up'), 556 ... ('A', 'B', 'up2'), 557 ... ('B', 'A', 'down'), 558 ... ('B', 'B', 'self'), 559 ... ('B', 'C', 'next'), 560 ... ('C', 'B', 'prev') 561 ... ]) 562 >>> g.getDestination('A', 'up') 563 'B' 564 >>> g.getDestination('A', 'up2') 565 'B' 566 >>> g.getDestination('B', 'down') 567 'A' 568 >>> g.getDestination('A', 'nonexistent') is None 569 True 570 >>> g.getDestination('A', 'nonexistent', 'default') 571 'default' 572 >>> g.getDestination('D', 'any') is None 573 True 574 """ 575 return self._byEdge.get(source, {}).get(edge, default)
Works like destination, but instead of raising a KeyError if
the node or edge is missing, it returns a default value (with a
default default of None).
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.getDestination('A', 'up')
'B'
>>> g.getDestination('A', 'up2')
'B'
>>> g.getDestination('B', 'down')
'A'
>>> g.getDestination('A', 'nonexistent') is None
True
>>> g.getDestination('A', 'nonexistent', 'default')
'default'
>>> g.getDestination('D', 'any') is None
True
577 def allEdgesTo( 578 self, 579 destination: Node 580 ) -> List[Tuple[Node, Edge]]: 581 """ 582 Searches the entire graph for edges whose destinations are the 583 specified destination, and returns a list of (node, edge) pairs 584 indicating the source node and edge name for each of those edges. 585 Self-edges are included in this list. 586 587 ## Example 588 589 >>> g = UniqueExitsGraph() 590 >>> g.add_edges_from([ 591 ... ('A', 'B', 'up'), 592 ... ('A', 'B', 'up2'), 593 ... ('B', 'A', 'down'), 594 ... ('B', 'B', 'self'), 595 ... ('B', 'C', 'next'), 596 ... ('C', 'B', 'prev') 597 ... ]) 598 >>> g.allEdgesTo('A') 599 [('B', 'down')] 600 >>> g.allEdgesTo('B') 601 [('A', 'up'), ('A', 'up2'), ('B', 'self'), ('C', 'prev')] 602 >>> g.allEdgesTo('C') 603 [('B', 'next')] 604 >>> g.allEdgesTo('D') 605 [] 606 """ 607 results = [] 608 for node in self: 609 fromThere = self[node] 610 toHere = fromThere.get(destination, {}) 611 for edgeKey in toHere: 612 results.append((node, edgeKey)) 613 614 return results
Searches the entire graph for edges whose destinations are the specified destination, and returns a list of (node, edge) pairs indicating the source node and edge name for each of those edges. Self-edges are included in this list.
Example
>>> g = UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> g.allEdgesTo('A')
[('B', 'down')]
>>> g.allEdgesTo('B')
[('A', 'up'), ('A', 'up2'), ('B', 'self'), ('C', 'prev')]
>>> g.allEdgesTo('C')
[('B', 'next')]
>>> g.allEdgesTo('D')
[]
616 def allEdges(self) -> List[Tuple[Node, Node, Edge]]: 617 """ 618 Returns a list of tuples containing source node, destination 619 node, and then edge node, which includes each edge in the graph 620 once. 621 """ 622 # TODO: Fix networkx type annotations 623 return self.edges(keys=True) # type: ignore
Returns a list of tuples containing source node, destination node, and then edge node, which includes each edge in the graph once.
625 def textMapObj( 626 self, 627 edgeSep: str = '::', 628 external: Optional[Set[Node]] = None, 629 explorationOrder: Optional[Tuple[Node, Sequence[Edge]]] = None, 630 edgeOrders: Union[ 631 Dict[Node, Sequence[Edge]], 632 Dict[Node, Dict[Edge, Any]], 633 None 634 ] = None 635 ): 636 """ 637 Returns a special object which is JSON-serializable and which 638 when serialized creates a semi-human-usable text-format map of 639 the graph. 640 641 The object consists of nested dictionaries, one per node, where 642 keys are node name + edge name strings (combined using the 643 `edgeSep` argument, default is '::'). The value for each key is 644 one of: 645 646 1. Another dictionary representing the node that edge leads 647 to, which can in turn have dictionary values... 648 2. A string naming a destination node that's already represented 649 elsewhere (or naming the current node for self-edges). 650 651 Any node present in the specified `external` set will be linked 652 to instead of listed out, even if it exists in the graph. The 653 `external` set **will be modified** by this function to include 654 all visited nodes in the graph. 655 656 If an `explorationOrder` is provided, it must be a tuple 657 specifying a start node followed by a sequence of edges that 658 indicates the path taken, and the edges will be visited 659 according to that order (this only matters in Python 3.7+ where 660 dictionaries have consistent order). A `ValueError` will be 661 raised if an invalid exploration order is provided. The path 662 list will be ignored if `edgeOrders` is provided explicitly. 663 664 TODO: What about unexplorable graphs (allow node names in place 665 of edge names in exploration order?)?!? 666 667 If `edgeOrders` is provided directly, it will override the 668 path part of the `explorationOrder` to determine the ordering of 669 edges at each node. If not and `explorationOrder` is provided, it 670 will be deduced from the `explorationOrder`. If neither is 671 present, ordering will follow whatever natural order is in the 672 graph, which in most cases should be order-of-creation. 673 674 Notes: 675 - For the format to avoid ambiguity, the `edgeSep` value must be 676 a string which does not appear in any node or edge names. 677 - Nodes and edge values will be converted to strings to build the 678 map. 679 - Node and edge properties are not represented in the resulting 680 object. 681 - For a variety of reasons, the result cannot be converted back 682 to a graph object. This is not intended for use as a JSON 683 serialization route (see the `networkx.readwrite.json_graph` 684 module for some built-in options). 685 - To get a string representation, one could do: 686 `json.dumps(graph.textMapObj())` 687 688 ## Examples 689 690 >>> from exploration import graphs as eg 691 >>> import json 692 >>> g = eg.UniqueExitsGraph() 693 >>> g.add_edges_from([ 694 ... ('A', 'B', 'up'), 695 ... ('A', 'B', 'up2'), 696 ... ('B', 'A', 'down'), 697 ... ('B', 'B', 'self'), 698 ... ('B', 'C', 'next'), 699 ... ('C', 'B', 'prev') 700 ... ]) 701 >>> print(json.dumps(g.textMapObj(), indent=2)) 702 { 703 "A::up": { 704 "B::down": "A", 705 "B::self": "B", 706 "B::next": { 707 "C::prev": "B" 708 } 709 }, 710 "A::up2": "B" 711 } 712 """ 713 # We use `external` as our visited set 714 if external is None: 715 external = set() 716 717 if explorationOrder is not None: 718 here, path = explorationOrder 719 else: 720 # Find first non-external node as our starting node 721 for here in self.nodes: 722 if here not in external: 723 break 724 725 # Path is empty 726 path = [] 727 728 # Determine edge ordering for each node from exploration order 729 # or by natural ordering if no explorationOrder is available 730 if edgeOrders is None: 731 edgeOrders = cast( 732 Dict[Node, Dict[Edge, Any]], 733 {} 734 ) 735 current = here 736 for i in range(len(path)): 737 edge = path[i] 738 # Add this edge next in the ordering for this node 739 orderHere: Dict[Edge, Any] = edgeOrders.setdefault(current, {}) 740 # Note: we use a dictionary here because dictionaries do 741 # preserve insertion ordering (3.7+) and we need to both 742 # keep things in order AND do a bunch of lookups to 743 # avoid duplicates. 744 if edge not in orderHere: 745 orderHere[edge] = True 746 747 # Move to next node 748 if edge not in self._byEdge[current]: 749 raise ValueError( 750 f"Invalid edge in exploration order path: at" 751 f" step {i} we reached node {current} and were" 752 f" supposed to take edge {edge} but that edge" 753 f" does not exist." 754 ) 755 current = self._byEdge[current][edge] 756 757 # Add any unexplored nodes and/or edges in natural order 758 for node in self.nodes: 759 orderHere = edgeOrders.setdefault(node, {}) 760 for edge in self._byEdge[node]: 761 if edge not in orderHere: 762 orderHere[edge] = True 763 764 result = {} 765 external.add(here) 766 # Now loop through keys of this node 767 for key in edgeOrders[here]: 768 combined = str(here) + edgeSep + str(key) 769 dest = self._byEdge[here][key] 770 if dest in external: 771 # links, including self-links 772 result[combined] = str(dest) 773 else: 774 # Recurse 775 result[combined] = self.textMapObj( 776 edgeSep, 777 external, 778 (dest, []), # empty path since we have edgeOrders 779 edgeOrders 780 ) 781 782 return result
Returns a special object which is JSON-serializable and which when serialized creates a semi-human-usable text-format map of the graph.
The object consists of nested dictionaries, one per node, where
keys are node name + edge name strings (combined using the
edgeSep argument, default is '::'). The value for each key is
one of:
- Another dictionary representing the node that edge leads to, which can in turn have dictionary values...
- A string naming a destination node that's already represented elsewhere (or naming the current node for self-edges).
Any node present in the specified external set will be linked
to instead of listed out, even if it exists in the graph. The
external set will be modified by this function to include
all visited nodes in the graph.
If an explorationOrder is provided, it must be a tuple
specifying a start node followed by a sequence of edges that
indicates the path taken, and the edges will be visited
according to that order (this only matters in Python 3.7+ where
dictionaries have consistent order). A ValueError will be
raised if an invalid exploration order is provided. The path
list will be ignored if edgeOrders is provided explicitly.
TODO: What about unexplorable graphs (allow node names in place of edge names in exploration order?)?!?
If edgeOrders is provided directly, it will override the
path part of the explorationOrder to determine the ordering of
edges at each node. If not and explorationOrder is provided, it
will be deduced from the explorationOrder. If neither is
present, ordering will follow whatever natural order is in the
graph, which in most cases should be order-of-creation.
Notes:
- For the format to avoid ambiguity, the
edgeSepvalue must be a string which does not appear in any node or edge names. - Nodes and edge values will be converted to strings to build the map.
- Node and edge properties are not represented in the resulting object.
- For a variety of reasons, the result cannot be converted back
to a graph object. This is not intended for use as a JSON
serialization route (see the
networkx.readwrite.json_graphmodule for some built-in options). - To get a string representation, one could do:
json.dumps(graph.textMapObj())
Examples
>>> from exploration import graphs as eg
>>> import json
>>> g = eg.UniqueExitsGraph()
>>> g.add_edges_from([
... ('A', 'B', 'up'),
... ('A', 'B', 'up2'),
... ('B', 'A', 'down'),
... ('B', 'B', 'self'),
... ('B', 'C', 'next'),
... ('C', 'B', 'prev')
... ])
>>> print(json.dumps(g.textMapObj(), indent=2))
{
"A::up": {
"B::down": "A",
"B::self": "B",
"B::next": {
"C::prev": "B"
}
},
"A::up2": "B"
}
Inherited Members
- networkx.classes.multidigraph.MultiDiGraph
- MultiDiGraph
- edge_key_dict_factory
- adj
- succ
- pred
- add_edge
- remove_edge
- edges
- out_edges
- in_edges
- degree
- in_degree
- out_degree
- is_multigraph
- is_directed
- to_undirected
- reverse
- networkx.classes.multigraph.MultiGraph
- to_directed_class
- to_undirected_class
- has_edge
- get_edge_data
- copy
- to_directed
- number_of_edges
- networkx.classes.digraph.DiGraph
- graph
- has_successor
- has_predecessor
- successors
- neighbors
- predecessors
- networkx.classes.graph.Graph
- node_dict_factory
- node_attr_dict_factory
- adjlist_outer_dict_factory
- adjlist_inner_dict_factory
- edge_attr_dict_factory
- graph_attr_dict_factory
- name
- nodes
- number_of_nodes
- order
- has_node
- add_weighted_edges_from
- update
- adjacency
- subgraph
- edge_subgraph
- size
- nbunch_iter