AFL.double_agent.AgentDriver module#

class AFL.double_agent.AgentDriver.DoubleAgentDriver(name: str = 'DoubleAgentDriver', overrides: Dict[str, Any] | None = None)#

Bases: AgentWebAppMixin, Driver

Persistent Config#

save_path: str

path to directory where data will be serialized to

append(db_uuid: str, concat_dim: str) → None#
Parameters:

  • db_uuid (str) – Dropbox UUID to retrieve xarray.Dataset from. The Dataset should be deposited using Client.deposit_obj in interactive mode in order to obtain the uuid of the deposited item. See example below

  • concat_dim (str) – xarray dimension in input dataset to concatenate to

defaults = {'pipeline': {}, 'save_path': '/home/AFL/', 'tiled_input_groups': []}#
initialize_pipeline(db_uuid: str = None, pipeline: List[Dict[str, Any]] = None, name: str = 'Pipeline') → None#

Set the double_agent.Pipeline to outline

Parameters:

  • db_uuid (str, optional) – Dropbox UUID to retrieve the double_agent.Pipeline from. The Dataset should be deposited using Client.deposit_obj in interactive mode in order to obtain the uuid of the deposited item. Either this or pipeline must be provided.

  • pipeline (List[Dict[str, Any]], optional) – List of pipeline operations in JSON format to construct the pipeline from. Either this or db_uuid must be provided.

  • name (str, default="Pipeline") – Name for the pipeline when constructing from operations list.

Example

Using dropbox: ```python from AFL.automation.APIServer.Client import Client from AFL.double_agent import *

client = Client(‘localhost’,port=5053) client.login(‘User’) db_uuid = client.deposit_obj(obj=Pipeline(),interactive=True)[‘return_val’] client.enqueue(task_name=’initialize_pipeline’,db_uuid=db_uuid) ```

Using operations list: `python ops = [{'class': 'AFL.double_agent.SomeOp', 'args': {...}}] client.enqueue(task_name='initialize_pipeline', pipeline=ops) `

predict(deposit: bool = False, save_to_disk: bool = False, sample_uuid: str | None = None, AL_campaign_name: str | None = None) → str#

Evaluate the pipeline set with .initialize_pipeline.

Parameters:

  • deposit (bool) – If True, the xarray.Dataset resulting from the Pipeline calculation will be placed in this APIServers dropbox for retrieval and the db_uuid will be returned.

  • save_to_disk (bool) – If True, the xarray.Dataset resulting from the Pipeline calculation will be serialized to disk in NetCDF format.

  • sample_uuid (Optional[str]) – Optionally provide a sample uuid to tag the calculation with

  • AL_campaign_name – Optionally provide an AL campaign name to tag the calculation with

status()#
class AFL.double_agent.AgentDriver.Driver(*args, **kwargs)#

Bases: object

gather_defaults()#
static queued(*args, **kwargs)#
static unqueued(*args, **kwargs)#
AFL.double_agent.AgentDriver.build_pipeline_from_json(ops_json: str, name: str = 'Pipeline') → Dict[str, Any]#

Helper that accepts a JSON string of operations.

AFL.double_agent.AgentDriver.build_pipeline_from_ops(ops: List[Dict[str, Any]], name: str = 'Pipeline') → Dict[str, Any]#

Create a pipeline from a list of operation JSON dictionaries.

AFL.double_agent.AgentDriver.get_pipeline_ops(strict: bool = False) → Dict[str, Any]#

Return metadata describing available pipeline operations with cache metadata.

AFL.double_agent.AgentDriver.mpl_plot_to_bytes(*args, **kwargs)#
AFL.double_agent.AgentDriver.xarray_to_bytes(*args, **kwargs)#