Builders

Simulation builder

The follow demonstrates how to build a sweep using the standard builder, SimulationBuilder

This example uses the following model.

import json
import os
import sys

current_dir = os.path.abspath(os.getcwd())

# create 'output' dir in COMPS under current working dir which is one dir above "Assets" dir
output_dir = os.path.join(current_dir, "output")
config_path = os.path.join(current_dir, "config.json")
if not os.path.exists(output_dir):
    os.mkdir(output_dir)

config = json.load(open(config_path, "r"))
print(config)

# write each configs to result.json in comps's simulation output
with open(os.path.join(output_dir, "result.json"), "w") as fp:
    json.dump(config, fp)

sys.exit(0)

It then builds sweeps through Arms

import os
import sys

from idmtools.assets import AssetCollection
from idmtools.builders import SimulationBuilder
from idmtools.core.platform_factory import platform
from idmtools.entities.experiment import Experiment
from idmtools.entities.templated_simulation import TemplatedSimulations
from idmtools_models.python.json_python_task import JSONConfiguredPythonTask
from idmtools_test import COMMON_INPUT_PATH

with platform('Calculon'):
    base_task = JSONConfiguredPythonTask(
        script_path=os.path.join(COMMON_INPUT_PATH, "compsplatform", "working_model.py"),
        # add common assets from existing collection
        common_assets=AssetCollection.from_id('41c1b14d-0a04-eb11-a2c7-c4346bcb1553', as_copy=True)
    )

    ts = TemplatedSimulations(base_task=base_task)
    # sweep parameter
    builder = SimulationBuilder()
    builder.add_sweep_definition(JSONConfiguredPythonTask.set_parameter_partial("min_x"), range(-2, 0))
    builder.add_sweep_definition(JSONConfiguredPythonTask.set_parameter_partial("max_x"), range(1, 3))
    ts.add_builder(builder)

    e = Experiment.from_template(ts, name=os.path.split(sys.argv[0])[1])
    e.run(wait_until_done=True)
    # use system status as the exit code
    sys.exit(0 if e.succeeded else -1)

See SimulationBuilder for more details.

Arm experiment builder

The follow demonstrates how to build a sweep using ArmSimulationBuilder

This example uses the following model.

import json
import os
import sys

current_dir = os.path.abspath(os.getcwd())

# create 'output' dir in COMPS under current working dir which is one dir above "Assets" dir
output_dir = os.path.join(current_dir, "output")
config_path = os.path.join(current_dir, "config.json")
if not os.path.exists(output_dir):
    os.mkdir(output_dir)

config = json.load(open(config_path, "r"))
print(config)

# write each configs to result.json in comps's simulation output
with open(os.path.join(output_dir, "result.json"), "w") as fp:
    json.dump(config, fp)

sys.exit(0)

It then builds sweeps through Arms

"""
        This file demonstrates how to use ArmExperimentBuilder in PythonExperiment's builder.
        We are then adding the builder to PythonExperiment.

        |__sweep arm1
            |_ a = 1
            |_ b = [2,3]
            |_ c = [4,5]
        |__ sweep arm2
            |_ a = [6,7]
            |_ b = 2
        Expect sims with parameters:
            sim1: {a:1, b:2, c:4}
            sim2: {a:1, b:2, c:5}
            sim3: {a:1, b:3, c:4}
            sim4: {a:1, b:3, c:5}
            sim5: {a:6, b:2}
            sim6: {a:7, b:2}
        Note:
            arm1 and arm2 are adding to total simulations
"""
import os
import sys
from functools import partial

from idmtools.builders import SweepArm, ArmType, ArmSimulationBuilder
from idmtools.core.platform_factory import platform
from idmtools.entities.experiment import Experiment
from idmtools.entities.templated_simulation import TemplatedSimulations
from idmtools_models.python.json_python_task import JSONConfiguredPythonTask
from idmtools_test import COMMON_INPUT_PATH

# define specific callbacks for a, b, and c
setA = partial(JSONConfiguredPythonTask.set_parameter_sweep_callback, param="a")
setB = partial(JSONConfiguredPythonTask.set_parameter_sweep_callback, param="b")
setC = partial(JSONConfiguredPythonTask.set_parameter_sweep_callback, param="c")


