idmtools_models.templated_script_task module#

Provides the TemplatedScriptTask.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.

class idmtools_models.templated_script_task.TemplatedScriptTask(command: str | ~idmtools.entities.command_line.CommandLine = <property object>, platform_requirements: ~typing.Set[~idmtools.entities.platform_requirements.PlatformRequirements] = <factory>, _ITask__pre_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, _ITask__post_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, common_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, script_path: str = None, script_binary: str = None, template: str = None, template_file: str = None, template_is_common: bool = True, variables: ~typing.Dict[str, ~typing.Any] = <factory>, path_sep: str = '/', extra_command_arguments: str = '', gather_common_asset_hooks: ~typing.List[~typing.Callable[[~idmtools.entities.itask.ITask], ~idmtools.assets.asset_collection.AssetCollection]] = <factory>, gather_transient_asset_hooks: ~typing.List[~typing.Callable[[~idmtools.entities.itask.ITask], ~idmtools.assets.asset_collection.AssetCollection]] = <factory>)[source]#

Bases: ITask

Defines a task to run a script using a template. Best suited to shell scripts.

Examples

In this example, we add modify the Python Path using TemplatedScriptTask and LINUX_PYTHON_PATH_WRAPPER

import os

from idmtools.core.platform_factory import Platform
from idmtools.entities.experiment import Experiment
from idmtools_models.python.python_task import PythonTask
from idmtools_models.templated_script_task import TemplatedScriptTask, get_script_wrapper_unix_task, LINUX_PYTHON_PATH_WRAPPER


platform = Platform("CALCULON")
# This task can be anytype of task that would run python. Here we are running a simple model script that consumes the example
# package "a_package"
task = PythonTask(script_path="model.py", python_path='python3.6')
# add our library. On Comps, you could use RequirementsToAssetCollection as well
task.common_assets.add_asset("a_package.py")
# we request a wrapper script for Unix. The wrapper should match the computation platform's OS
# We also use the built-it LINUX_PYTHON_PATH_WRAPPER template which modifies our PYTHONPATH to load libraries from Assets/site-packages and Assets folders
wrapper_task: TemplatedScriptTask = get_script_wrapper_unix_task(task, template_content=LINUX_PYTHON_PATH_WRAPPER)
# we have to set the bash path remotely
wrapper_task.script_binary = "/bin/bash"

# Now we define our experiment. We could just as easily use this wrapper in a templated simulation builder as well
experiment = Experiment.from_task(name=os.path.basename(__file__), task=wrapper_task)
experiment.run(wait_until_done=True)

In this example, we modify environment variable using TemplatedScriptTask and LINUX_DICT_TO_ENVIRONMENT

import os
from idmtools.core.platform_factory import Platform
from idmtools.entities.experiment import Experiment
from idmtools_models.python.python_task import PythonTask
from idmtools_models.templated_script_task import get_script_wrapper_unix_task, LINUX_DICT_TO_ENVIRONMENT


platform = Platform("CALCULON")
# here we define the task we want to use the environment variables. In this example we have a simple python script that prints the EXAMPLE environment variable
task = PythonTask(script_path="model.py")
# Get a task to wrap the script in a shell script. Which get_script_wrapper function you use depends on the platform's OS
wrapper_task = get_script_wrapper_unix_task(
    task=task,
    # and set some values here
    variables=dict(EXAMPLE='It works!')
)
# some platforms need to you hint where their script binary is. Usually this is only applicable to Unix platforms(Linux, Mac, etc)
wrapper_task.script_binary = "/bin/bash"

# Now we define our experiment. We could just as easily use this wrapper in a templated simulation builder as well
experiment = Experiment.from_task(name=os.path.basename(__file__), task=wrapper_task)
experiment.run(wait_until_done=True)
script_path: str = None#

Name of script

script_binary: str = None#

If platform requires path to script executing binary(ie /bin/bash)

template: str = None#

The template contents

template_file: str = None#

The template file. You can only use either template or template_file at once

template_is_common: bool = True#

Controls whether a template should be an experiment or a simulation level asset

variables: Dict[str, Any]#
path_sep: str = '/'#

Platform Path Separator. For Windows execution platforms, use , otherwise use the default of /

extra_command_arguments: str = ''#

Extra arguments to add to the command line

gather_common_asset_hooks: List[Callable[[ITask], AssetCollection]]#

Hooks to gather common assets

