Monitor

class vibe_core.monitor.VibeWorkflowDocumenter(name, sources, sinks, parameters, description)

Bases: object

Documenter class for VibeWorkflow objects.

This class implements the logic for printing/formatting information about workflows, including formatting the text elements and adding styling tags. It contains the methods to print out a description of the workflow with its sources, sinks and parameters.

Parameters:

  • name (str) – The name of the workflow.

  • sources (Dict[str, str]) – Dictionary with source names and types.

  • sinks (Dict[str, str]) – Dictionary with sink names and types.

  • parameters (Dict[str, Any]) – Dictionary with parameter names and default values.

  • description (TaskDescription) – A TaskDescription object containing the short and long description of the workflow.

print_documentation()

Prints the full documentation of the workflow.

This method prints the header of the documentation, the sources, the sinks, the parameters and the tasks provided in the parsed workflow yaml file.

property formatted_parameters: Dict[str, str]

Returns a dictionary of workflow’s parameters with their default values.

Returns:

A dictionary containing the formatted parameters and default values.

class vibe_core.monitor.VibeWorkflowRunMonitor(api_time_zone, detailed_task_info=False, multi_run=False)

Bases: object

Class that abstracts the formatting of workflow run status.

Parameters:

  • api_time_zone (tzfile) – The time zone of the API server.

  • detailed_task_info (bool, default: False) – If True, detailed information about task progress will be included in the output (defaults to False).

time_or_now(time)

Converts a given datetime object to the client’s timezone. If no datetime object is provided, the current time is used.

Parameters:

time (Optional[datetime]) – Datetime object to convert to the client’s timezone.

Returns:

datetime – The datetime object converted to the client’s timezone.

update_run_status(monitored_runs, monitored_warnings)

Updates the monitor table with the latest runs status.

Parameters:

  • runs – List of workflow runs to monitor. If only one run is provided, the method will monitor that run directly.

  • monitored_warnings (List[Union[str, Warning]]) – List of warnings to display in the table.

Return type:

None

vibe_core.monitor.format_typing(type_dict)

Formats the types in a type dictionary.

This function takes a dictionary with type strings and formats it, replacing the “typing.” prefix in the values with an empty string.

Parameters:

type_dict (Dict[str, str]) – The type dictionary to format.

Returns:

Dict[str, str] – The formatted dictionary.

vibe_core.monitor.strftimedelta(start, end)

Returns the time delta between two datetimes as a string in the format ‘HH:MM:SS’.

Parameters:

  • start (datetime) – Start datetime object.

  • end (datetime) – End datetime object.

Returns:

str – The timedelta formatted as a ‘HH:MM:SS’ string.