Utils

General utility functions used across FarmVibes.AI codebase.

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

Bases: object

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)

Build 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:

The mermaid edge string.

Return type:

str

vibe_core.utils.draw_mermaid_diagram(vertices, edges)

Draw 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:

The mermaid diagram string.

Return type:

str

vibe_core.utils.ensure_list(input)

Ensure that the given input is a list.

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

Parameters:

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

Returns:

A list containing the input item.

Return type:

List[T]

vibe_core.utils.format_double_escaped(s)

Encode and decode 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, List[Dict[str, Any]] | Dict[str, Any]]) – The input object.

Returns:

A dictionary with the IDs of the input object.

Return type:

Dict[str, str | List[str]]

vibe_core.utils.rename_keys(x, key_dict)

Rename 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.