idmtools_models.templated_script_task module#

Provides the TemplatedScriptTask.

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.