Cloud Python API

Last modified on April 24, 2024

To start using the API

1. Install the AnyLogic Cloud client library using the pip package installer:

   pip install https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0-py3-none-any.whl
   

   Verify that the installation of the package has been completed successfully:

   $ python
   >>> import anylogiccloudclient.client.cloud_client

   If no errors appear, you can use the API to work with the cloud-based models.
2. Obtain the API key in the AnyLogic Cloud UI.
3. Create a new instance of the CloudClient object:

   For public version of AnyLogic Cloud:

   from anylogiccloudclient.client.cloud_client import CloudClient
   client = CloudClient( <the API key> )

   For Private Cloud:

   from anylogiccloudclient.client.cloud_client import CloudClient
   client = CloudClient( <the API key>, <the Cloud instance URL> )

4. Use the API to work with the cloud-based models.

You can also install the package manually. To download the package, use the following link:

https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0.tar.gz

In case of the manual installation, make sure the value of the PYTHONPATH variable specifies the location of the AnyLogic Cloud client library before using it.

AnyLogic Python API is synchronous (see Synchronous and asynchronous API).

Another thing to keep in mind is that the API treats names (for example, model, experiment, input, and output names) as case-insensitive.

The CloudClient class is responsible for authentication and communication with the AnyLogic Cloud. Typically, there is only one object of class CloudClient in your Python code.

An object of Inputs class is constructed in preparation of a model run (of any kind) by calling the CloudClient functions: create_default_inputs() or create_inputs_from_experiment(). It contains full information about the model, model version, and the input values. It should not be confused with the inputs field in the Version object.

Inputs of distribution type (for Monte Carlo 2^nd order experiments) are coming in future releases of AnyLogic Cloud API.

An object of this class is returned after the call of the get_outputs() or get_outputs_and_run_if_absent() functions of a ModelRun constructed for a single run simulation experiment.

An object of this class is returned after the call of the get_outputs() or get_outputs_and_run_if_absent() functions of a ModelRun constructed for a parameter variation or other multi run experiment. It simplifies navigation within the complex outputs structure.

The ModelRun class is responsible for communication with and control of a model run (without animation) executed in Cloud. It can be considered as a front-end mirror of a back-end experiment run. Objects of the ModelRun class are created and returned by calling the functions create_simulation(), create_parameter_variation(), create_monte_carlo(), or create_parameter_variation_with_replications() of the CloudClient. A ModelRun object contains full information about the model, version, inputs, and experiment type.

Simulation run without animation (minimalistic)

This is done with the API function ModelRun.get_outputs_and_run_if_absent().

from anylogiccloudclient.client.cloud_client import CloudClient

client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")
inputs = client.create_inputs_from_experiment(version, "Baseline")
inputs.set_input("Server capacity", 8)
simulation = client.create_simulation(inputs)
outputs = simulation.get_outputs_and_run_if_absent()

print("Raw outputs = " + str(outputs.get_raw_outputs()))
print("For Server Capacity = " + str(inputs.get_input("Server capacity")))
print("Mean queue size = " + str(outputs.value("Mean queue size|Mean queue size")))
print("Server utilization = " + str(outputs.value("Utilization|Server utilization")))

Python is considered a back-end programming language, that is why the inputs and the outputs are accessible through the console only:

1. The CloudClient object is created, given the API key.
2. The server is asked to find the latest version of the model with the name "Service System Demo".
3. When (and if) such model and version is found, we create the Inputs object with the default input values.
4. We use the set_input() function to change the value of the "Server capacity" parameter to 8.
5. We ask the CloudClient to create a simulation object with the inputs. This is done on the client side, there is no communication with the server.
6. The getOutputsAndRunIfAbsent() function is called, which checks the simulation status:
   • If such simulation has been completed, it gets the outputs
   • If such simulation has not been run yet, it runs the simulation and waits for the outputs. The SingleRunOutputs object is returned
7. Finally, the obtained output values and the corresponding input are displayed in the console using the API of SingleRunOutputs and Inputs.

If an error occurs during any step, the error message will be displayed in the console.

