Run Workflow from Task File with Input Signals

Run Workflow from Task File with Input Signals.

This example demonstrates how to run a SIMA workflow defined in an external .stask file using the SIMAPy API and the SIMA Runtime Engine (SRE).

The script shows the complete workflow of:

  1. Creating input signals and writing them to a file

  2. Importing a pre-defined workflow from a .stask file

  3. Executing the workflow with SRE

  4. Reading and processing the workflow results from output files

Features:
Workflow File Integration:

Shows how to use externally defined workflows (.stask files)

Signal Generation:

Creates input signals programmatically

File I/O:

Demonstrates reading and writing signal data from/to files

Results Processing:

Extracts and displays structured data from workflow output

Requirements:

  • Environment variable SRE_EXE must be set to point to the SIMA Runtime executable

  • A .stask file containing the workflow definition (included in repository)

Example

Basic usage of this script:

>>> python run_from_stask_with_file.py

This example can be adapted to work with any workflow defined in a .stask file by modifying the input signals and output processing as needed.

main()[source]

Execute the workflow from .stask file example.

Creates input signals, writes them to a file, imports a workflow from a .stask file, runs the workflow, and processes the results.

process_results(output_file)[source]

Process and display the results from the workflow execution.

Parameters:

output_file (Path) – Path to the output JSON file containing workflow results

Source Code

run_from_stask_with_file.py
  1"""Run Workflow from Task File with Input Signals.
  2
  3This example demonstrates how to run a SIMA workflow defined in an external .stask file
  4using the SIMAPy API and the SIMA Runtime Engine (SRE).
  5
  6The script shows the complete workflow of:
  7    1. Creating input signals and writing them to a file
  8    2. Importing a pre-defined workflow from a .stask file
  9    3. Executing the workflow with SRE
 10    4. Reading and processing the workflow results from output files
 11
 12Features:
 13    Workflow File Integration:
 14        Shows how to use externally defined workflows (.stask files)
 15    Signal Generation:
 16        Creates input signals programmatically
 17    File I/O:
 18        Demonstrates reading and writing signal data from/to files
 19    Results Processing:
 20        Extracts and displays structured data from workflow output
 21
 22Requirements:
 23    - Environment variable `SRE_EXE` must be set to point to the SIMA Runtime executable
 24    - A .stask file containing the workflow definition (included in repository)
 25
 26Example:
 27    Basic usage of this script:
 28    
 29    >>> python run_from_stask_with_file.py
 30    
 31    This example can be adapted to work with any workflow defined in a .stask file
 32    by modifying the input signals and output processing as needed.
 33"""
 34import os
 35import shutil
 36from pathlib import Path
 37from simapy.sre import SIMA
 38from simapy.sima_reader import SIMAReader
 39from simapy.sima_writer import SIMAWriter
 40from simapy.sima.signals.container import Container
 41from simapy.sima.signals.equallyspacedsignal import EquallySpacedSignal
 42
 43
 44def main():
 45    """Execute the workflow from .stask file example.
 46    
 47    Creates input signals, writes them to a file, imports a workflow from a .stask file,
 48    runs the workflow, and processes the results.
 49    """
 50    # Set up workspace
 51    ws = Path("output/workflow/ws")
 52    if ws.exists():
 53        shutil.rmtree(ws, ignore_errors=True)
 54    os.makedirs(ws, exist_ok=True)
 55
 56    # Create input signals
 57    a = EquallySpacedSignal(name="a")
 58    a.value = [1.0, 2.0, 3.0]
 59    b = EquallySpacedSignal(name="b")
 60    b.value = [3.0, 4.0, 5.0]
 61
 62    # Write signals to input file
 63    SIMAWriter().write([a, b], ws / "input.json")
 64
 65    # Locate stask file in folder "input" at the root of this repository
 66    stask = Path(__file__).parent / ".." / ".." / "input" / "workflow" / "simple_workflow_run.stask"
 67
 68    # Prepare SRE commands
 69    commands = []
 70    commands.append("--import")
 71    commands.append("file=" + str(stask.absolute()))
 72    commands.append("--run")
 73    commands.append("task=WorkflowTask")
 74    commands.append("workflow=outer")
 75
 76    # Run the workflow with SRE
 77    exe = os.getenv('SRE_EXE')
 78    sima = SIMA(exe=exe)
 79    sima.run(ws, commands)
 80    
 81    # Process results
 82    output = ws / "Output/output.json"
 83    process_results(output)
 84
 85
 86def process_results(output_file: Path):
 87    """Process and display the results from the workflow execution.
 88    
 89    Args:
 90        output_file: Path to the output JSON file containing workflow results
 91    """
 92    sima_reader = SIMAReader()
 93    items = sima_reader.read(output_file)
 94    
 95    if not items:
 96        print("No results found in output file")
 97        return
 98        
 99    set_container: Container = items[0]
100    run_containers = set_container.containers
101    
102    for run_container in run_containers:
103        print(f"Run {run_container.name}")
104        if run_container.signals:
105            result: EquallySpacedSignal = run_container.signals[0]
106            print(f"{result.name}={result.value}")
107        else:
108            print("No signals found in container")
109
110
111if __name__ == "__main__":
112    main()
113
114
115
116
117