Skip to content

execute_workflow

The main automation function that utilized a WorkflowConfig to run a sequence of phases based on its parameters.

Run a complete, cached HFSS experiment workflow.

The engine performs four deterministic phases:

  1. Prepare – create an isolated results folder and, if requested, copy the template .aedt project into it.
  2. Build – apply the current sweep parameters to the HFSS design through the configured builder object.
  3. Simulate – execute one or more analyses for every parameter set and store their JSON results.
  4. Aggregate – flatten and merge selected results into CSV files for downstream analysis.

Parameters:

Name Type Description Default
config WorkflowConfig

Parsed workflow definition. See the full field reference in api/workflow_config.md.

 required

Returns:

Type Description
None

None

Raises:

Type Description
ValueError

If the configuration is incomplete.
RuntimeError

If an HFSS session cannot be opened or a simulation fails unexpectedly.
Example 
from pathlib import Path
from quansys.workflow import (
    WorkflowConfig, PyaedtFileParameters,
    DesignVariableBuilder, execute_workflow
)
from quansys.simulation import EigenmodeAnalysis
from pycaddy.sweeper import DictSweep

cfg = WorkflowConfig(
    pyaedt_file_parameters=PyaedtFileParameters(
        file_path=Path("resources/simple_design.aedt"),
        non_graphical=True
    ),
    builder=DesignVariableBuilder(design_name="my_design"),
    builder_sweep=[DictSweep(parameters={"chip_base_width": ["3 mm", "4 mm"]})],
    simulations={
        "classical": EigenmodeAnalysis(setup_name="Setup1",
                                       design_name="my_design")
    },
    aggregation_dict={"classical_agg": ["classical"]}
)

execute_workflow(cfg)
# → results/aggregations/classical_agg.csv is produced
Source code in src/quansys/workflow/workflow.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def execute_workflow(config: WorkflowConfig) -> None:
    """
    Run a complete, cached HFSS experiment workflow.

    The engine performs four deterministic phases:

    1. **Prepare** – create an isolated results folder and, if requested,
       copy the template ``.aedt`` project into it.
    2. **Build** – apply the current sweep parameters to the HFSS design
       through the configured builder object.
    3. **Simulate** – execute one or more analyses for every parameter set
       and store their JSON results.
    4. **Aggregate** – flatten and merge selected results into CSV files
       for downstream analysis.

    Args:
        config (WorkflowConfig):
            Parsed workflow definition.
            See the full field reference in
            [`api/workflow_config.md`](../api/workflow_config.md).

    Returns:
        None

    Raises:
        ValueError: If the configuration is incomplete.
        RuntimeError: If an HFSS session cannot be opened or a simulation
            fails unexpectedly.

    Example:
        ```python
        from pathlib import Path
        from quansys.workflow import (
            WorkflowConfig, PyaedtFileParameters,
            DesignVariableBuilder, execute_workflow
        )
        from quansys.simulation import EigenmodeAnalysis
        from pycaddy.sweeper import DictSweep

        cfg = WorkflowConfig(
            pyaedt_file_parameters=PyaedtFileParameters(
                file_path=Path("resources/simple_design.aedt"),
                non_graphical=True
            ),
            builder=DesignVariableBuilder(design_name="my_design"),
            builder_sweep=[DictSweep(parameters={"chip_base_width": ["3 mm", "4 mm"]})],
            simulations={
                "classical": EigenmodeAnalysis(setup_name="Setup1",
                                               design_name="my_design")
            },
            aggregation_dict={"classical_agg": ["classical"]}
        )

        execute_workflow(cfg)
        # → results/aggregations/classical_agg.csv is produced
        ```
    """
    project = Project(root=config.root_folder)
    iteration_proj = project.sub("iterations")

    chain_sweep = ChainSweep(sweepers=config.builder_sweep)

    for params in chain_sweep.generate():

        # 1. PREPARE (copy template .aedt if policy allows)
        run_params = _prepare_folder_phase(
            cfg=config.prepare_folder,
            pyaedt=config.pyaedt_file_parameters,
            params=params,
            project=iteration_proj,
        )

        # 2. BUILD (apply parameter sweep values)
        _build_phase(config.builder, run_params, params, iteration_proj)

        # 3. SIMULATIONS
        _simulations_phase(
            config.simulations,
            params,
            run_params.model_copy(),
            config.keep_hfss_solutions,
            iteration_proj
        )

    # 4. AGGREGATION
    aggregation_proj = project.sub("aggregations")
    for name, identifiers in config.aggregation_dict.items():
        aggregator = Aggregator(identifiers=identifiers)
        _aggregation_phase(name, aggregator, aggregation_proj, iteration_proj)