Robust design with optimization with proxy solver#

This example demonstrates how to create robust design optimization workflow.

It creates multiple parametric systems using Proxy Solver node as a solver and then runs the workflow by parts.

Perform required imports#

Perform the required imports.

import time
from typing import Union

from ansys.optislang.core import Optislang
import ansys.optislang.core.node_types as node_types
from ansys.optislang.core.nodes import (
    DesignFlow,
    ExecutionOption,
    IntegrationNode,
    Node,
    ParametricSystem,
    ProxySolverNode,
)
from ansys.optislang.core.project_parametric import (
    ComparisonType,
    MixedParameter,
    ObjectiveCriterion,
    OptimizationParameter,
    ParameterType,
)
from ansys.optislang.core.utils import find_all_osl_exec

Create solver#

Define a simple calculator function to solve the variations.

def calculator(hid, X1, X2, X3, X4, X5):
    from math import sin, sqrt

    Y = 0.5 * X1 + X2 + 0.5 * X1 * X2 + 5 * sin(X3) + 0.2 * X4 + 0.1 * X5
    Z = ((-1) * sqrt(abs(Y))) ** 3
    return Y, Z


def calculate(designs):
    result_design_list = []
    print(f"Calculate {len(designs)} designs")
    for design in designs:
        hid = design["hid"]
        parameters = design["parameters"]
        X1 = 0.0
        X2 = 0.0
        X3 = 0.0
        X4 = 0.0
        X5 = 0.0
        for parameter in parameters:
            if parameter["name"] == "X1":
                X1 = parameter["value"]
            elif parameter["name"] == "X2":
                X2 = parameter["value"]
            elif parameter["name"] == "X3":
                X3 = parameter["value"]
            elif parameter["name"] == "X4":
                X4 = parameter["value"]
            elif parameter["name"] == "X5":
                X5 = parameter["value"]
        Y, Z = calculator(hid, X1, X2, X3, X4, X5)

        result_design = {}
        result_design["hid"] = hid
        responses = []
        responses.append({"name": "Y", "value": Y})
        responses.append({"name": "Z", "value": Z})
        result_design["responses"] = responses
        result_design_list.append(result_design)

    print(f"Return {len(result_design_list)} designs")
    return result_design_list

Create workflow creation and execution routines#

Define a routine that adds a proxy solver node into parametric system and registers parameters, responses and criteria.

Define a routine that handles ProxySolver node execution.

def add_solver_node_to_parent_system(
    parent_system: ParametricSystem,
    parameter_type: Union[
        ParameterType.DETERMINISTIC, ParameterType.MIXED
    ] = ParameterType.DETERMINISTIC,
) -> IntegrationNode:
    """Create and set up solver node within the parent system.

    Parameters
    ----------
    parent_system : ParametricSystem
        Parent system to which the solver node will be added.
    parameter_type: Union[ParameterType.DETERMINISTIC, ParameterType.MIXED]
        Parameter type to be created in parent system.

    Returns
    -------
    IntegrationNode
        The created solver node.
    """
    solver: ProxySolverNode = parent_system.create_node(
        type_=node_types.ProxySolver, name="Proxy Solver", design_flow=DesignFlow.RECEIVE_SEND
    )

    # Set the desired maximum number of designs you handle in one go.
    multi_design_launch_num = -1  # set -1 to solve all designs simultaneously
    solver.set_property("MultiDesignLaunchNum", multi_design_launch_num)
    solver.set_property("ForwardHPCLicenseContextEnvironment", True)

    # Load the available parameters and responses.

    load_json = {}
    load_json["parameters"] = []
    load_json["responses"] = []

    for i in range(1, 6):
        parameter = {"dir": {"value": "input"}, "name": f"X{i}", "value": 1.0}
        load_json["parameters"].append(parameter)

    response_y = {"dir": {"value": "output"}, "name": "Y", "value": 3.0}
    response_z = {"dir": {"value": "output"}, "name": "Z", "value": 3.0}
    load_json["responses"].append(response_y)
    load_json["responses"].append(response_z)

    solver.load(args=load_json)

    # Register parameters and responses to be available in the amop system.

    solver.register_locations_as_parameter()
    solver.register_locations_as_response()

    # Change parameter bounds.

    for i in range(1, 6):
        if parameter_type == ParameterType.DETERMINISTIC:
            parent_system.parameter_manager.modify_parameter(
                OptimizationParameter(name=f"X{i}", reference_value=1.0, range=(-3.14, 3.14))
            )
        elif parameter_type == ParameterType.MIXED:
            parent_system.parameter_manager.modify_parameter(
                MixedParameter(name=f"X{i}", reference_value=1.0, range=(-3.14, 3.14))
            )

    # Create a criterion in the amop system

    parent_system.criteria_manager.add_criterion(
        ObjectiveCriterion(name="obj_y", expression="Y", criterion=ComparisonType.MIN)
    )
    parent_system.criteria_manager.add_criterion(
        ObjectiveCriterion(name="obj_z", expression="Z", criterion=ComparisonType.MIN)
    )
    return solver


