FarmVibes.AI Client

FarmVibes.AI client.

This module provides a client for the FarmVibes.AI service, which allows users to interact with the platform, such as inspecting workflow description and managing workflow runs (listing, creating, monitoring, cancelling, etc.)

class vibe_core.client.Client

Bases: ABC

An abstract base class for clients.

abstract list_workflows()

List all available workflows.

Returns:: A list of workflow names.
Raises:: NotImplementedError – If the method is not implemented by a subclass.
Return type:: List[str]

abstract run(workflow, geometry, time_range)

Run a workflow.

Parameters:

workflow (str) – The name of the workflow to run.
geometry (BaseGeometry) – The geometry to run the workflow on.
time_range (Tuple[datetime, datetime]) – The time range to run the workflow on.

Returns:

A WorkflowRun object.

Raises:

NotImplementedError – If the method is not implemented by a subclass.

Return type:

WorkflowRun

class vibe_core.client.ClusterType(value)

Bases: StrEnum

An enumeration of cluster types.

client()

Create a client based on the cluster type.

Returns:: A FarmvibesAiClient object based on the cluster type.

class vibe_core.client.FarmvibesAiClient(baseurl)

Bases: Client

A client for the FarmVibes.AI service.

Parameters:: baseurl (str) – The base URL of the FarmVibes.AI service.

_form_payload(workflow, parameters, geometry, time_range, input_data, run_name)

Form a payload dictionary for submitting a workflow run.

Parameters:

workflow (str | Dict[str, Any]) – The name of the workflow to run or a dict containing the workflow definition.
parameters (Dict[str, Any] | None) – A dict of optional parameters to pass to the workflow. The keys and values depend on the specific workflow definition.
geometry (BaseGeometry | None) – The geometry to use for the input data. It must be a valid shapely geometry object (e.g., Point or Polygon). It will be converted to GeoJSON format internally. Alternatively it can be None if input_data is provided instead.
time_range (Tuple[datetime, datetime] | None) – The time range to use for the input data. It must be a tuple of two datetime objects representing the start and end dates. Alternatively it can be None if input_data is provided instead.
input_data (Dict[str, T | List[T]] | List[T] | T | None) – The input data to use for the workflow run. It must be an instance of InputData or one of its subclasses (e.g., SpatioTemporalJson or SpatioTemporalRaster). Alternatively it can be None if geometry and time_range are provided instead.
run_name (str) – The name to assign to the workflow run.

Returns:

A dict containing the payload for submitting a workflow run. The keys are ‘run_name’, ‘workflow’, ‘parameters’, and ‘user_input’.

Return type:

Dict[str, Any]

Note

Either (geometry and time_range) or (input_data) must be provided. Providing both will result in an error.

cancel_run(run_id)

Cancel a workflow run.

This method sends a request to the FarmVibes.AI service to cancel a workflow run that is in progress or pending. If the cancellation is successful, the workflow run status will be set to ‘cancelled’.

Parameters:: run_id (str) – The ID of the workflow run to cancel.
Returns:: The message from the FarmVibes.AI service indicating whether the cancellation was successful or not.
Return type:: str

Note

The cancellation may take some time to take effect depending on the state of the workflow run and the service availability.

Warning

A workflow run that is already completed or failed cannot be cancelled.

delete_run(run_id)

Delete a workflow run.

This method sends a request to the FarmVibes.AI service to delete a completed workflow run (i.e. a run with the a status of ‘done’, ‘failed’, or ‘cancelled’). If the deletion is successful, all cached data the workflow run produced that is not shared with other workflow runs will be deleted and status will be set to ‘deleted’.

Parameters:: run_id (str) – The ID of the workflow run to delete.
Returns:: The message from the FarmVibes.AI service indicating whether the deletion request was successful or not.
Return type:: str

Note

The deletion may take some time to take effect depending on the state of the workflow run and the service availability.

Warning

