TierkreisGraph

TierkreisGraph#

class tierkreis.core.tierkreis_graph.TierkreisGraph(name: str = '')[source]#

Bases: object

Python in-memory representation of a Tierkreis graph. Can be loaded from and written to protobuf format. Supports building graphs programmatically.

Methods

`add_box`	Add a box node by specifying the inner graph and the inputs to connect to the node.
`add_const`	Add a constant value to the graph.
`add_edge`	Add an edge to the graph, connecting output port to input port.
`add_func`	Add a function call to the graph, connecting it to the specified inputs.
`add_match`	Add a Match node to the graph.
`add_node`	Add a node to the graph, connecting it to the specified inputs.
`add_tag`	Add a tag node to the graph.
`annotate_input`	Annotate an input port of the graph with a type.
`annotate_output`	Annotate an output port of the graph with a type.
`copy_value`	Copy a value to two output ports.
`discard`	Add a discard node, discarding the value at the specified output port.
`edges`	Iterator over all edges in the graph.
`from_proto`	Load from protobuf message.
`get_edge`	Retrieve an edge from the graph by source and target ports.
`in_edges`	Iterator over incoming edges to a node.
`inline_boxes`	Inline boxes by inserting the graphs they contain in to the parent graph.
`inputs`	A list of all output ports of the input node (the inputs to the graph).
`insert_graph`	Given a graph and wires to connect to that graph's inputs, inline that graph into <self>, and return the outputs of the inserted graph.
`make_pair`	Utility function to create a pair from two values.
`make_vec`	Utility function to create a vector from an iterator of incoming wires.
`nodes`	Iterator over all nodes in the graph.
`out_edge_from_port`	If there is an edge at port, return it, else None.
`out_edges`	Iterator over outgoing edges from a node.
`outputs`	A list of all input ports of the output node (the outputs of the graph).
`remove_edge`	Remove an edge from the graph.
`remove_nodes`	Remove nodes from the graph by index.
`set_outputs`	Set outputs of the graph by keyword arguments mapping output port names to incoming wires.
`to_proto`	Build protobuf message from graph.
`to_python`
`unpack_pair`	Utility function to unpack a pair into two values.
`vec_last_n_elems`	Utility to pop n_elements from a vector and return the corresponding output ports.

Attributes

`input`	Get a reference to the input node of the graph.
`input_node_idx`
`n_nodes`	The number of nodes in the graph.
`output`	Get a reference to the output node of the graph.
`output_node_idx`

exception MissingEdge(source: tierkreis.core.tierkreis_graph.NodePort, target: tierkreis.core.tierkreis_graph.NodePort)[source]#

Bases: Exception

with_traceback()#: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

add_box(graph: TierkreisGraph, /, **kwargs: NodePort | NodeRef) → NodeRef[source]#: Add a box node by specifying the inner graph and the inputs to connect to the node.

add_const(value: Any) → NodeRef[source]#: Add a constant value to the graph. Non-tierkreis values are converted if possible.

add_edge(source: NodePort | NodeRef, node_port_to: NodePort, edge_type: Type | TierkreisType | None = None) → TierkreisEdge[source]#

Add an edge to the graph, connecting output port to input port.

Parameters:

source – Source of the edge, either an output port or a node with a single output.
node_port_to – Target input port.
edge_type – Optional type annotation, defaults to None

Raises:

MismatchedGraphs – If source and target nodes are not in the same graph.
ValueError – If there is already an edge from source to target.

Returns:

Reference to the added edge.

Return type:

TierkreisEdge

add_func(_tk_function: str | FunctionName, _retry_secs: int | None = None, /, **kwargs: NodePort | NodeRef) → NodeRef[source]#

Add a function call to the graph, connecting it to the specified inputs.

Parameters:

_tk_function – Name of the function to call.
_retry_secs – Optionally specify how long to wait before retrying, defaults to None

Returns:

Reference to the added node.

Return type:

NodeRef