def run_proxy_solver_in_parent_system(osl: Optislang, solver: ProxySolverNode):
    """Run the solver node within the parent system.

    Parameters
    ----------
    osl : Optislang
        The optiSLang instance.
    solver : IntegrationNode
        The solver node to run.
    """
    # Loop until get_status() returns "FINISHED" for the project system. Use the GET_DESIGNS query and the SET_DESIGNS command
    # for the Proxy Solver node to get designs and set responses until the system is done.
    while True:
        status = osl.application.project.get_status()
        if status == "FINISHED":
            osl.log.info(f"Project status: {status}")
            break
        elif status == "STOPPED":
            osl.log.info(f"Project status: {status}")
            break
        design_list = solver.get_designs()
        if len(design_list):
            responses_dict = calculate(design_list)
            solver.set_designs(responses_dict)
        time.sleep(0.1)

Create optiSLang instance#

Find the optiSLang >= 25.1 executable. Initialize the Optislang class instance with the executable.

available_optislang_executables = find_all_osl_exec()
version, executables = available_optislang_executables.popitem(last=False)
# ProxySolver node is available since version 25R1
if not version >= 251:
    raise KeyError("OptiSLang installation >= 25R1 wasn't found, please specify path manually.")

osl = Optislang(executable=executables[0], loglevel="INFO")
osl.log.info(f"Using optiSLang version {osl.osl_version_string}")

Create workflow#

root_system = osl.application.project.root_system

AMOP system of your choice#

amop_system: ParametricSystem = root_system.create_node(type_=node_types.AMOP, name="AMOP")

# Optionally modify algorithm settings
# num_discretization = 2000
# amop_settings = amop_system.get_property("AlgorithmSettings")
# amop_settings["num_discretization"] = num_discretization
# amop_system.set_property("AlgorithmSettings", amop_settings)

# Fast running solver settings

amop_system.set_property("AutoSaveMode", "no_auto_save")
amop_system.set_property("SolveTwice", True)
amop_system.set_property("UpdateResultFile", "at_end")
# amop_system.set_property("WriteDesignStartSetFlag", False)

# Add the Proxy Solver node.

amop_proxy_solver: ProxySolverNode = add_solver_node_to_parent_system(amop_system)

Optimization on MOP#

oco_on_mop: ParametricSystem = root_system.create_node(type_=node_types.OCO, name="OCO_MOP")

# oco_on_mop.set_property("PreferCriteriaFromSlot", True)
oco_on_mop.set_property("AutoSaveMode", "no_auto_save")
oco_on_mop.set_property("SolveTwice", True)
oco_on_mop.set_property("UpdateResultFile", "at_end")
# oco_on_mop.set_property("ParameterMergingMode", "merge_from_slot")

oco_mop_solver: IntegrationNode = oco_on_mop.create_node(
    type_=node_types.Mopsolver, name="MOP Solver", design_flow=DesignFlow.RECEIVE_SEND
)

# connect
amop_system.get_output_slots("OParameterManager")[0].connect_to(
    oco_on_mop.get_input_slots("IParameterManager")[0]
)
amop_system.get_output_slots("OMDBPath")[0].connect_to(
    oco_mop_solver.get_input_slots("IMDBPath")[0]
)

ref_val = float(1.0000000000000000001)

for i in range(1, 6):
    oco_mop_solver.register_location_as_parameter(
        location={
            "base": "X1",
            "dir": {"enum": ["input", "output"], "value": "input"},
            "id": f"X{i}",
            "suffix": "",
            "value_type": {
                "enum": ["value", "cop", "rmse", "error", "abs_error", "density"],
                "value": "value",
            },
        },
        # {'is_important': True},
        reference_value=ref_val,
    )

oco_mop_solver.register_location_as_response(
    location={
        "base": "Y",
        "dir": {"value": "output"},
        "id": "Y",
        "suffix": "",
        "value_type": {"value": "value"},
    },
    reference_value=0.6987874926243327,
)

oco_mop_solver.register_location_as_response(
    location={
        "base": "Z",
        "dir": {"value": "output"},
        "id": "Z",
        "suffix": "",
        "value_type": {"value": "value"},
    },
    reference_value=-0.5841409930323823,
)

for i in range(1, 6):
    oco_on_mop.parameter_manager.modify_parameter(
        OptimizationParameter(name=f"X{i}", reference_value=1.0, range=(-3.14, 3.14))
    )

oco_on_mop.criteria_manager.add_criterion(
    ObjectiveCriterion(name="obj_y", expression="Y", criterion=ComparisonType.MIN)
)
oco_on_mop.criteria_manager.add_criterion(
    ObjectiveCriterion(name="obj_z", expression="Z", criterion=ComparisonType.MIN)
)