gather_transient_asset_hooks: List[Callable[[ITask], AssetCollection]]#

Hooks to gather transient assets

gather_common_assets() AssetCollection[source]#

Gather common(experiment-level) assets for task.

Returns:

AssetCollection containing common assets

gather_transient_assets() AssetCollection[source]#

Gather transient(experiment-level) assets for task.

Returns:

AssetCollection containing transient assets

reload_from_simulation(simulation: Simulation)[source]#

Reload a templated script task. When reloading, you will only have the rendered template available.

Parameters:

simulation

Returns:

None

pre_creation(parent: Simulation | IWorkflowItem, platform: IPlatform)[source]#

Before creating simulation, we need to set our command line.

Parameters:
  • parent – Parent object

  • platform – Platform item is being ran on

Returns:

__init__(command: str | ~idmtools.entities.command_line.CommandLine = <property object>, platform_requirements: ~typing.Set[~idmtools.entities.platform_requirements.PlatformRequirements] = <factory>, _ITask__pre_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, _ITask__post_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, common_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, script_path: str = None, script_binary: str = None, template: str = None, template_file: str = None, template_is_common: bool = True, variables: ~typing.Dict[str, ~typing.Any] = <factory>, path_sep: str = '/', extra_command_arguments: str = '', gather_common_asset_hooks: ~typing.List[~typing.Callable[[~idmtools.entities.itask.ITask], ~idmtools.assets.asset_collection.AssetCollection]] = <factory>, gather_transient_asset_hooks: ~typing.List[~typing.Callable[[~idmtools.entities.itask.ITask], ~idmtools.assets.asset_collection.AssetCollection]] = <factory>) None#
class idmtools_models.templated_script_task.ScriptWrapperTask(command: str | ~idmtools.entities.command_line.CommandLine = <property object>, platform_requirements: ~typing.Set[~idmtools.entities.platform_requirements.PlatformRequirements] = <factory>, _ITask__pre_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, _ITask__post_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, common_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, template_script_task: ~idmtools_models.templated_script_task.TemplatedScriptTask = None, task: ~idmtools.entities.itask.ITask = None)[source]#

Bases: ITask

Allows you to wrap a script with another script.

Raises:

ValueError if the template Script Task is not defined

template_script_task: TemplatedScriptTask = None#
task: ITask = None#
static from_dict(task_dictionary: Dict[str, Any])[source]#

Load the task from a dictionary.

property command#

Our task property. Again, we have to overload this because of wrapping a task.

property wrapped_task#

Our task we are wrapping with a shell script.

Returns:

Our wrapped task

gather_common_assets()[source]#

Gather all the common assets.

Returns:

Common assets(Experiment Assets)

gather_transient_assets() AssetCollection[source]#

Gather all the transient assets.

Returns:

Transient Assets(Simulation level assets)

reload_from_simulation(simulation: Simulation)[source]#

Reload from simulation.

Parameters:

simulation – simulation

Returns:

None

pre_creation(parent: Simulation | IWorkflowItem, platform: IPlatform)[source]#

Before creation, create the true command by adding the wrapper name.

Here we call both our wrapped task and our template_script_task pre_creation :param parent: Parent Task :param platform: Platform Templated Task is executing on

Returns:

None

post_creation(parent: Simulation | IWorkflowItem, platform: IPlatform)[source]#

Post creation of task.

Here we call both our wrapped task and our template_script_task post_creation

Parameters:
  • parent – Parent of task

  • platform – Platform we are running on

Returns:

None

__init__(command: str | ~idmtools.entities.command_line.CommandLine = <property object>, platform_requirements: ~typing.Set[~idmtools.entities.platform_requirements.PlatformRequirements] = <factory>, _ITask__pre_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, _ITask__post_creation_hooks: ~typing.List[~typing.Callable[[Simulation | IWorkflowItem, IPlatform], ~typing.NoReturn]] = <factory>, common_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, transient_assets: ~idmtools.assets.asset_collection.AssetCollection = <factory>, template_script_task: ~idmtools_models.templated_script_task.TemplatedScriptTask = None, task: ~idmtools.entities.itask.ITask = None) None#
idmtools_models.templated_script_task.get_script_wrapper_task(task: ITask, wrapper_script_name: str, template_content: str | None = None, template_file: str | None = None, template_is_common: bool = True, variables: Dict[str, Any] | None = None, path_sep: str = '/') ScriptWrapperTask[source]#