A workflow run that is in progress or pending cannot be deleted. A cancelled workflow run can be deleted. So, in order to delete workflow run that is in progress or pending, it first needs to be cancelled and then it can be deleted.

describe_run(run_id)

Describe a workflow run.

This method returns a RunConfigUser object containing the description of a workflow run, such as its name, status, inputs and outputs.

Parameters:: run_id (str) – The ID of the workflow run to describe.
Returns:: A RunConfigUser object containing the workflow run description.
Return type:: RunConfigUser

describe_workflow(workflow_name)

Describe a workflow.

This method returns a dictionary containing the description of a workflow, such as its inputs, outputs, parameters and tasks.

Parameters:: workflow_name (str) – The name of the workflow to describe. It must be one of the names returned by list_workflows().
Returns:: A dictionary containing the workflow description. The keys are ‘name’, ‘description’, ‘inputs’, ‘outputs’ and ‘parameters’.
Return type:: Dict[str, Any]

Note

The description is returned as a TaskDescription object, which is a dataclass that represents the structure and properties of a workflow.

document_workflow(workflow_name)

Print the documentation of a workflow.

This method prints a formatted documentation of a workflow, including its name, description, inputs, outputs and parameters.

Parameters:: workflow_name (str) – The name of the workflow to document.

Note

The documentation is printed to stdout and can be redirected to other outputs if needed.

get_api_time_zone()

Get the time zone of the FarmVibes.AI REST-API.

This method returns a tzfile object representing the time zone of the FarmVibes.AI REST-API. The time zone is determined by parsing the ‘date’ header from the response of a GET request to the base URL of the service. If the ‘date’ header is missing or invalid, a warning is issued and the client time zone is used instead.

Returns:: The time zone of the FarmVibes.AI REST-API as a tzfile object.
Return type:: tzfile

Note

The tzfile object is a subclass of datetime.tzinfo that represents a time zone using an Olson database file.

get_last_runs(n)

Get the last ‘n’ workflow runs.

This method returns a list of VibeWorkflowRun objects containing the details of the last n workflow runs.

Parameters:: n (int) – The number of workflow runs to get (with n>0).
Returns:: A list of VibeWorkflowRun objects.
Return type:: List[VibeWorkflowRun]

get_run_by_id(id)

Get a workflow run by ID.

This method returns a VibeWorkflowRun object containing the details of a workflow run by its ID.

Parameters:: id (str) – The ID of the workflow run to get.
Returns:: A VibeWorkflowRun object.
Return type:: VibeWorkflowRun

get_system_metrics()

Get system metrics from the FarmVibes.AI service.

This method returns a dictionary containing various system metrics, such as CPU usage, memory usage and disk space.

Returns:: A dictionary containing system metrics.
Return type:: MetricsDict

get_workflow_yaml(workflow_name)

Get the YAML definition of a workflow.

This method returns a string containing the YAML definition of a workflow. The YAML definition specifies the name and operations of the workflow in a human-readable format.

Parameters:: workflow_name (str) – The name of the workflow. It must be one of the names returned by list_workflows().
Returns:: The YAML definition of the workflow.
Return type:: str

list_runs(ids=None, fields=None)

List workflow runs on the FarmVibes.AI service.

Parameters:

ids (str | List[str] | None) – The IDs of the workflow runs to list. If None, all workflow runs will be listed.
fields (str | List[str] | None) – The fields to return for each workflow run. If None, all fields will be returned.

Returns:

A list of workflow runs. Each run is represented by a dictionary with keys corresponding to the requested fields and values containing the field values.

list_workflows()

List all available workflows on the FarmVibes.AI service.

This method returns a list of workflow names that can be used to submit workflow runs or to get more details about a specific workflow.

Returns:: A list of workflow names.
Return type:: List[str]

monitor(runs=1, refresh_time_s=1, refresh_warnings_time_min=5, timeout_min=None, detailed_task_info=False)

Monitor workflow runs.

This method will block and print the status of the runs each refresh_time_s seconds, until the workflow runs finish or it reaches timeout_min minutes. It will also print warnings every refresh_warnings_time_min minutes.

