LINQ API Client

class linq.client.Linq

The Automata LINQ API client.

Use this object to validate, plan and otherwise interact with workflows.

get_api_version() → str

Get Workflow Builder API version or unknown if it’s not possible to retrieve version.

Returns:: API version.

get_organizations() → list[Organization]

Get the list of organizations the user is a member of.

Returns:: the list of organizations
Raises:: APIError – if the request fails

create_workflow(workflow: Workflow) → WorkflowInfo

Create a new workflow.

Parameters:: workflow – the workflow definition
Returns:: the workflow information
Raises:: APIError – if creating the workflow fails

get_workflows(page: int = 1, page_size: int = 100) → list[WorkflowInfo]

Get a list of workflows.

Parameters:

page – The page number to retrieve.
page_size – The number of items per page.

Returns:

a list of all workflow information

Raises:

APIError – if the request fails

update_workflow(workflow_id: UUID, workflow: Workflow) → WorkflowInfo

Update an existing workflow.

Parameters:

workflow_id – the workflow ID to update
workflow – the workflow definition to update to

Returns:

the updated workflow information

Raises:

APIError – if updating the workflow fails

delete_workflow(workflow_id: UUID) → None

Delete a workflow.

Parameters:: workflow_id – the workflow ID
Returns:: None
Raises:: APIError – if deleting the workflow fails