if __name__ == "__main__":
    with platform('CALCULON'):
        base_task = JSONConfiguredPythonTask(script_path=os.path.join(COMMON_INPUT_PATH, "python", "model1.py"))
        # define that we are going to create multiple simulations from this task
        ts = TemplatedSimulations(base_task=base_task)

        # define our first sweep Sweep Arm
        arm1 = SweepArm(type=ArmType.cross)
        builder = ArmSimulationBuilder()
        arm1.add_sweep_definition(setA, 1)
        arm1.add_sweep_definition(setB, [2, 3])
        arm1.add_sweep_definition(setC, [4, 5])
        builder.add_arm(arm1)

        # adding more simulations with sweeping
        arm2 = SweepArm(type=ArmType.cross)
        arm2.add_sweep_definition(setA, [6, 7])
        arm2.add_sweep_definition(setB, [2])
        builder.add_arm(arm2)

        # add our builders to our template
        ts.add_builder(builder)

        # create experiment from the template
        experiment = Experiment.from_template(ts, name=os.path.split(sys.argv[0])[1],
                                              tags={"string_tag": "test", "number_tag": 123, "KeyOnly": None})
        # run the experiment
        experiment.run()
        # in most real scenarios, you probably do not want to wait as this will wait until all simulations
        # associated with an experiment are done. We do it in our examples to show feature and to enable
        # testing of the scripts
        experiment.wait()
        # use system status as the exit code
        sys.exit(0 if experiment.succeeded else -1)

See ArmSimulationBuilder for more details

Multiple argument sweep

The follow demonstrates how to build a sweep when multiple arguments are required at the same time. Typically, defining sweeps per argument is best like in the example Simulation builder but in some cases, such as when we need all parameters to create an object, we want these parameters passed to a single callback at the same time. This example uses the following model.

import json
import os
import sys

current_dir = os.path.abspath(os.getcwd())

# create 'output' dir in COMPS under current working dir which is one dir above "Assets" dir
output_dir = os.path.join(current_dir, "output")
config_path = os.path.join(current_dir, "config.json")
if not os.path.exists(output_dir):
    os.mkdir(output_dir)

config = json.load(open(config_path, "r"))
print(config)

# write each configs to result.json in comps's simulation output
with open(os.path.join(output_dir, "result.json"), "w") as fp:
    json.dump(config, fp)

sys.exit(0)

We then make a class within our example script that requires the parameters a, b, and c be defined at creation time. With this defined, we then add our sweep callback.

"""
This file demonstrates doing a multi-argument sweep

Sometimes you need multiple parameters at the same time, usually to create objects within a callback. The *
"""

import os
import sys
from dataclasses import dataclass
from idmtools.builders import SimulationBuilder
from idmtools.core.platform_factory import Platform
from idmtools.entities.experiment import Experiment
from idmtools_models.python.json_python_task import JSONConfiguredPythonTask
from idmtools_test import COMMON_INPUT_PATH


@dataclass
class ModelConfig:
    a: int
    b: int
    c: int


# define a custom sweep callback that sets b to a + 2
def param_update(simulation, a_value, b_value, c_value):
    mc = ModelConfig(a_value, b_value, c_value)
    simulation.task.set_parameter('a', mc.a)
    simulation.task.set_parameter('b', mc.b)
    simulation.task.set_parameter('c', mc.c)
    return dict(a=a_value, b=b_value, c=c_value)


if __name__ == "__main__":
    # define what platform we want to use. Here we use a context manager but if you prefer you can
    # use objects such as Platform('Calculon') instead
    with Platform('Calculon'):
        # define our base task
        base_task = JSONConfiguredPythonTask(script_path=os.path.join(COMMON_INPUT_PATH, "python", "model1.py"),
                                             parameters=dict())
        # define our input csv sweep
        builder = SimulationBuilder()
        # we can use add_sweep_definition call to do multiple parameter sweeping now
        builder.add_sweep_definition(param_update, range(2), range(2), range(2))
        #builder.add_multiple_parameter_sweep_definition(param_update, range(2), range(2), range(2))

        # define our experiment with its metadata
        experiment = Experiment.from_builder(
            builders=[builder], base_task=base_task,
            name=os.path.split(sys.argv[0])[1],
            tags={"string_tag": "test", "number_tag": 123}
        )

        # run experiment
        experiment.run()
        # wait until done with longer interval
        # in most real scenarios, you probably do not want to wait as this will wait until all simulations
        # associated with an experiment are done. We do it in our examples to show feature and to enable
        # testing of the scripts
        experiment.wait(refresh_interval=10)
        # use system status as the exit code
        sys.exit(0 if experiment.succeeded else -1)