Skip to content

Creating a custom API workstation

Note

Please note that this page shows you how to create the same workstation as the NEXUS Workstation page, except that it will use API calls (e.g. via the SDK) instead of the NEXUS file exchange module.

To implement Atinary SDLabs on your own robots or processes, you can create your custom workstation objects. A workstation executes tasks provided by the optimizer.

Pre-requisites

Real Robot Example

Let us assume that our physical workstation is a robot that runs the following catalyzed chemical reaction: A+B -> C+D

Robot parameters

  • Fraction of A [0.0 - 1.0]
  • Fraction of B [0.0 - 1.0]
  • Catalyst type which can be (\(\alpha\), \(\beta\) or \(\gamma\))
  • Catalyst amount (1 to 10 catalyst units)

Robot measurements

  • Reaction yield (i.e. how much of desired product C is produced)
  • Reaction selectivity (i.e. how much of desired product C is produced relative to undesired by-product D)

Here, A and B are reactants and C is the desired product. D is a by-product. These should be modified to match your experimental setup.

Workstation in SDLabs

We will now create our robot according to the provided information. To define a workstation, we need to define metadata, parameters (instructions) and measurements.

Metadata

wst_name = 'My Chemistry Robot'
wst_description = 'Managed via API. Robot runs reaction A+B->C+D. It measures the reaction yield and selectivity.'
wst_bandwidth = 100  # Optional limit, stating that the Robot can run up to 100 reactions at a time

Parameters

We can have numerical (continuous or discrete) and categorical parameters. Let's create your workstation's parameters:

Continuous Parameters

Define the amounts of reactants A and B as fractions ranging from 0.0 to 1.0.

prm_a = sct.ParameterObj(
    name='fraction_a',
    description='Fraction of reactant A',
    type='continuous',
    high_value=1.0,
    low_value=0.0
)

prm_b = sct.ParameterObj(
    name='fraction_b',
    description='Fraction of reactant B',
    type='continuous',
    high_value=1.0,
    low_value=0.0
)

Categorical Parameter

Define the catalyst type categorical parameter. We can also record category properties such as the catalyst particle size, which is different for each category.

prm_cat_type = sct.ParameterObj(
    name='catalyst_type',
    description='Catalyst type',
    type='categorical',
    descriptors = [
        {
            'category': 'alpha',
            'properties': [{'key': 'particle size', 'value': 0.4}]
        },
        {
            'category': 'beta',
            'properties': [{'key': 'particle size', 'value': 0.76}]
        },
        {
            'category': 'gamma',
            'properties': [{'key': 'particle size', 'value': 1.23}]
        }
    ]
)

Warning

The properties attribute is an array of key-value pairs. The value field of each pair should always be numerical (as seen in the example). Properties are optional, but if they are defined for one category, they must be set for all the categories of that parameter.

Discrete Parameter

Define the amount of catalyst as a discrete parameter ranging from 0 to 10 in strides of 1 unit.

prm_cat_amt = sct.ParameterObj(
    name='catalyst_amount',
    description='Catalyst amount',
    type='discrete',
    high_value=10.0,
    low_value=0.0,
    stride=1.0
)

Create Parameters in SDLabs

Now that we have created the base parameters objects, we can save them using the ParameterApi.parameter_create() method, which will return the parameters with their id.

wst_params = [
    prm_api.parameter_create(parameter_obj=prm).object
    for prm in [prm_a, prm_b, prm_cat_type, prm_cat_amt]
]

Measurements

Measurements are modelled as an array of strings: 

wst_measurements = ['yield', 'selectivity']

Create Workstation in SDLabs

my_wst = wst_api.workstation_create(
    workstation_obj=sct.WorkstationObj(
        name=wst_name,
        description=wst_description,
        conn_type=sct.ConnectionType.API,
        bandwidth=wst_bandwidth,
        measurements=wst_measurements,
        parameters=[p.id for p in wst_params]
    )
).object

If a simple print(my_wst) contains a representation of your newly created workstation, it means that my_wst is ready to use!

