exploration.graphs

  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
 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 destinationsFrom(self, source: Node) -> Dict[Edge, Node]:
436        """
437        Given a source node, returns a dictionary mapping the keys of all
438        outgoing edges from that node to their destination nodes. Raises
439        a `KeyError` if the node is not present in the graph.
440
441        Editing the dictionary returned could cause serious problems, so
442        please don't; it will be updated live as the graph is changed.
443
444        ## Example
445
446        >>> g = UniqueExitsGraph()
447        >>> g.add_edges_from([
448        ...     ('A', 'B', 'up'),
449        ...     ('A', 'B', 'up2'),
450        ...     ('B', 'A', 'down'),
451        ...     ('B', 'B', 'self'),
452        ...     ('B', 'C', 'next'),
453        ...     ('C', 'B', 'prev')
454        ... ])
455        >>> g.destinationsFrom('A')
456        {'up': 'B', 'up2': 'B'}
457        >>> g.destinationsFrom('B')
458        {'down': 'A', 'self': 'B', 'next': 'C'}
459        >>> g.destinationsFrom('C')
460        {'prev': 'B'}
461        >>> g.destinationsFrom('D')
462        Traceback (most recent call last):
463        ...
464        KeyError...
465        """
466        return self._byEdge[source]
467
468    def destination(self, source: Node, edge: Edge) -> Node:
469        """
470        Given a source node and an edge key, looks up and returns the
471        destination node for that edge. Raises a `KeyError` if there is no
472        edge from the specified node with the specified name.
473
474        ## Example
475
476        >>> g = UniqueExitsGraph()
477        >>> g.add_edges_from([
478        ...     ('A', 'B', 'up'),
479        ...     ('A', 'B', 'up2'),
480        ...     ('B', 'A', 'down'),
481        ...     ('B', 'B', 'self'),
482        ...     ('B', 'C', 'next'),
483        ...     ('C', 'B', 'prev')
484        ... ])
485        >>> g.destination('A', 'up')
486        'B'
487        >>> g.destination('A', 'up2')
488        'B'
489        >>> g.destination('B', 'down')
490        'A'
491        >>> g.destination('A', 'nonexistent')
492        Traceback (most recent call last):
493        ...
494        KeyError...
495        >>> g.destination('D', 'any')
496        Traceback (most recent call last):
497        ...
498        KeyError...
499        """
500        return self._byEdge[source][edge]
501
502    def getDestination(
503        self,
504        source: Node,
505        edge: Edge,
506        default: Any = None
507    ) -> Optional[Node]:
508        """
509        Works like `destination`, but instead of raising a `KeyError` if
510        the node or edge is missing, it returns a default value (with a
511        default default of `None`).
512
513        ## Example
514
515        >>> g = UniqueExitsGraph()
516        >>> g.add_edges_from([
517        ...     ('A', 'B', 'up'),
518        ...     ('A', 'B', 'up2'),
519        ...     ('B', 'A', 'down'),
520        ...     ('B', 'B', 'self'),
521        ...     ('B', 'C', 'next'),
522        ...     ('C', 'B', 'prev')
523        ... ])
524        >>> g.getDestination('A', 'up')
525        'B'
526        >>> g.getDestination('A', 'up2')
527        'B'
528        >>> g.getDestination('B', 'down')
529        'A'
530        >>> g.getDestination('A', 'nonexistent') is None
531        True
532        >>> g.getDestination('A', 'nonexistent', 'default')
533        'default'
534        >>> g.getDestination('D', 'any') is None
535        True
536        """
537        return self._byEdge.get(source, {}).get(edge, default)
538
539    def allEdgesTo(
540        self,
541        destination: Node
542    ) -> List[Tuple[Node, Edge]]:
543        """
544        Searches the entire graph for edges whose destinations are the
545        specified destination, and returns a list of (node, edge) pairs
546        indicating the source node and edge name for each of those edges.
547        Self-edges are included in this list.
548
549        ## Example
550
551        >>> g = UniqueExitsGraph()
552        >>> g.add_edges_from([
553        ...     ('A', 'B', 'up'),
554        ...     ('A', 'B', 'up2'),
555        ...     ('B', 'A', 'down'),
556        ...     ('B', 'B', 'self'),
557        ...     ('B', 'C', 'next'),
558        ...     ('C', 'B', 'prev')
559        ... ])
560        >>> g.allEdgesTo('A')
561        [('B', 'down')]
562        >>> g.allEdgesTo('B')
563        [('A', 'up'), ('A', 'up2'), ('B', 'self'), ('C', 'prev')]
564        >>> g.allEdgesTo('C')
565        [('B', 'next')]
566        >>> g.allEdgesTo('D')
567        []
568        """
569        results = []
570        for node in self:
571            fromThere = self[node]
572            toHere = fromThere.get(destination, {})
573            for edgeKey in toHere:
574                results.append((node, edgeKey))
575
576        return results
577
578    def textMapObj(
579        self,
580        edgeSep: str = '::',
581        external: Optional[Set[Node]] = None,
582        explorationOrder: Optional[Tuple[Node, Sequence[Edge]]] = None,
583        edgeOrders: Union[
584            Dict[Node, Sequence[Edge]],
585            Dict[Node, Dict[Edge, Any]],
586            None
587        ] = None
588    ):
589        """
590        Returns a special object which is JSON-serializable and which
591        when serialized creates a semi-human-usable text-format map of
592        the graph.
593
594        The object consists of nested dictionaries, one per node, where
595        keys are node name + edge name strings (combined using the
596        `edgeSep` argument, default is '::'). The value for each key is
597        one of:
598
599        1. Another dictionary representing the node that edge leads
600            to, which can in turn have dictionary values...
601        2. A string naming a destination node that's already represented
602            elsewhere (or naming the current node for self-edges).
603
604        Any node present in the specified `external` set will be linked
605        to instead of listed out, even if it exists in the graph. The
606        `external` set **will be modified** by this function to include
607        all visited nodes in the graph.
608
609        If an `explorationOrder` is provided, it must be a tuple
610        specifying a start node followed by a sequence of edges that
611        indicates the path taken, and the edges will be visited
612        according to that order (this only matters in Python 3.7+ where
613        dictionaries have consistent order). A `ValueError` will be
614        raised if an invalid exploration order is provided. The path
615        list will be ignored if `edgeOrders` is provided explicitly.
616
617        TODO: What about unexplorable graphs (allow node names in place
618        of edge names in exploration order?)?!?
619
620        If `edgeOrders` is provided directly, it will override the
621        path part of the `explorationOrder` to determine the ordering of
622        edges at each node. If not and `explorationOrder` is provided, it
623        will be deduced from the `explorationOrder`. If neither is
624        present, ordering will follow whatever natural order is in the
625        graph, which in most cases should be order-of-creation.
626
627        Notes:
628        - For the format to avoid ambiguity, the `edgeSep` value must be
629            a string which does not appear in any node or edge names.
630        - Nodes and edge values will be converted to strings to build the
631            map.
632        - Node and edge properties are not represented in the resulting
633            object.
634        - For a variety of reasons, the result cannot be converted back
635            to a graph object. This is not intended for use as a JSON
636            serialization route (see the `networkx.readwrite.json_graph`
637            module for some built-in options).
638        - To get a string representation, one could do:
639            `json.dumps(graph.textMapObj())`
640
641        ## Examples
642
643        >>> from exploration import graphs as eg
644        >>> import json
645        >>> g = eg.UniqueExitsGraph()
646        >>> g.add_edges_from([
647        ...     ('A', 'B', 'up'),
648        ...     ('A', 'B', 'up2'),
649        ...     ('B', 'A', 'down'),
650        ...     ('B', 'B', 'self'),
651        ...     ('B', 'C', 'next'),
652        ...     ('C', 'B', 'prev')
653        ... ])
654        >>> print(json.dumps(g.textMapObj(), indent=2))
655        {
656          "A::up": {
657            "B::down": "A",
658            "B::self": "B",
659            "B::next": {
660              "C::prev": "B"
661            }
662          },
663          "A::up2": "B"
664        }
665        """
666        # We use `external` as our visited set
667        if external is None:
668            external = set()
669
670        if explorationOrder is not None:
671            here, path = explorationOrder
672        else:
673            # Find first non-external node as our starting node
674            for here in self.nodes:
675                if here not in external:
676                    break
677
678            # Path is empty
679            path = []
680
681        # Determine edge ordering for each node from exploration order
682        # or by natural ordering if no explorationOrder is available
683        if edgeOrders is None:
684            edgeOrders = cast(
685                Dict[Node, Dict[Edge, Any]],
686                {}
687            )
688            current = here
689            for i in range(len(path)):
690                edge = path[i]
691                # Add this edge next in the ordering for this node
692                orderHere: Dict[Edge, Any] = edgeOrders.setdefault(current, {})
693                # Note: we use a dictionary here because dictionaries do
694                # preserve insertion ordering (3.7+) and we need to both
695                # keep things in order AND do a bunch of lookups to
696                # avoid duplicates.
697                if edge not in orderHere:
698                    orderHere[edge] = True
699
700                # Move to next node
701                if edge not in self._byEdge[current]:
702                    raise ValueError(
703                        f"Invalid edge in exploration order path: at"
704                        f" step {i} we reached node {current} and were"
705                        f" supposed to take edge {edge} but that edge"
706                        f" does not exist."
707                    )
708                current = self._byEdge[current][edge]
709
710            # Add any unexplored nodes and/or edges in natural order
711            for node in self.nodes:
712                orderHere = edgeOrders.setdefault(node, {})
713                for edge in self._byEdge[node]:
714                    if edge not in orderHere:
715                        orderHere[edge] = True
716
717        result = {}
718        external.add(here)
719        # Now loop through keys of this node
720        for key in edgeOrders[here]:
721            combined = str(here) + edgeSep + str(key)
722            dest = self._byEdge[here][key]
723            if dest in external:
724                # links, including self-links
725                result[combined] = str(dest)
726            else:
727                # Recurse
728                result[combined] = self.textMapObj(
729                    edgeSep,
730                    external,
731                    (dest, []),  # empty path since we have edgeOrders
732                    edgeOrders
733                )
734
735        return result