Convenience function that will wrap a task for you with some defaults.

Parameters:
  • task – Task to wrap

  • wrapper_script_name – Wrapper script name

  • template_content – Template Content

  • template_file – Template File

  • template_is_common – Is the template experiment level

  • variables – Variables

  • path_sep – Path sep(Window or Linux)

Returns:

ScriptWrapperTask wrapping the task

idmtools_models.templated_script_task.get_script_wrapper_windows_task(task: ITask, wrapper_script_name: str = 'wrapper.bat', template_content: str = '{% for key, value in vars.items() %}\nset {{key}}="{{value}}"\n{% endfor %}\necho Running %*\n%*', template_file: str | None = None, template_is_common: bool = True, variables: Dict[str, Any] | None = None) ScriptWrapperTask[source]#

Get wrapper script task for windows platforms.

The default content wraps a another task with a batch script that exports the variables to the run environment defined in variables. To modify python path, use WINDOWS_PYTHON_PATH_WRAPPER

You can adapt this script to modify any pre-scripts you need or call others scripts in succession

Parameters:
  • task – Task to wrap

  • wrapper_script_name – Wrapper script name(defaults to wrapper.bat)

  • template_content – Template Content.

  • template_file – Template File

  • template_is_common – Is the template experiment level

  • variables – Variables for template

Returns:

ScriptWrapperTask

See Also::

idmtools_models.templated_script_task.TemplatedScriptTask idmtools_models.templated_script_task.get_script_wrapper_task() idmtools_models.templated_script_task.get_script_wrapper_unix_task()

idmtools_models.templated_script_task.get_script_wrapper_unix_task(task: ITask, wrapper_script_name: str = 'wrapper.sh', template_content: str = '{% for key, value in vars.items() %}\nexport {{key}}="{{value}}"\n{% endfor %}\necho Running args $@\n"$@"\n', template_file: str | None = None, template_is_common: bool = True, variables: Dict[str, Any] | None = None)[source]#

Get wrapper script task for unix platforms.

The default content wraps a another task with a bash script that exports the variables to the run environment defined in variables. To modify python path, you can use LINUX_PYTHON_PATH_WRAPPER

You can adapt this script to modify any pre-scripts you need or call others scripts in succession

Parameters:
  • task – Task to wrap

  • wrapper_script_name – Wrapper script name(defaults to wrapper.sh)

  • template_content – Template Content

  • template_file – Template File

  • template_is_common – Is the template experiment level

  • variables – Variables for template

Returns:

ScriptWrapperTask

See Also: idmtools_models.templated_script_task.TemplatedScriptTask idmtools_models.templated_script_task.get_script_wrapper_task() idmtools_models.templated_script_task.get_script_wrapper_windows_task()

class idmtools_models.templated_script_task.TemplatedScriptTaskSpecification[source]#

Bases: TaskSpecification

TemplatedScriptTaskSpecification provides the plugin specs for TemplatedScriptTask.

get(configuration: dict) TemplatedScriptTask[source]#

Get instance of TemplatedScriptTask with configuration.

Parameters:

configuration – configuration for TemplatedScriptTask

Returns:

TemplatedScriptTask with configuration

get_description() str[source]#

Get description of plugin.

Returns:

Plugin description

get_example_urls() List[str][source]#

Get example urls related to TemplatedScriptTask.

Returns:

List of urls that have examples related to CommandTask

get_type() Type[TemplatedScriptTask][source]#

Get task type provided by plugin.

Returns:

TemplatedScriptTask

get_version() str[source]#

Returns the version of the plugin.

Returns:

Plugin Version

class idmtools_models.templated_script_task.ScriptWrapperTaskSpecification[source]#

Bases: TaskSpecification

ScriptWrapperTaskSpecification defines the plugin specs for ScriptWrapperTask.

get(configuration: dict) ScriptWrapperTask[source]#

Get instance of ScriptWrapperTask with configuration.

Parameters:

configuration – configuration for ScriptWrapperTask

Returns:

TemplatedScriptTask with configuration

get_description() str[source]#

Get description of plugin.

Returns:

Plugin description

get_example_urls() List[str][source]#

Get example urls related to ScriptWrapperTask.

Returns:

List of urls that have examples related to CommandTask

get_type() Type[ScriptWrapperTask][source]#

Get task type provided by plugin.

Returns:

TemplatedScriptTask

get_version() str[source]#

Returns the version of the plugin.

Returns:

Plugin Version