Run Workflow with SIMA Runtime Engine

Run Workflow with SIMA Runtime Engine.

This example demonstrates how run a SIMA workflow programmatically

The script shows the complete workflow of:

  1. Importing a workflow definition file

  2. Setting up the storage root to point to the current directory where the script is located

  3. Setting up input parameters (a scale value in this case)

  4. Running the workflow

  5. Processing and validating the results from the workflow execution

Example

Basic usage of this script:

>>> python run_workflow_with_sre.py
main()[source]

Execute the main workflow example.

Creates and runs a simple workflow using the SIMA Runtime Engine (SRE). The workflow takes a scale parameter and multiplies it by values from a signal. Results are validated after execution completes.

Source Code

run_workflow_with_sre.py
 1"""Run Workflow with SIMA Runtime Engine.
 2
 3This example demonstrates how run a SIMA workflow programmatically
 4
 5The script shows the complete workflow of:
 6    1. Importing a workflow definition file
 7    2. Setting up the storage root to point to the current directory where the script is located
 8    3. Setting up input parameters (a scale value in this case)
 9    4. Running the workflow
10    5. Processing and validating the results from the workflow execution
11
12
13Example:
14    Basic usage of this script:
15    
16    >>> python run_workflow_with_sre.py
17    
18"""
19import os
20import shutil
21from pathlib import Path
22from simapy.sre import SIMA
23from simapy.sima_reader import SIMAReader
24from simapy.sima import signals
25
26def main():
27    """Execute the main workflow example.
28    
29    Creates and runs a simple workflow using the SIMA Runtime Engine (SRE).
30    The workflow takes a scale parameter and multiplies it by values from a signal.
31    Results are validated after execution completes.
32    """
33    output = Path("output/workflow")
34    ws = output / "ws"
35    if ws.exists():
36        shutil.rmtree(ws, ignore_errors=True)
37    os.makedirs(ws, exist_ok=True)
38
39    input = Path(__file__).parent / ".." / ".." / "input"
40    json_file = input / "workflow" / "workflow_task.json"
41    # This will be used as an input
42    scale = 3.0
43
44    here =  Path(__file__).parent
45    commands = []
46    # First we import the json file to import the workflow
47    commands.append("--import")
48    commands.append("file=" + str(json_file.absolute()))
49    # Then we set the storage root to the current directory
50    commands.append("--storageroot")
51    commands.append("task=ExternalStorage")
52    commands.append("root=" + str(here))
53    # Then we run the workflow
54    commands.append("--run")
55    commands.append("task=WorkflowTask")
56    commands.append("workflow=workflow")
57    commands.append(f"input=scale={scale}")
58
59    # Requires that the environment is set, but an alternative path may be given
60    exe = os.getenv('SRE_EXE')
61    # Set show_output to False to disable console output, this will make the progress bare more visible
62    # The console output is still piped to a file within the workspace
63    sima = SIMA(exe=exe,show_output=False)
64    
65    # Run the workflow with SRE
66    sima.run(ws, commands)
67    
68    # Read and validate the output
69    output = ws / "out.json"
70    sima_reader = SIMAReader()
71    items = sima_reader.read(output)
72    
73    # Print validation results
74    if len(items) == 1:
75        print("Successfully executed workflow!")
76        # Access the result signal
77        container: signals.Container = items[0]
78        mynum: signals.DimensionalScalar = container.signals[1]
79        print(f"{mynum.name}={mynum.value}")
80        if mynum.value != scale:
81            raise ValueError(f"Number not as expected {mynum.value} vs {scale}")
82
83    else:
84        print("Failed to execute workflow or read results")
85
86
87if __name__ == "__main__":
88    main()