Cloud Python API

Обновлено 11.11.2024

Чтобы использовать Python API, выполните следующие действия:

1. Установите клиентскую библиотеку AnyLogic Cloud с помощью установщика пакетов pip:

   pip install https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0-py3-none-any.whl
   Для Private Cloud 2.3.1 и предыдущих версий: pip install https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0-py3-none-any-old.whl

2. Убедитесь, что установка пакета успешно завершена:

   $ python
   >>> import anylogiccloudclient.client.cloud_client

   Если не возникло ошибок, вы можете использовать API для работы с моделями Cloud.

Вы также можете установить пакет вручную. Для загрузки пакета пройдите по ссылке:

https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0.tar.gz
Для Private Cloud 2.3.1 и предыдущих версий: https://cloud.anylogic.com/files/api-8.5.0/clients/anylogiccloudclient-8.5.0-old.tar.gz

При ручной установке убедитесь, что значение переменной PYTHONPATH указывает на местоположение клиентской библиотеки AnyLogic Cloud перед ее использованием.

AnyLogic Python API является синхронным (см. Синхронный и асинхронный API).

Напомним, что API обрабатывает имена (например, модели, эксперимента, входных и выходных данных) как нечувствительные к регистру.

Класс CloudClient отвечает за аутентификацию и взаимодействие с AnyLogic Cloud. Как правило, в коде Python может быть только один экземпляр класса CloudClient.

Объект класса Inputs создается при подготовке к запуску модели (любого типа) путем вызова функций CloudClient: create_default_inputs() или create_inputs_from_experiment(). Он содержит всю информацию о модели, версии модели и входных данных. Не путайте этот класс с полем inputs объекта Version.

Входные данные типа распределения (для экспериментов Монте-Карло второго порядка) появятся в будущих версиях AnyLogic Cloud API.

Объект этого класса возвращается после вызова функций get_outputs() или get_outputs_and_run_if_absent() объекта ModelRun, созданного для простого эксперимента с одним «прогоном».

Объект этого класса возвращается после вызова функций get_outputs() или get_outputs_and_run_if_absent() объекта ModelRun, созданного для эксперимента с варьированием параметров или другого эксперимента с несколькими прогонами. Это упрощает навигацию по сложной структуре выходных данных.

Класс ModelRun отвечает за обмен информацией во время прогона модели и контролирует этот прогон (без анимации, в Cloud). По сути, это внешний интерфейс внутренних процессов прогона эксперимента. Объекты класса ModelRun создаются и возвращаются путем вызова функций create_simulation(), create_parameter_variation(), create_monte_carlo() или create_parameter_variation_with_replications() класса CloudClient. Объект ModelRun содержит всю информацию о модели, версии, входных данных и типе эксперимента.

«Прогон» эксперимента без анимации (простой)

Используйте API функцию 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 считается бэкенд (back-end) языком программирования, поэтому входные и выходные данные доступны только через консоль:

1. С помощью ключа API создается объект CloudClient.
2. Сервер ищет последнюю версию модели с именем "Service System Demo".
3. Когда (и если) такая модель и версия найдены, мы создаем объект Inputs с входными данными по умолчанию.
4. Используем функцию set_input(), чтобы изменить значение параметра "Server capacity" на 8.
5. Далее запрашиваем CloudClient создать объект simulation с входными данными. Это делается на стороне клиента, коммуникации с сервером нет.
6. Вызывается функция getOutputsAndRunIfAbsent(), которая проверяет статус эксперимента:
   • Если этот эксперимент завершен, функция возвращает выходные данные
   • Если эксперимент еще не запускался, эта функция запускает его и ожидает получения выходных данных.
   В обоих случаях возвращается объект SingleRunOutputs
7. Наконец, полученные выходные значения и соответствующие входные данные отображаются в консоли с помощью API SingleRunOutputs и Inputs.

Если на каком-либо шаге произойдет ошибка, в консоли отобразится сообщение об ошибке.