validate_workflow(workflow: Workflow, validate_for_execution: bool = False, validate_for_infeasibility: bool = False, parameter_values: list[Annotated[BooleanParameterValueInput | IntegerParameterValueInput | FloatParameterValueInput | StringParameterValueInput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = []) → WorkflowValidationResult

Validate a workflow by submitting it to the API.

Parameters:

workflow – the workflow to be validated
validate_for_execution – set to True for deeper validation of actions/configuration set to False (default) to ignore hardware/driver-specific issues early during workflow development

Returns:

the validation result for the workflow

Raises:

APIError – if the validation request fails

plan_workflow(workflow_id: str, parameter_values: list[Annotated[BooleanParameterValueInput | IntegerParameterValueInput | FloatParameterValueInput | StringParameterValueInput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = []) → WorkflowPlanStatus

Plan a workflow by submitting it to the API.

Parameters:

workflow_id – the ID of the workflow to be planned
parameter_values – the parameter values to use for planning - essential if batching with a parameter

Returns:

the plan status for the workflow

Raises:

APIError – if the planning request fails

get_plan_status(workflow_id: str, plan_id: str) → WorkflowPlanStatus

Get the status of a planned workflow execution.

Parameters:

workflow_id – the ID of the workflow
plan_id – the ID of the plan

Returns:

the status of the plan for a workflow

Raises:

APIError – if the plan status request fails

get_plan_result(workflow_id: str, plan_id: str) → WorkflowPlanResult

Get the result of a planned workflow execution.

Parameters:

workflow_id – the ID of the workflow
plan_id – the ID of the plan

Returns:

the result of the plan for a workflow

Raises:

APIError – if the plan result request fails

get_supported_scheduler_versions() → dict[str, list[str]]

Returns all supported scheduler versions as a dict of versions per scheduler. The versions are sorted with the latest version first.

{<scheduler>: [latest_version, previous_version,…, oldest_version]} :return: a dict of supported schedulers and their versions :raises APIError: if the request fails

get_workflow(id: UUID) → WorkflowConfig

Get a workflow by ID.

Parameters:: id – the ID of the workflow
Returns:: the workflow
Raises:: APIError – if the get workflow request fails

get_sdk_workflow(id: UUID) → Workflow

Get a workflow by ID, as an SDK Workflow object.

Parameters:: id – the ID of the workflow
Returns:: the workflow
Raises:: APIError – if the get workflow request fails

get_all_drivers(scheduler: str, version: str) → list[Driver]

Get all drivers for a given scheduler and version.

Parameters:

scheduler – the scheduler name
version – the scheduler version

Returns:

a list of all the drivers for the given scheduler and version

Raises:

APIError – if the request fails

get_workcells(workspace_id: UUID) → list[Workcell]

Get a list of workcells.

Returns:: a list of all workcells
Raises:: APIError – if the request fails

deploy_workflow(*, workflow_id: UUID, device_id: UUID, reason: str = '[CODE_REASON_NOT_REQUIRED]', file_path: Path | None = None) → None

Deploy workflow with a given ID to a device (workcell).

Parameters:

workflow_id – the ID of the workflow to deploy
device_id – the ID of the device to deploy the workflow to
reason – the reason for deploying the workflow (optional)
file_path – the path to the parameter CSV file (optional)

Returns:

None

Raises:

APIError – if the deployment fails

get_status(*, device_id: str) → DeviceState

Get the status of the given workcell.

Parameters:: device_id – the ID of the device to get the status of
Returns:: the status of the workcell
Raises:: APIError – if the status request fails

get_latest_error(*, device_id_or_serial: UUID | str) → DeviceError | None

Get the latest error of a given workcell.

Parameters:: device_id – the ID of the device to get the status of
Returns:: the error status of the workcell
Raises:: APIError – if the error request fails

respond_to_error(*, device_id: UUID, error_id: UUID, action: DeviceErrorAvailableActions) → None: Respond to an error on the given workcell.

start_workflow(*, device_id: UUID, times_to_run: int = 1, reason: str = '[CODE_REASON_NOT_REQUIRED]') → None

Start the workflow currently deployed to the given device (workcell).

Parameters:

device_id – the ID of the device to start the workflow on
times_to_run – the number of times the workflow should run (default 1)
reason – the reason for starting the workflow (optional)

Returns:

None

Raises:

APIError – if the start request fails

pause_workflow(*, device_id: UUID, reason: str = '[CODE_REASON_NOT_REQUIRED]') → None

Pause the workflow currently running on the given device (workcell).

Parameters:

device_id – the ID of the device to pause the workflow on
reason – the reason for pausing the workflow (optional)

Returns:

None

Raises:

APIError – if the pause request fails

resume_workflow(*, device_id: UUID, reason: str = '[CODE_REASON_NOT_REQUIRED]') → None: Resume the workflow currently paused on the given device (workcell).

stop_workflow(*, device_id: UUID, reason: str = '[CODE_REASON_NOT_REQUIRED]') → None

Stop the workflow currently deployed to the given device (workcell).

param device_id: the ID of the device to stop the workflow on :param reason: the reason for stopping the workflow (optional) :return: None :raises APIError: if the stop request fails

reset_workflow(*, device_id: UUID, reason: str = '[CODE_REASON_NOT_REQUIRED]') → None: Reset the workflow currently deployed to the given device (workcell).

publish_workflow(*, workflow_id: UUID, simulate_drivers: bool = False, simulate_time: bool = False, reason: str = '[CODE_REASON_NOT_REQUIRED]', parameter_values: list[Annotated[BooleanParameterValueInput | IntegerParameterValueInput | FloatParameterValueInput | StringParameterValueInput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = []) → PublishedWorkflow: Publish the workflow currently deployed to the given device (workcell).

get_scripts_info(version: str | None = None) → OrgScriptsInfo: Retrieve the information about the available scripts for the users organization.

list_scripts_versions() → list[str]: List all available scripts versions.

list_machine_credentials() → list[MachineCredentials]

List all machine credentials.

Returns:: a list of all machine credentials
Raises:: APIError – if the request fails

create_machine_credentials(*, name: str) → MachineCredentialsWithSecret

Create a new machine credentials with a given name.

Parameters:: name – the name of the machine credentials to create
Returns:: the machine credentials with client secret
Raises:: APIError – if creating the machine credentials fails

delete_machine_credentials(*, id: str) → None

Delete a machine credentials with a given ID.

Parameters:: id – the ID of the machine credentials to delete
Returns:: None
Raises:: APIError – if deleting the machine credentials fails

rotate_machine_credentials_secret(*, id: str) → MachineCredentialsWithSecret

Rotate the secret of a machine credentias with a given ID.

Parameters:: id – the ID of the machine credentials to rotate the secret for
Returns:: the machine credentials with the new secret
Raises:: APIError – if rotating the secret fails

get_workflow_puml(workflow: Workflow, parameter_values: list[Annotated[BooleanParameterValueInput | IntegerParameterValueInput | FloatParameterValueInput | StringParameterValueInput, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = []) → tuple[str, str]

Get the PlantUML representation of a workflow before & after interpretation.

Parameters:: workflow – the workflow to be validated
Returns:: (before, after) PlantUML representations of the workflow
Raises:: APIError – if the request fails