Create and run a Template

  • You can now create a new template that uses this workstation in SDLabs and run a campaign from it.
  • Then, follow the next steps below:

Get suggested Parameters

Use the following call to poll SDLabs and check for the latest suggested parameters: 

from time import sleep

campaign_id = ''  # YOUR CURRENT CAMPAIGN ID HERE
SLEEP_TIME = 10  # in seconds
latest_params: sct.LatestObservationsObj = []

while not len(latest_params):
    time.sleep(SLEEP_TIME)
    latest_params = wst_api.latest_parameters(
        campaign_ids=[campaign_id]
    ).objects

The resulting object latest_params is now a list of 'LatestObservationsObj' with the following attributes (plus a few identifiers and metadata, like 'batch_number', 'iteration', 'campaign_id', 'workstation_level', etc.): 

LatestObservationsObj(
    id='rqt_bbcc0077bb4455bbdd662299eeaa3344',
    parameters=[
        ObservationParameterValueObj(
            id='prm_242ae3d2cc1aff441170cc0ed3f34aca',
            name='fraction_a',
            value='0.9321',  # Values are always stringified!
        ),
        ObservationParameterValueObj(
            id='prm_44a2e3d2cc1aff441177cc1eddff44cc',
            name='fraction_b',
            value='0.01209',  # Values are always stringified!
        ),
        ObservationParameterValueObj(
            id='prm_8844d62aa060e3e7155f1448831119001',
            name='catalyst_amount',
            value='4.0',  # Values are always stringified!
        ),
        ObservationParameterValueObj(
            id='prm_34a0ead3ce1aaf401007cc1ed4df33c3',
            name='catalyst_type',
            value='gamma',
        )
    ],
    measurements=[
        MeasurementValueObj(
            id=None,
            name='yield',
            value=''  # this value is an empty string: you need to fill with your experiment's actual values 
        ), 
        MeasurementValueObj(
            id=None,
            name='selectivity',
            value=''  # this value is an empty string: you need to fill with your experiment's actual values 
        )
    ],
    notes='(optional) Write here any comment you have on this specific experiment.',
)

Run your Experiment

Once you have your parameters from the platform, run the experiment on 'My Chemistry Robot' and get the corresponding measurements.

Return the experiment's Measurements

First, fill the value field of each 'MeasurementValueObj' object (here the yield and selectivity measured by your robot) with the corresponding value from your experiment. 

# `latest_obs` is of type LatestObservationsObj
for latest_obs in latest_params:
    latest_obs.notes = f"itr {latest_obs.iteration}, bat {latest_obs.batch_number}, lvl {latest_obs.workstation_level}"

    # `measure` is of type ObservationValueObj
    for measure in latest_obs.measurements:
        # ADD HERE your actual experiment's value (mind the expected STRING type!)
        measure.value = str(random.random())

Modify input parameters

If your experiment actually deviated from the suggested parameters (e.g. you couldn't reach the suggested catalyst_amount of 4.0, so you only experimented with 2.0), then you may change the value of the parameter in the value field of the corresponding 'ObservationParameterValueObj' parameter. Please ignore the value_raw fields that you might have noticed in the 'ObservationParameterValueObj' objects (it should always be None).

Then, you can return your experiment's data as follows:

# This call expects a List of LatestObservationsObj
response = wst_api.latest_measurements(
    latest_params
).objects

Where the returned response is a List of 'LatestMeasurementsResponseObj' objects, each containing: * the corresponding request id * the request status (OK or Error) * an event in case of error (with datetime, error type and short description).

For instance, you can parse the response as follows: 

for resp_result in response:
    if resp_result.status == sct.ObservationStatus.OK:
        print(f"Request {resp_result.id} successfully submitted.")
    else:
        print(f"Request {resp_result.id} {resp_result.status} with {resp_result.event.type}: {resp_result.event.message} ")

Important

Please note that successfully submitted does not necessarily mean successfully processed: e.g. your observation may land out of the parameter space if you changed the parameter value, and you will only be able to see it while fetching your campaign latest status and events...

You can repeat the process with your next iteration until your campaign is terminated!