Чтобы получить значение конкретных выходных данных, нужно указать имя этих данных именно так, как оно обозначено при загрузке модели в Cloud (подробнее см. в разделе об объекте Output. Регистр не учитывается: можно использовать и нижний, и верхний регистры.

Запрос результатов эксперимента завершенного «прогона»

Здесь используется ключевая функция 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")))

Мы используем функцию set_input(), чтобы изменить значение параметра “Server capacity” на 8. Создав объекты inputs и simulation, мы вызываем simulation.get_outputs(). Если выходные данные существуют, они возвращаются. Если их нет, возникает ошибка.

«Прогон» эксперимента с опросом о ходе выполнения

В этом примере мы используем функцию ModelRun.get_progress() для получения информации о ходе выполнения эксперимента. Существует также другой способ запуска эксперимента и получения результатов: последовательный вызов функций: run(), wait_for_completion() и get_outputs().

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")))

При стандартном создании входных данных мы увеличиваем значение системных входных данных {STOP_TIME}, чтобы иметь возможность следить за ходом выполнения эксперимента с увеличенной продолжительностью. Эти входные данные есть в каждой модели, загруженной в Cloud. Однако если прогон с таким же временем остановки и другими входными данными уже был выполнен, эксперимент не запустится, а выходные данные будут получены почти мгновенно. Таким образом, чтобы следить за ходом выполнения, попробуйте установить другие значения для входных данных.

Мы используем функцию progress_thread.start(), которая вызывается после simulation.run(). Она инициирует повторный вызов функции s.get_progress(), которая возвращает информацию о ходе выполнения. Для прогона простого эксперимента нас интересует только поле total. Для экспериментов с несколькими прогонами оно содержит больше деталей. Функция time.sleep() вызывается после завершения эксперимента. Она ждет еще 5 секунд, чтобы убедиться, что получено окончательное значение прогресса.

Вызов simulation.run() инициирует запуск эксперимента (только если эксперимент не запускался ранее) и возвращает тот же объект simulation. Она не ждет завершения эксперимента. Чтобы дождаться завершения эксперимента, мы используем функцию simulation.wait_for_completion(), которая также возвращает тот же объект. Только теперь мы запрашиваем результаты эксперимента, вызывая функцию simulation.get_outputs(). Эта последовательность здесь исключительно для демонстрационных целей. Ее можно заменить одним вызовом simulation.get_outputs_and_run_if_absent().

Запуск нестандартного потока операций

В этом примере показан нестандартный поток операций, в котором некоторые эксперименты выполняются параллельно, а некоторые — последовательно, на основе результатов других экспериментов. Этот сценарий может быть интересен тем, кто, например, планирует настроить алгоритм оптимизации в 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")))

В этом сценарии мы с помощью функции range() сначала выполняем параллельно два прогона эксперимента с параметром Server Capacity, равным 4 в одном прогоне и 9 в другом. Два запуска инициируются вызовом s.run().

Выходные данные параллельных прогонов эксперимента возвращаются в виде массива в том же порядке.

Когда вы параллельно запускаете несколько экспериментов вручную, как в этом примере, каждый вызов s.wait_for_completion() выполняет свой собственный независимый опрос. Много параллельных запусков могут привести к высокому HTTP-трафику. Если есть такие опасения, попробуйте изменить значение параметра polling_period этих функций.

Запуск эксперимента варьирования параметров

В этом примере показано, как запустить эксперимент варьирования параметров. Один из входных параметров будет варьироваться в дискретном диапазоне. Чтобы продемонстрировать еще одну функцию AnyLogic Cloud API, возьмем входные значения существующего простого эксперимента из AnyLogic Cloud и изменим значение параметра со скалярного на диапазон.

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")))

Чтобы повторно использовать набор входных данных из эксперимента в AnyLogic Cloud, мы используем функцию CloudClient.create_inputs_from_experiment() вместо create_default_inputs(). Скопировав входные данные эксперимента Baseline, мы изменяем входные данные Mean service time на диапазонный тип (от 1,8 до 2,0 с шагом 0,1), вызвав функцию Inputs.set_range_input(). Таким образом, будет выполнено три прогона эксперимента.

Затем нужно создать объект ModelRun типа вариации параметров. Для этого мы выполняем функцию CloudClient.create_parameter_variation().

Запуск эксперимента с вариацией параметров вызывается функцией get_outputs_and_run_if_absent() точно так же, как и в предыдущих примерах, но есть одно важное отличие. Все выходные данные экспериментов с несколькими «прогонами» могут представлять собой очень большой фрагмент данных, поэтому пользователь API должен четко указать, какие выходные данные нужны. Для этого нужно указать в качестве параметра имена выходных данных в массиве в функциях get_outputs_and_run_if_absent() или get_outputs(). В этом примере показаны выходные данные Mean queue size|Mean queue size. Если параметр не указан для эксперимента с несколькими «прогонами», возвращаются только скалярные выходные данные, если таковые имеются.

Результаты эксперимента с несколькими «прогонами» возвращаются в виде объекта MultiRunOutputs, который имеет ряд функций, упрощающих навигацию.

Запуск варьирования параметров с репликациями

В данном примере представлен запуск эксперимента варьирования параметров с четко заданным количеством репликаций. Один из входных параметров будет варьироваться в диапазоне. Также мы будем использовать входные значения реально существующего простого эксперимента в AnyLogic Cloud.

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"]))

Как видите, этот эксперимент по большей части не отличается от приведенного примера с вариацией параметров. Основное отличие состоит в том, что мы используем функцию set_number_of_replications() для четкого указания количества репликаций.

Затем с помощью функции CloudClient.create_parameter_variation_with_replications() мы создаем объект ModelRun для нужного типа эксперимента.

«Прогон» эксперимента запускается вызовом функции get_outputs_and_run_if_absent() точно так же, как и в предыдущих примерах, но есть одно важное отличие. Все выходные данные экспериментов с несколькими «прогонами» могут представлять собой очень большой фрагмент данных, поэтому пользователь API должен четко указать, какие выходные данные нужны. Для этого нужно указать в качестве параметра имена выходных данных в массиве в функциях get_outputs_and_run_if_absent() или get_outputs(). В этом примере показаны выходные данные Total time in system | Total time in system. Если параметр не указан для эксперимента с несколькими прогонами, возвращаются только скалярные выходные данные, если таковые имеются.

Результаты эксперимента с несколькими прогонами возвращаются в виде объекта MultiRunOutputs, который имеет ряд функций, упрощающих навигацию.

Запуск Монте-Карло 1-го порядка

В данном примере запустим эксперимент Монте-Карло 1-го порядка с четко заданным количеством репликаций. При выполнении через API эксперимент Монте-Карло запускается заданное количество раз, получает наборы значений, а затем возвращает их.

Один из входных параметров будет варьироваться в диапазоне. Мы также будем использовать входные значения из существующего простого эксперимента AnyLogic Cloud.

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_number_of_replications(4)
monte_carlo = client.create_monte_carlo(inputs)
outputs = monte_carlo.get_outputs_and_run_if_absent(["Total time in system|Total time in system"])
        
print("Monte Carlo 1st order 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"]))

Мы повторно используем набор входных данных из эксперимента, находящегося в AnyLogic Cloud, с помощью функции create_inputs_from_experiment(). Количество репликаций задается с помощью функции set_number_of_replications().

Затем мы создаем объект ModelRun для нужного типа эксперимента с помощью функции CloudClient.create_monte_carlo().

Прогон эксперимента запускается вызовом функции get_outputs_and_run_if_absent() точно так же, как и в предыдущих примерах, но есть одно важное отличие. Все выходные данные экспериментов с несколькими прогонами могут представлять собой очень большой фрагмент данных, поэтому пользователь API должен четко указать, какие выходные данные нужны. Для этого нужно указать в качестве параметра имена выходных данных в массиве в функциях get_outputs_and_run_if_absent() или get_outputs(). В этом примере показаны выходные данные Total time in system | Total time in system. Если параметр не указан для эксперимента с несколькими прогонами, возвращаются только скалярные выходные данные, если таковые имеются.

Выходные данные эксперимента с несколькими прогонами возвращаются в виде объекта MultiRunOutputs, который имеет ряд функций, упрощающих навигацию.

Как мы можем улучшить эту статью?

Отправить