To get the value of a particular output, we need to specify its name exactly as it is constructed when the model is uploaded to Cloud (for more information refer to the Output object section. The case does not matter: you can use either lower case or upper case.

Querying simulation results of a completed run

The key function used here is ModelRun.get_outputs().

from anylogiccloudclient.client.cloud_client import CloudClient

client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
model = client.get_model_by_name("Service System Demo")
version = client.get_latest_model_version(model)
inputs = client.create_default_inputs(version)
inputs.set_input("Server capacity", 8)
simulation = client.create_simulation(inputs)
outputs = simulation.get_outputs()

print("Mean queue size: " + str(outputs.value("Mean queue size|Mean queue size")))

We use the set_input() function to change the value of the "Server capacity" parameter to 8. Having constructed the inputs and the simulation objects, we call simulation.get_outputs(). If the outputs exist, they are returned. If there are no outputs, an error occurs.

Simulation run with progress polling

In this example we use the function ModelRun.get_progress() to obtain the progress of the running simulation. We also show a different way of running the simulation and getting results via the sequence of run(), wait_for_completion(), and get_outputs() function calls.

import threading
import time

from anylogiccloudclient.client.cloud_client import CloudClient

def _update_progress(s):
    while s.get_status() == "FRESH" or s.get_status() == "RUNNING":
        print("Progress: " + str(s.get_progress()["total"]) + "%")
        time.sleep(5)

client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")
inputs = client.create_default_inputs(version)
inputs.set_input("Server capacity", 21)
inputs.set_input("{STOP_TIME}", 10000000)
simulation = client.create_simulation(inputs)

simulation.run()

progress_thread = threading.Thread(target=_update_progress, args=(simulation,))
progress_thread.start()

simulation.wait_for_completion()
outputs = simulation.get_outputs()

print("Mean queue size: " + str(outputs.value("Mean queue size|Mean queue size")))

During the standard construction of the inputs, we change the system input {STOP_TIME} to a bigger value to be able to observe the actual progress by making the simulation run longer. Such input exists in every model uploaded to Cloud. However, if a run with the same stop time and other inputs has been performed already, the simulation won’t run, and the outputs will be delivered almost instantly. So, to watch the progress moving, consider setting other values for the inputs.

We use the progress_thread.start() function, which is called after simulation.run(). It initiates a repeated call of s.get_progress(), which returns information about the progress. For a simple simulation run, we are only interested in the total field. For multi-run experiments it contains more details. The function time.sleep() is called once the simulation is completed. It waits for additional 5 seconds to make sure that the final progress value has been received.

The call of simulation.run() initiates the simulation run (only if such run has not been done before) and returns the same simulation object. It does not wait for the simulation to complete. To wait for the experiment to complete, we use the function simulation.wait_for_completion(), which also returns the same object. It is only now that we query the simulation results by calling simulation.get_outputs(). This sequence is here solely for demo purposes; it could be replaced by a single call of simulation.get_outputs_and_run_if_absent().

Running a custom workflow

This example implements a custom workflow where some simulations are run in parallel and some — sequentially, based on the results of other simulations. This scenario may be useful for those who want to set up, for example, a custom optimization algorithm in AnyLogic Cloud.

from anylogiccloudclient.client.cloud_client import CloudClient

capacities = [4, 9]
client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")

inputs = [client.create_default_inputs(version) for num in range(2)]
[inputs[num].set_input("Server capacity", capacities[num]) for num in range(2)]

simulations = [client.create_simulation(i) for i in inputs]
[s.run() for s in simulations]
[s.wait_for_completion() for s in simulations]

outputs = [s.get_outputs() for s in simulations]
for o in outputs:
    print("Mean queue size: " + str(o.value("Mean queue size|Mean queue size")))

In this scenario, we first perform two simulation runs in parallel with the Server Capacity parameter set to 4 in one run and 9 in another by the range() function. The two runs are initiated by calling s.run().

The outputs of simulation runs performed in parallel are returned also as an array in the same order.

When you run multiple simulations in parallel manually, like in this example, each call of s.wait_for_completion() performs its own independent polling, therefore many parallel runs may result in high HTTP traffic. If that is suspected, consider changing the optional parameter polling_period of these functions.

Running parameter variation

This example shows how to run a parameter variation experiment. One of the input parameters will be a discrete range. To demonstrate one more feature of the AnyLogic Cloud API, we will take the input values from an existing simulation experiment defined in the standard AnyLogic Cloud web interface and change a parameter value from a scalar to a range.

from anylogiccloudclient.client.cloud_client import CloudClient

client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")
inputs = client.create_inputs_from_experiment(version, "Baseline")
inputs.set_range_input("Mean service time", 1.8, 2, 0.1)
variation = client.create_parameter_variation(inputs)
outputs = variation.get_outputs_and_run_if_absent()

print("Mean queue size: " + str(outputs.get_values_of_output("Mean queue size|Mean queue size")))

To reuse the set of inputs from an experiment defined in the AnyLogic Cloud web interface we use the function CloudClient.create_inputs_from_experiment() instead of create_default_inputs(). Having copied the inputs of the "Baseline" experiment,  we change the input "Mean service time" to a range type (from 1.8 to 2.0 with step 0.1) by calling Inputs.set_range_input(). Therefore, three simulation runs are to be performed.

Then we need to create a ModelRun object of parameter variation type, which is done by the CloudClient.create_parameter_variation() function.

The run of parameter variation is invoked by calling the get_outputs_and_run_if_absent() function just like in our previous examples, but there is one important difference. Full outputs of a multiple run experiment may be a very large piece of data, so the API user has to explicitly specify which outputs need to be delivered. This is done by listing the output names in the array passed to get_outputs_and_run_if_absent() or get_outputs() as a parameter. In this example, we are interested in the "Mean queue size|Mean queue size" output. If the parameter is omitted for a multi run experiment, only scalar outputs will be returned, if any.

Outputs of a multi run experiment are returned as a MultiRunOutputs object, which has a number of functions simplifying navigation.

Running parameter variation with replications

In this example, we will run a parameter variation experiment with the explicitly set number of replications. One of the input parameters will be a discrete range. We will also use the input values from an existing simulation experiment defined in the standard AnyLogic Cloud web interface.

from anylogiccloudclient.client.cloud_client import CloudClient

client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")
inputs = client.create_inputs_from_experiment(version, "Baseline")
inputs.set_range_input("Mean service time", 1.8, 2, 0.1)
inputs.set_number_of_replications(4)

variation_with_replications = client.create_parameter_variation_with_replications(inputs)
outputs = variation_with_replications.get_outputs_and_run_if_absent(["Total time in system|Total time in system"])

print("Parameter variation with replications experiment completed")
invalues = outputs.get_values_of_input("Mean service time")
outvalues = outputs.get_values_of_output("Total time in system|Total time in system")
for i in range(len(invalues)):
    print("When Mean service time = " + str(invalues[i]) + ", mean Total time in system = " + str(outvalues[i]["statistics"]["mean"]))

As you can see, this experiment is basically identical to the one demonstrated in the parameter variation example. The main difference is that we use the set_number_of_replications() function to specify the number of replications explicitly.

Then we create a ModelRun object for the needed experiment type, which is done by the CloudClient.create_parameter_variation_with_replications() function.

The experiment run is triggered by calling the get_outputs_and_run_if_absent() function just like in our previous examples, but there is one important difference. Full outputs of a multiple run experiment may be a very large piece of data, so the API user has to explicitly specify which outputs need to be delivered. This is done by listing the output names in the array passed to get_outputs_and_run_if_absent() or get_outputs() as a parameter. In this example, we are interested in the "Total time in system | Total time in system" output. If the parameter is omitted for a multi run experiment, only scalar outputs will be returned, if any.

Outputs of a multi run experiment are returned as a MultiRunOutputs object, which has a number of functions simplifying navigation.

Running Monte Carlo 1^st order

In this example, we will run a Monte Carlo 1^st order experiment with the explicitly set number of replications. When executed via the API, the Monte Carlo experiment runs the simulation a specified number of times, obtains the collections of values, and then returns them.

One of the input parameters will be a discrete range. We will also use the input values from an existing simulation experiment defined in the standard AnyLogic Cloud web interface.

from anylogiccloudclient.client.cloud_client import CloudClient
        
client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20")
version = client.get_latest_model_version("Service System Demo")
inputs = client.create_default_inputs(version)
inputs.set_number_of_replications(4)
monte_carlo = client.create_monte_carlo(inputs)
outputs = monte_carlo.get_outputs_and_run_if_absent(["Utilization|Server utilization"])
print("Monte Carlo 1st order experiment completed")
print("Server utilization distribution: " + str(outputs.get_values_of_output("Utilization|Server utilization")))

We create the default inputs for the experiment already defined in the AnyLogic Cloud GUI with the create_default_inputs() function. The number of replications is set with the set_number_of_replications() function.

Then we create a ModelRun object for the needed experiment type, which is done by the CloudClient.create_monte_carlo() function.

The experiment run is triggered by calling the get_outputs_and_run_if_absent() function just like in our previous examples, but there is one important difference. Full outputs of a multiple run experiment may be a very large piece of data, so the API user has to explicitly specify which outputs need to be delivered. This is done by listing the output names in the array passed to get_outputs_and_run_if_absent() or get_outputs() as a parameter. In this example, we are interested in the "Utilization|Server utilization" output. If the parameter is omitted for a multi run experiment, only scalar outputs will be returned, if any.

Outputs of a multi run experiment are returned as a MultiRunOutputs object, which has a number of functions simplifying navigation.

Using a file as an input and output

In this example, we will run a simple model that requires an input file. The model will write new values to this file. Then, we will download the file (by then it will contain new values) as output.

from anylogiccloudclient.client.cloud_client import CloudClient
 
def main():
 
    print("Authorizing and uploading input file data.xlsx ...")
    client = CloudClient("e05a6efa-ea5f-4adf-b090-ae0ca7d16c20", "https://cloud.anylogic.com")
    input_file_hash = client.upload_file("data.xlsx")
 
    print("Setting up simulation File IO API Demo ...")
    version = client.get_latest_model_version("File IO API Demo")
    inputs = client.create_default_inputs(version)
    inputs.set_input("Data", input_file_hash)
 
    print( "Running simulation...")
    simulation = client.create_simulation(inputs)
    outputs = simulation.get_outputs_and_run_if_absent()
 
    print( "Downloading data.xlsx file populated with output data...")
    output_file = outputs.value("Data")
    client.download_file(output_file, "output_" + output_file.name)
 
if __name__ == '__main__':
    main()

First, we calculate the hash code from the file that is located in the file system on the machine we use to execute the experiment via Python API, using the upload_file() function. Then, we specify the model (which in this case already is present in Cloud) and tell Cloud the file to use as input, using the calculated hash code.

When the experiment is completed, we convert the resulting data to the output file. After that, the file can simply be downloaded with the download_file() function.

How can we improve this article?

Send