add_match(variant_value: NodePort | NodeRef, /, **variant_handlers: NodePort | NodeRef) → NodeRef[source]#: Add a Match node to the graph. The variant value is the only positional argument, handlers are specified as keyword arguments.

add_node(_tk_node: TierkreisNode, /, **incoming_wires: NodePort | NodeRef) → NodeRef[source]#

Add a node to the graph, connecting it to the specified inputs. The node to add is the only positional argument, the remaining keyword arguments specify the input port and the sources to connect to those ports.

Parameters:: _tk_node – The node to add to the graph.
Returns:: A reference to the added node.
Return type:: NodeRef

add_tag(tag: str, *, value: NodePort | NodeRef) → NodeRef[source]#: Add a tag node to the graph. Tag the value with the specified tag string.

annotate_input(input_port: str, edge_type: Type | TierkreisType | None)[source]#: Annotate an input port of the graph with a type.

annotate_output(output_port: str, edge_type: Type | TierkreisType | None)[source]#: Annotate an output port of the graph with a type.

copy_value(value: NodePort | NodeRef) → Tuple[NodePort, NodePort][source]#: Copy a value to two output ports.

discard(out_port: NodePort) → None[source]#: Add a discard node, discarding the value at the specified output port.

edges() → Iterator[TierkreisEdge][source]#: Iterator over all edges in the graph.

classmethod from_proto(pg_graph: Graph) → TierkreisGraph[source]#: Load from protobuf message.

get_edge(source: NodePort, target: NodePort) → TierkreisEdge[source]#: Retrieve an edge from the graph by source and target ports. Raises MissingEdge if the edge does not exist.

in_edges(node: NodeRef | int) → Iterator[TierkreisEdge][source]#: Iterator over incoming edges to a node.

inline_boxes(recursive=False) → TierkreisGraph[source]#

Inline boxes by inserting the graphs they contain in to the parent graph. Optionally do this recursively.

Returns:: Inlined graph
Return type:: TierkreisGraph

property input: NodeRef#: Get a reference to the input node of the graph.

inputs() → List[str][source]#: A list of all output ports of the input node (the inputs to the graph).

insert_graph(graph: TierkreisGraph, /, **kwargs: NodePort | NodeRef) → Dict[str, NodePort | NodeRef][source]#: Given a graph and wires to connect to that graph’s inputs, inline that graph into <self>, and return the outputs of the inserted graph.

make_pair(first_port: NodePort | NodeRef, second_port: NodePort | NodeRef) → NodePort[source]#: Utility function to create a pair from two values.

make_vec(element_ports: Iterable[NodePort | NodeRef]) → NodePort[source]#: Utility function to create a vector from an iterator of incoming wires.

property n_nodes: int#: The number of nodes in the graph.

nodes() → Iterator[TierkreisNode][source]#: Iterator over all nodes in the graph.

out_edge_from_port(source: NodePort) → TierkreisEdge | None[source]#: If there is an edge at port, return it, else None.

out_edges(node: NodeRef | int) → Iterator[TierkreisEdge][source]#: Iterator over outgoing edges from a node.

property output: NodeRef#: Get a reference to the output node of the graph.

outputs() → List[str][source]#: A list of all input ports of the output node (the outputs of the graph).

remove_edge(edge: TierkreisEdge)[source]#: Remove an edge from the graph.

remove_nodes(nodes: Iterable[int])[source]#: Remove nodes from the graph by index.

set_outputs(**kwargs: NodePort | NodeRef) → None[source]#: Set outputs of the graph by keyword arguments mapping output port names to incoming wires.

to_proto() → Graph[source]#: Build protobuf message from graph.

unpack_pair(pair_port: NodePort | NodeRef) → Tuple[NodePort, NodePort][source]#: Utility function to unpack a pair into two values.

vec_last_n_elems(vec_port: NodePort, n_elements: int) → List[NodePort][source]#: Utility to pop n_elements from a vector and return the corresponding output ports.

TierkreisGraph

Contents

TierkreisGraph#