Parameters:

runs (List[VibeWorkflowRun] | VibeWorkflowRun | int) – A list of workflow runs, a single run object, or an integer. The method will monitor the provided workflow runs. If a list of runs is provided, the method will provide a summarized table with the status of each run. If only one run is provided, the method will monitor that run directly. If an integer > 0 is provided, the method will fetch the respective last runs and provide the summarized monitor table.
refresh_time_s (int) – Refresh interval in seconds (defaults to 1 second).
refresh_warnings_time_min (int) – Refresh interval in minutes for updating the warning messages (defaults to 5 minutes).
timeout_min (int | None) – The maximum time to monitor the workflow run, in minutes. If not provided, the method will monitor indefinitely.
detailed_task_info (bool) – If True, detailed information about task progress will be included in the output (defaults to False).

Raises:

ValueError – If no workflow runs are provided (empty list).

resubmit_run(run_id)

Resubmit a workflow run with the given run ID.

Parameters:: run_id (str) – The ID of the workflow run to resubmit.
Returns:: The resubmitted workflow run.
Return type:: VibeWorkflowRun

run(workflow, name, *, geometry=None, time_range=None, input_data=None, parameters=None)

Run a workflow.

This method instantiates a workflow run using the provided data and parameters.

Parameters:

workflow (str | Dict[str, Any]) – The name of the workflow to run or a dict containing the workflow definition.
name (str) – The name to assign to the workflow run.
geometry (BaseGeometry | None) – The geometry to use for the input data. It must be a valid shapely geometry object (e.g., Point or Polygon). It will be converted to GeoJSON format internally. Alternatively it can be None if input_data is provided instead.
time_range (Tuple[datetime, datetime] | None) – The time range to use for the input data. It must be a tuple of two datetime objects representing the start and end dates. Alternatively it can be None if input_data is provided instead.
input_data (Dict[str, T | List[T]] | List[T] | T | None) – The input data to use for the workflow run. It must be an instance of InputData or one of its subclasses (e.g., SpatioTemporalJson or SpatioTemporalRaster). Alternatively it can be None if geometry and time_range are provided instead.
parameters (Dict[str, Any] | None) – A dict of optional parameters to pass to the workflow. The keys and values depend on the specific workflow definition.

Returns:

A VibeWorkflowRun object.

Return type:

VibeWorkflowRun

Note

Either (geometry and time_range) or (input_data) must be provided. Providing both will result in an error.

verify_disk_space()

Verify that there is enough disk space available for the cache.

This method checks the system metrics returned by the FarmVibes.AI service and compares the disk free space with a predefined threshold. If the disk free space is below the threshold, a warning message is displayed to the user, suggesting to clear the cache.

Note

The disk free space threshold is defined by DISK_FREE_THRESHOLD_BYTES.

Raises:: RuntimeWarning –

default_headers: Dict[str, str] = {'Accept': 'application/json', 'Content-Type': 'application/json'}: The default headers to use for requests to the FarmVibes.AI service.

class vibe_core.client.VibeWorkflowRun(id, name, workflow, parameters, client)

Bases: WorkflowRun, MonitoredWorkflowRun

Represent a workflow run in FarmVibes.AI.

Parameters:

id (str) – The ID of the workflow run.
name (str) – The name of the workflow run.
workflow (str | Dict[str, Any]) – The name of the workflow associated to the run.
parameters (Dict[str, Any]) – The parameters associated to the workflow run, as a dict. The keys and values depend on the specific workflow definition.
client (FarmvibesAiClient) – An instance of the FarmVibesAiClient class.

block_until_cancelled(timeout_s=None)

Block until the workflow run is cancelled, with an optional timeout in seconds.

Parameters:: timeout_s (optional) – Timeout in seconds to wait for the workflow to be cancelled. If not provided, the method will wait indefinitely.
Returns:: The workflow run object.
Raises:: RuntimeError – If the run is not cancelled before timeout_s.
Return type:: VibeWorkflowRun