Filter designs#

filter_node: IntegrationNode = root_system.create_node(
    type_=node_types.DataMining, name="VALIDATOR_FILTER_NODE"
)

# connect
osl.osl_server.create_input_slot(filter_node.uid, "IBestDesigns")
oco_on_mop.get_output_slots("OBestDesigns")[0].connect_to(
    filter_node.get_input_slots("IBestDesigns")[0]
)


ofilter = {
    "OFilteredBestDesigns": [
        {
            "First": {"name": "AddDesignsFromSlot"},
            "Second": [
                {"design_container": []},
                {"string": "OBestDesigns"},
                {"design_entry": False},
            ],
        }
    ]
}


dmm = filter_node.get_property("DataMiningManager")
dmm["id_filter_list_map"] = ofilter
filter_node.set_property("DataMiningManager", dmm)

getbestdesigns = {
    "First": {"name": "GetBestDesigns"},
    "Second": [{"design_container": []}, {"design_entry": 2}],  # number of best designs - user set?
}

dmm = filter_node.get_property("DataMiningManager")
dmm["id_filter_list_map"]["OFilteredBestDesigns"].append(getbestdesigns)
filter_node.set_property("DataMiningManager", dmm)

filter_node.load()
filter_node.register_location_as_output_slot(
    location="OFilteredBestDesigns", name="OFilteredBestDesigns"
)

Validator system#

validator_system: ParametricSystem = osl.application.project.root_system.create_node(
    type_=node_types.Sensitivity, name="Validator System"
)
validator_proxy_solver = add_solver_node_to_parent_system(validator_system)

# Connect
filter_node.get_output_slots("OFilteredBestDesigns")[0].connect_to(
    validator_system.get_input_slots("IStartDesigns")[0]
)
oco_on_mop.get_output_slots("OCriteria")[0].connect_to(
    validator_system.get_input_slots("ICriteria")[0]
)

Design filter for postprocessing#

append_node: IntegrationNode = osl.application.project.root_system.create_node(
    type_=node_types.DataMining, name="Append Designs"
)

# python script to workaround missing pyoptislang functionalities
command = f"append_node = find_actor('Append Designs')\n" "append_node.init_append_best_designs()\n"
osl.application.project.run_python_script(command)

validator_system.get_output_slots("ODesigns")[0].connect_to(
    append_node.get_input_slots("IDesigns")[0]
)
oco_on_mop.get_output_slots("OMDBPath")[0].connect_to(append_node.get_input_slots("IMDBPath")[0])

Postprocessing node#

postprocessing_node: Node = root_system.create_node(
    type_=node_types.Postprocessing, name="PostProcessing"
)

# connect
append_node.get_output_slots("OValidatedMDBPath")[0].connect_to(
    postprocessing_node.get_input_slots("IMDBPath")[0]
)

Optimization on solver#

oco_on_solver: ParametricSystem = root_system.create_node(type_=node_types.OCO, name="OCO_SOLVER")

# oco_on_solver.set_property("PreferCriteriaFromSlot", True)
oco_on_solver.set_property("AutoSaveMode", "no_auto_save")
oco_on_solver.set_property("SolveTwice", True)
oco_on_solver.set_property("UpdateResultFile", "at_end")
# oco_on_solver.set_property("ParameterMergingMode", "merge_from_slot")

oco_proxy_solver = add_solver_node_to_parent_system(oco_on_solver)

# connect
validator_system.get_output_slots("OBestDesigns")[0].connect_to(
    oco_on_solver.get_input_slots("IStartDesigns")[0]
)

Robustness system#

robustness: ParametricSystem = root_system.create_node(
    type_=node_types.Robustness, name="Robustness"
)
robustness_solver = add_solver_node_to_parent_system(robustness, ParameterType.MIXED)

# connect
oco_on_solver.get_output_slots("OBestDesigns")[0].connect_to(
    robustness.get_input_slots("INominalDesigns")[0]
)

MOP node#

mop_node = root_system.create_node(type_=node_types.Mop, name="MOP")

# connect
robustness.get_output_slots("OMDBPath")[0].connect_to(mop_node.get_input_slots("IMDBPath")[0])
robustness.get_output_slots("OParameterManager")[0].connect_to(
    mop_node.get_input_slots("IParameterManager")[0]
)

osl.log.info("Workflow created")

Optionally save project#

If you want to save the project to some desired location, uncomment and edit these lines:

from pathlib import Path
dir_path = Path(r"<insert-desired-location>")
project_name = "rdo_workflow.opf"
osl.application.save_as(dir_path / project_name)

Run workflow#

Run the workflow created by the preceding scripts. In this example, workflow is run in multiple steps. At first, all systems/nodes are set to inactive except the AMOP system with it’s proxy solver. The AMOP system is then executed. Then, the other systems/nodes are activated and run one by one.