Data Model

Data model classes definition used throughout FarmVibes.AI.

class vibe_core.datamodel.Message(message, id=None, location=None)

Bases: object

Dataclass that represents an API message.

id: Optional[str] = None: The id of the message.

location: Optional[str] = None: The location of the message.

message: str: The message.

class vibe_core.datamodel.MetricsDict

Bases: TypedDict

Type definition for metrics dictionary.

cpu_usage: float: The current system-wide CPU utilization as a percentage.

disk_free: Optional[int]: The amount of free disk space in bytes.

free_mem: int: The amount of free memory in bytes.

load_avg: Tuple[float, ...]

Average system load.

The number of processes in the system run queue averaged over the last 1, 5, and 15 minutes respectively as a tuple.

total_mem: int: The total amount of memory in bytes.

used_mem: int: The amount of used memory in bytes.

class vibe_core.datamodel.MonitoredWorkflowRun(workflow, name, id, status, task_details)

Bases: object

Dataclass that represents the monitored workflow run information.

id: str: The id of the run.

name: str: The name of the run.

status: RunStatus: The status of the run.

task_details: Dict[str, RunDetails]: The details of the tasks of the run.

workflow: Union[str, Dict[str, Any]]: The workflow name or workflow dictionary definition of the run.

class vibe_core.datamodel.Region(name, geojson=<factory>)

Bases: object

Dataclass that represents a region.

geojson: Dict[str, Any]: The geojson of the region.

name: str: The name of the region.

class vibe_core.datamodel.RunBase(name, workflow, parameters)

Bases: object

Base dataclass for a run.

name: str: The name of the run.

parameters: Optional[Dict[str, Any]]: The parameters of the run.

workflow: Union[str, Dict[str, Any]]: The workflow of the run.

class vibe_core.datamodel.RunConfig(name, workflow, parameters, user_input, id, details, task_details, spatio_temporal_json, output='')

Bases: RunConfigInput

Dataclass that represents a run config.

set_output(value)

Set the output of the run config.

Parameters:: value (Dict[str, List[Dict[str, Any]] | Dict[str, Any]]) – The value to set the output to.

details: RunDetails: The details of the run config.

id: UUID: The id of the run config.

output: str = '': The output of the run.

spatio_temporal_json: Optional[SpatioTemporalJson]: The spatio temporal json of the run config.

task_details: Dict[str, RunDetails]: The details of the tasks of the run config.

class vibe_core.datamodel.RunConfigInput(name, workflow, parameters, user_input)

Bases: RunBase

Dataclass that represents a run config input.

user_input: Union[SpatioTemporalJson, Dict[str, Any], List[Any]]: The user input of the run config (usually a region/geometry and time range).

class vibe_core.datamodel.RunConfigUser(name, workflow, parameters, user_input, id, details, task_details, spatio_temporal_json, output='')

Bases: RunConfig

Dataclass that represents a run config for the user.

static finished(status)

Check if a run has finished.

This method checks if a given status is either vibe_core.datamodel.RunStatus.done, vibe_core.datamodel.RunStatus.cancelled, or vibe_core.datamodel.RunStatus.failed.

Parameters:: status (RunStatus) – The status to check.
Returns:: Whether the run has finished.

classmethod from_runconfig(run_config)

Create a RunConfigUser from a RunConfig.

Parameters:: run_config (RunConfig) – The run config to create the user run config from.
Returns:: The user run config.

class vibe_core.datamodel.RunDetails(start_time=None, submission_time=None, end_time=None, reason=None, status=RunStatus.pending, subtasks=None)

Bases: object

Dataclass that encapsulates the details of a run.

end_time: Optional[datetime] = None: The end time of the run.

reason: Optional[str] = None: A description of the reason for the status of the run.

start_time: Optional[datetime] = None: The start time of the run.

status: RunStatus = 'pending': The status of the run.

submission_time: Optional[datetime] = None: The submission time of the run.

subtasks: Optional[List[Any]] = None: Details about the subtasks of the run.

class vibe_core.datamodel.RunStatus(value)

Bases: StrEnum

Enum that represents the status of a run.

static finished(status)

Check if a run has finished.

This method checks if a run status is either vibe_core.datamodel.RunStatus.done, vibe_core.datamodel.RunStatus.cancelled, or vibe_core.datamodel.RunStatus.failed.

Parameters:: status (RunStatus) – The status to check.
Returns:: Whether the run has finished.

cancelled = 'cancelled': The run is cancelled.

deleted = 'deleted': The run has been deleted.

deleting = 'deleting': The run is being deleted.

done = 'done': The run is done.

failed = 'failed': The run has failed.

pending = 'pending': The run is pending

queued = 'queued': The run is queued.

running = 'running': The run is running.

class vibe_core.datamodel.SpatioTemporalJson(start_date, end_date, geojson)

Bases: object

Dataclass that represents a spatio temporal json.

end_date: datetime: The end date of the spatio temporal json.

geojson: Dict[str, Any]: The geojson of the spatio temporal json.

start_date: datetime: The start date of the spatio temporal json.

class vibe_core.datamodel.TaskDescription(inputs=<factory>, outputs=<factory>, parameters=<factory>, task_descriptions=<factory>, short_description='', long_description='')

Bases: object

Dataclass that represents a task description.

inputs: Dict[str, str]: The inputs of the task.

long_description: str = '': The long description of the task.

outputs: Dict[str, str]: The outputs of the task.

parameters: Dict[str, Union[str, Dict[str, str]]]: The task parameters.

short_description: str = '': The short description of the task.

task_descriptions: Dict[str, str]: The descriptions of subtasks.

class vibe_core.datamodel.WorkflowRun

Bases: ABC

An abstract base class for workflow runs.

abstract property output: Dict[str, BaseVibe | List[BaseVibe]]

Get the output of the workflow run.

Returns:: The output of the workflow run as a vibe_core.data.BaseVibeDict.
Raises:: NotImplementedError – If the method is not implemented by a subclass.

abstract property status: str

Get the status of the workflow run.

Returns:: The status of the workflow run as a string.
Raises:: NotImplementedError – If the method is not implemented by a subclass.

vibe_core.datamodel.decode(data)

Decode the given data using zlib and base64 encodings.

Parameters:: data (str) – The string to decode.
Returns:: The decoded string.
Return type:: str

vibe_core.datamodel.encode(data)

Encode a string using zlib and base64 encoding.

This function compresses the data string with zlib and then encodes it into a base64 string.

Parameters:: data (str) – The string to be encoded.
Returns:: The encoded string.
Return type:: str