`collection.dag`¶

Implementation of Direct Acyclic Graphs.

Module Contents¶

Classes¶

`DAGIterator`
`DAG`	Represent a Directed Acyclic Graph.

Attributes¶

VertexID

collection.dag.VertexID¶

exception collection.dag.DAGError(message: str | list[str], origin: str | None = None)¶

Bases: e3.error.E3Error

Exception raised by functions defined in E3.

class collection.dag.DAGIterator(dag: DAG, enable_busy_state: bool = False)¶

NOT_VISITED = 0¶

BUSY = 1¶

VISITED = 2¶

__iter__() → DAGIterator¶

__next__() → tuple[None, None] | tuple[VertexID, Any]¶

Retrieve next_element with with_predecessors=False.

The intermediate function is needed in Python 3.x

next_element() → tuple[VertexID | None, Any, frozenset[VertexID]]¶

Retrieve next element in topological order.

Returns:: a vertex id, data, predecessors. (None, None, frozenset()) is returned if no element is available).

leave(vertex_id: VertexID) → None¶

Switch element from BUSY to VISITED state.

Parameters:: vertex_id – the vertex to leave

class collection.dag.DAG¶

Represent a Directed Acyclic Graph.

Variables:

vertex_data (dict) – a dictionary containing all vertex data indexed by vertex id
tags – a dictionary containing “tags” associated with a vertex data, indexed by vertex id

property vertex_predecessors: dict[VertexID, frozenset[VertexID]]¶

Return predecessors.

Meant only for backward compatibility. Use vertex_predecessors_items.

Returns:: a dictionary containing the list of predecessors for each vertex, indexed by vertex id

reset_caches() → None¶

Reset caches for DAG properties (cycle and cached topological order).

This is a mandatory step when changing DAG edges.

vertex_predecessors_items() → collections.abc.Iterator[tuple[VertexID, frozenset[VertexID]]]¶

Return predecessors.

Returns:: a list of (vertex id, predecessors)

get_predecessors(vertex_id: VertexID) → frozenset[VertexID]¶: Get set of predecessors for a given vertex.

set_predecessors(vertex_id: VertexID, predecessors: frozenset[VertexID]) → None¶

Set predecessors for a given vertex.

Invalidate the global dictionary of vertex successors.

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.

add_tag(vertex_id: VertexID, data: Any) → None¶

Tag a vertex.

Parameters:

vertex_id – ID of the vertex to tag
data – tag content

get_tag(vertex_id: VertexID) → Any¶

Retrieve a tag associated with a vertex.

Parameters:: vertex_id – ID of the vertex
Returns:: tag content

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, <tag A>) get_context(E) will return (0, E, <tag E>) get_context(F) will return (1, C, <tag C>)

When using reverse_order=True, get_context will follow successors instead of predecessors.

get_context(B, reverse_order=True) will return (1, E, <tag E>)

Parameters:

vertex_id – ID of the vertex
max_distance – do not return resultsh having a distance higher than max_distance
max_element – return only up-to max_element elements
reverse_order – when True follow successors instead of predecessors

Returns:

a list of tuple (distance:int, tagged vertex id, tag content)

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.

Parameters:

vertex_id – the name of the vertex
data – data for the vertex.
predecessors – list of predecessors (vertex ids) or None

Raise:

DAGError if cycle is detected or else vertex already exist

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.

Parameters:

vertex_id – the name of the vertex
data – data for the vertex. If None and vertex already exist then data value is preserved
predecessors – list of predecessors (vertex ids) or None. If vertex already exists predecessors are added to the original predecessors
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

shortest_path(source: VertexID, target: VertexID) → list[VertexID] | None¶

Compute the shortest path between two vertices of the DAG.

Parameters:

source – vertex id of the source
target – vertex id of the target. If target is equal to source then the algorithm try to find the shortest cycle

Returns:

a list of vertex. If there is no path between two nodes then return None

check() → None¶

Check for cycles and inexisting nodes.

Raise:: DAGError if the DAG is not valid

get_closure(vertex_id: VertexID) → set[VertexID]¶

Retrieve closure of predecessors for a vertex.

Parameters:: vertex_id – the vertex to inspect
Returns:: a set of vertex_id

reverse_graph(enable_checks: bool = True) → DAG¶

Compute the reverse DAG.

Parameters:: enable_checks – whether to check that “self” is valid (no cycle)
Returns:: the reverse DAG (edge inverted)

__iter__() → collections.abc.Iterator[tuple[VertexID, Any]]¶

__contains__(vertex_id: VertexID) → bool¶: Check if a vertex is present in the DAG.

__getitem__(vertex_id: VertexID) → Any¶: Get data associated with a vertex.

__or__(other: DAG) → DAG¶: Merge two dags.

as_dot() → str¶

Return a Graphviz graph representation of the graph.

Returns:: the dot source file

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.

Parameters:: name_key – the data key-value to use as the name of the vertex; if None, the vertex ID is chosen as the name
Returns:: a text tree

prune(fun: collections.abc.Callable[[DAG, VertexID], bool], preserve_context: bool = True) → DAG¶

Create a pruned graph.

Parameters:

fun – function that return True whenever a node should be pruned. The function receive as parameter the dag and the node id
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.

Returns:

a new DAG

__len__() → int¶

__str__() → str¶: Return str(self).

collection.dag¶

Module Contents¶

Classes¶

Attributes¶

`collection.dag`¶