Monitor

Classes and functions definition for monitoring workflow runs.

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()

Print 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]

Return 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) – If True, detailed information about task progress will be included in the output (defaults to False).

time_or_now(time)

Convert a given datetime object to the client’s timezone.

If no datetime object is provided, the current time is used.

Parameters:

time (datetime | None) – Datetime object to convert to the client’s timezone.

Returns:

The datetime object converted to the client’s timezone.

Return type:

datetime

update_run_status(monitored_runs, monitored_warnings)

Update the monitor table with the latest runs status.

Parameters:

  • monitored_runs (List[MonitoredWorkflowRun]) – List of workflow runs to monitor. If only one run is provided, the method will monitor that run directly.

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

vibe_core.monitor.format_typing(type_dict)

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

The formatted dictionary.

Return type:

Dict[str, str]

vibe_core.monitor.strftimedelta(start, end)

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

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

Return type:

str