Utils

class vibe_core.utils.MermaidVerticesMap(sources, sinks, tasks)

Bases: object

A map of vertices for a mermaid diagram extracted from a WorkflowSpec.

Each entry maps the source/sink/task name to the vertex label.

sinks: Dict[str, str]

Sink map.

sources: Dict[str, str]

Source map.

tasks: Dict[str, str]

Task map.

vibe_core.utils.build_mermaid_edge(origin, destination, vertices_origin, vertices_destination)

Builds a mermaid edge from a pair of vertices.

Parameters:

• origin (Tuple[str, str]) – A pair of source/sink/task and port names.

• destination (Tuple[str, str]) – A pair of source/sink/task and port names.

• vertices_origin (Dict[str, str]) – The vertex map to retrieve the mermaid vertex label for the origin.

• vertices_destination (Dict[str, str]) – The vertex map to retrieve the mermaid vertex label for the destination.

Returns:

str – The mermaid edge string.

vibe_core.utils.draw_mermaid_diagram(vertices, edges)

Draws a mermaid diagram from a set of vertices and edges.

Parameters:

• vertices (MermaidVerticesMap) – A map of vertices for a mermaid diagram extracted from a WorkflowSpec.

• edges (List[str]) – A list of edges already formated with mermaid syntax.

Returns:

str – The mermaid diagram string.

vibe_core.utils.ensure_list(input)

Ensures that the given input is a list.

If the input is a single item, it is wrapped in a list.

Parameters:

input (Union[List[TypeVar(T)], TypeVar(T)]) – List or single item to be wrapped in a list.

Returns:

List[TypeVar(T)] – A list containing the input item.

vibe_core.utils.format_double_escaped(s)

Encodes and decodes a double escaped input string.

Useful for formatting status/reason strings of VibeWorkflowRun.

Parameters:

s (str) – Input string to be processed.

Returns:

Formatted string.

vibe_core.utils.get_input_ids(input)

Retrieve the IDs from an input OpIOType object.

This method will extract the IDs from an OpIOType object and return them as a dictionary, where the keys are the names of the inputs and values are either strings or lists of strings.

Parameters:

input (Dict[str, Union[List[Dict[str, Any]], Dict[str, Any]]]) – The input object.

Returns:

Dict[str, Union[str, List[str]]] – A dictionary with the IDs of the input object.

vibe_core.utils.rename_keys(x, key_dict)

Renames the keys of a dictionary.

This utility function takes a dictionary x and a dictionary key_dict mapping old keys to their new names, and returns a copy of x with the keys renamed.

Parameters:

• x (Dict[str, Any]) – The dictionary with the keys to be renamed.

• key_dict (Dict[str, str]) – Dictionary mapping old keys to their new names.

Returns:

A copy of x with the keys renamed.