block_until_complete(timeout_s=None)

Block until the workflow run execution completes or fails.

Also allows defining a timeout in seconds.

Parameters:: timeout_s (optional) – Timeout in seconds to wait for the workflow to complete. If not provided, the method will wait indefinitely.
Returns:: The workflow run object.
Raises:: RuntimeError – If the run does not complete before timeout_s.
Return type:: VibeWorkflowRun

block_until_deleted(timeout_s=None)

Block until the workflow run is deleted, with an optional timeout in seconds.

Parameters:: timeout_s (optional) – Timeout in seconds to wait for the workflow to be deleted. If not provided, the method will wait indefinitely.
Returns:: The workflow run object.
Raises:: RuntimeError – If the run does not complete before timeout_s.
Return type:: VibeWorkflowRun

cancel()

Cancel the workflow run.

Returns:: The workflow run.
Return type:: VibeWorkflowRun

delete()

Delete the workflow run.

Returns:: The workflow run.
Return type:: VibeWorkflowRun

monitor(refresh_time_s=1, refresh_warnings_time_min=5, timeout_min=None, detailed_task_info=False)

Monitor the workflow run.

This method will call vibe_core.client.FarmvibesAiClient.monitor() to monitor the workflow run.

Parameters:

refresh_time_s (int) – Refresh interval in seconds (defaults to 1 second).
refresh_warnings_time_min (int) – Refresh interval in minutes for updating the warning messages (defaults to 5 minutes).
timeout_min (int | None) – The maximum time to monitor the workflow run, in minutes. If not provided, the method will monitor indefinitely.
detailed_task_info (bool) – If True, detailed information about task progress will be included in the output (defaults to False).

resubmit()

Resubmit the current workflow run.

Returns:: The resubmitted workflow run instance.
Return type:: VibeWorkflowRun

property output: Dict[str, BaseVibe | List[BaseVibe]] | None: Get the output of the workflow run.

property reason: str | None

Get the reason of the workflow run.

The reason is a string that describes the status of the workflow run. In case of failure, it also contains the reason of the failure.

property status: RunStatus: Get the status of the workflow run.

property task_details: Dict[str, RunDetails]: Get the task details of the workflow run.

property task_status: Dict[str, str]: Get the task status of the workflow run.

vibe_core.client.get_default_vibe_client(type='')

Get the default vibe client.

If given a cluster type, it will try to connect to a cluster of that type assuming the data files are present. Otherwise, it will try to connect to a remote cluster first, and if that fails, try to connect to the local one (if it exists).

Parameters:: type (str | ClusterType) – The type of the cluster (from ClusterType) to connect to.
Returns:: The vibe client.
Return type:: FarmvibesAiClient

vibe_core.client.get_local_service_url()

Retrieve the local service URL used to submit workflow runs to the FarmVibes.AI service.

This function attempts to read the service URL from a file, and if that fails, it will return a fallback URL.

Returns:: The local service URL.
Return type:: str

vibe_core.client.get_remote_service_url()

Get the remote service URL.

Returns:: The remote service URL.
Return type:: str

vibe_core.client.get_vibe_client(url)

Get a vibe client given an API base URL.

Parameters:: url (str) – The URL.
Returns:: The vibe client.
Return type:: FarmvibesAiClient

vibe_core.client.DISK_FREE_THRESHOLD_BYTES = 53687091200: Threshold for disk space in bytes.

vibe_core.client.FALLBACK_SERVICE_URL

Fallback URL for FarmVibes.AI service.

vibe_core.client.FARMVIBES_AI_REMOTE_SERVICE_URL_PATH

Path to the local remote service URL file.

vibe_core.client.FARMVIBES_AI_SERVICE_URL_PATH

Path to the local service URL file.

vibe_core.client.TASK_SORT_KEY = 'submission_time': Key for sorting tasks.

vibe_core.client.XDG_CONFIG_HOME

Path to configuration file for FarmVibes.AI service.