:py:mod:`collection.dag` ======================== .. py:module:: collection.dag .. autoapi-nested-parse:: Implementation of Direct Acyclic Graphs. Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: collection.dag.DAGIterator collection.dag.DAG Attributes ~~~~~~~~~~ .. autoapisummary:: collection.dag.VertexID .. py:data:: VertexID .. py:exception:: DAGError(message: str | list[str], origin: str | None = None) Bases: :py:obj:`e3.error.E3Error` Exception raised for DAG operations errors. .. py:class:: DAGIterator(dag: DAG, enable_busy_state: bool = False) Iterator for traversing DAG vertices in topological order. .. py:attribute:: NOT_VISITED :value: 0 .. py:attribute:: BUSY :value: 1 .. py:attribute:: VISITED :value: 2 .. py:method:: __iter__() -> DAGIterator Return iterator for DAG traversal. .. py:method:: __next__() -> tuple[None, None] | tuple[VertexID, Any] Retrieve next_element with with_predecessors=False. The intermediate function is needed in Python 3.x .. py:method:: next_element() -> tuple[VertexID | None, Any, frozenset[VertexID]] Retrieve next element in topological order. :return: a vertex id, data, predecessors. (None, None, frozenset()) is returned if no element is available). .. py:method:: leave(vertex_id: VertexID) -> None Switch element from BUSY to VISITED state. :param vertex_id: the vertex to leave .. py:class:: DAG Represent a Directed Acyclic Graph. :ivar vertex_data: a dictionary containing all vertex data indexed by vertex id :vartype vertex_data: dict :ivar tags: a dictionary containing "tags" associated with a vertex data, indexed by vertex id .. py:property:: vertex_predecessors :type: dict[VertexID, frozenset[VertexID]] Return predecessors. Meant only for backward compatibility. Use vertex_predecessors_items. :return: a dictionary containing the list of predecessors for each vertex, indexed by vertex id .. py:method:: reset_caches() -> None Reset caches for DAG properties (cycle and cached topological order). This is a mandatory step when changing DAG edges. .. py:method:: vertex_predecessors_items() -> collections.abc.Iterator[tuple[VertexID, frozenset[VertexID]]] Return predecessors. :return: a list of (vertex id, predecessors) .. py:method:: get_predecessors(vertex_id: VertexID) -> frozenset[VertexID] Get set of predecessors for a given vertex. :param vertex_id: vertex identifier .. py:method:: set_predecessors(vertex_id: VertexID, predecessors: frozenset[VertexID]) -> None Set predecessors for a given vertex. Invalidate the global dictionary of vertex successors. :param vertex_id: vertex identifier :param predecessors: set of predecessor vertex identifiers .. py:method:: get_successors(vertex_id: VertexID) -> frozenset[VertexID] Get set of successors for a given vertex. If the global dictionary of vertex successors has not been computed or if it has been invalidated then recompute it. :param vertex_id: vertex identifier .. py:method:: add_tag(vertex_id: VertexID, data: Any) -> None Tag a vertex. :param vertex_id: ID of the vertex to tag :param data: tag content .. py:method:: get_tag(vertex_id: VertexID) -> Any Retrieve a tag associated with a vertex. :param vertex_id: ID of the vertex :return: tag content .. py:method:: get_context(vertex_id: VertexID, max_distance: int | None = None, max_element: int | None = None, reverse_order: bool = False) -> list[tuple[int, VertexID, Any]] Get tag context. Returns the list of predecessors tags along with their vertex id and the distance between the given vertex and the tag. On each predecessors branch the first tag in returned. So for the following graph:: A* / \ B C* / \ \ D E* F where each node with a * are tagged get_context(D) will return (2, A, ) get_context(E) will return (0, E, ) get_context(F) will return (1, C, ) When using reverse_order=True, get_context will follow successors instead of predecessors. get_context(B, reverse_order=True) will return (1, E, ) :param vertex_id: ID of the vertex :param max_distance: do not return resultsh having a distance higher than ``max_distance`` :param max_element: return only up-to ``max_element`` elements :param reverse_order: when True follow successors instead of predecessors :return: a list of tuple (distance:int, tagged vertex id, tag content) .. py:method:: add_vertex(vertex_id: VertexID, data: Any = None, predecessors: collections.abc.Sequence[VertexID] | None = None) -> None Add a new vertex into the DAG. Note that this function always checks that there is no cycle in the current DAG. If you prefer to insert all your node first and then do the check at the end use update_vertex calls followed by a call to check. :param vertex_id: the name of the vertex :param data: data for the vertex. :param predecessors: list of predecessors (vertex ids) or None :raise: DAGError if cycle is detected or else vertex already exist .. py:method:: update_vertex(vertex_id: VertexID, data: Any = None, predecessors: collections.abc.Sequence[VertexID] | set[VertexID] | frozenset[VertexID] | None = None, enable_checks: bool = True) -> None Update a vertex into the DAG. :param vertex_id: the name of the vertex :param data: data for the vertex. If None and vertex already exist then data value is preserved :param predecessors: list of predecessors (vertex ids) or None. If vertex already exists predecessors are added to the original predecessors :param enable_checks: if False check that all predecessors exists and that no cycle is introduce is not perform (for performance) :raise: DAGError if cycle is detected .. py:method:: shortest_path(source: VertexID, target: VertexID) -> list[VertexID] | None Compute the shortest path between two vertices of the DAG. :param source: vertex id of the source :param target: vertex id of the target. If target is equal to source then the algorithm try to find the shortest cycle :return: a list of vertex. If there is no path between two nodes then return None .. py:method:: check() -> None Check for cycles and inexisting nodes. :raise: DAGError if the DAG is not valid .. py:method:: get_closure(vertex_id: VertexID) -> set[VertexID] Retrieve closure of predecessors for a vertex. :param vertex_id: the vertex to inspect :return: a set of vertex_id .. py:method:: reverse_graph(enable_checks: bool = True) -> DAG Compute the reverse DAG. :param enable_checks: whether to check that "self" is valid (no cycle) :return: the reverse DAG (edge inverted) .. py:method:: __iter__() -> collections.abc.Iterator[tuple[VertexID, Any]] Return iterator for DAG vertices in topological order. .. py:method:: __contains__(vertex_id: VertexID) -> bool Check if a vertex is present in the DAG. :param vertex_id: vertex identifier .. py:method:: __getitem__(vertex_id: VertexID) -> Any Get data associated with a vertex. :param vertex_id: vertex identifier .. py:method:: __or__(other: DAG) -> DAG Merge two dags. :param other: another DAG to merge with this one .. py:method:: as_dot() -> str Return a Graphviz graph representation of the graph. :return: the dot source file .. py:method:: as_tree(name_key: str | None = None) -> str Return a tree representation of the graph. This is similar to the output of the `tree` bash command. :param name_key: the data key-value to use as the name of the vertex; if None, the vertex ID is chosen as the name :return: a text tree .. py:method:: prune(fun: collections.abc.Callable[[DAG, VertexID], bool], preserve_context: bool = True) -> DAG Create a pruned graph. :param fun: function that return True whenever a node should be pruned. The function receive as parameter the dag and the node id :param preserve_context: if True ensure that context is preserved (i.e: that calls to get_context will return the same value for both current graph and pruned graph). This means that any attempt to remove a node containing a tag will result in a DAGError. :return: a new DAG .. py:method:: __len__() -> int Return the number of vertices in the DAG. .. py:method:: __str__() -> str Return string representation of DAG with vertices and their predecessors.