Mechanical Product Instance Manager

Objective

Drive an Ansys Mechanical product from a solution. Start the product instance in a long-running transaction, upload a geometry file, run a custom Mechanical script that solves a static structural analysis, download the solve.out file, then shut the instance down.

Source code for this example is in the example solution.

Feature highlight

Ansys Mechanical is a powerful simulation tool for structural analysis, supporting a wide range of applications including stress analysis, vibration, thermal, and fatigue simulations.

This example shows how to create a product instance manager for Mechanical products using SAF. The product instance manager enables management of the Mechanical product instance lifecycle, including starting, stopping, and accessing the product’s API.

In this example, you learn how to:

• Start a product instance from a long-running transaction decorated with @create_instance.

• Reuse the running instance in the other transaction methods through the @instance decorator.

• Call the Mechanical API to upload a geometry file and run a custom Python script that solves the model.

• Move files between the solution and the product with the storage_scope and an EntityHandle field.

• Publish progress on named event streams with transaction.raise_event.

• Wire one callback per button and use event listeners to refresh the console logs and the notifications.

Prerequisites

Note

This example requires the Mechanical 2025 R2 Service Pack 4 (25R2 SP4) product to be installed on your machine. To work with a Mechanical product instance, be sure to install the core-pim and instance-management-mechanical extras from the ansys-saf-sdk package. You can do this by manually editing your pyproject.toml.

ansys-saf-sdk = {version = "^0.3.0", extras = ["core-pim", "instance-management-mechanical"]}

This will install the supported version of ansys-mechanical-core to control the Mechanical product instances. This example uses PIM as the product instance management system. If you want to use HPS instead, replace the core-pim extra with core-hps. For more information, see Configuration.

The geometry file required for the example script, example_01_geometry.agdb, is bundled with the solution as a method asset and retrieved with self.transaction.get_asset_entity_handle("example_01_geometry.agdb") in upload_example_file_to_mechanical, so no manual setup is required.

Coding

To create a product instance manager for Mechanical products and interact with it, work through the following sequence of sections.

Backend

Create the solution definition.

Key concept — Product instance manager

A product instance manager wraps a running Ansys product session. The @create_instance decorator starts the product and binds the session to a name, while the @instance decorator injects that same session into any other transaction method.

Key concept — Instance lifecycle

A product instance outlives the transaction method that created it. It stays available to every transaction method of the solution until a transaction explicitly shuts it down.

Key concept — Event stream

A transaction method reports progress with self.transaction.raise_event. Each event is published on a named stream, so the frontend can react while the transaction is still running.

Key concept — Termination event

Passing enable_termination_event=True to @transaction makes SAF automatically raise a structured MethodState event on a stream named after the transaction once it completes or fails, so the frontend can track its outcome without parsing log text. The persisted state can also be read at any time with step.get_long_running_method_state(transaction_name).

Define the step fields

instance_created tracks whether a Mechanical instance is currently running, and output_handle stores the downloaded solve.out file.

solution/instance_management/mechanical_step.py

instance_created: bool = False
version: str = "252"
output_handle: EntityHandle = NO_ENTITY

Define the step model

• The launch_mechanical long-running transaction is decorated with @create_instance to create an instance of the Mechanical product manager, MechanicalManager, while the shutdown_mechanical transaction is decorated with @instance to reuse and shut down the existing instance.

• upload_example_file_to_mechanical and initialize_variable_workflow are decorated with @instance and upload the example geometry to the instance, then set up the part_file_path variable used by the script.

• run_script is decorated with @instance and calls the Mechanical API to run a custom Python script that solves a static structural analysis.

• download_output_file is decorated with @instance and downloads the solve.out file produced by the solve, storing it in an EntityHandle field.

• The launch_mechanical, run_script, download_output_file, and shutdown_mechanical transactions pass enable_termination_event=True so the frontend can track their completion status.

solution/instance_management/mechanical_step.py

    @transaction(self=StepSpec(download=["version"], upload=["instance_created"]), enable_termination_event=True)
    @create_instance("mechanical_instance", MechanicalManager)
    @long_running
    def launch_mechanical(self, mechanical_instance: MechanicalManager) -> None:
        """Launch the Mechanical instance."""
        self.transaction.raise_event(
            message="Initializing Mechanical instance.",
            stream_name="mechanical-output-stream",
        )
        try:
            mechanical_instance.initialize(version=self.version)
        except Exception as e:
            self.transaction.raise_event(
                message=f"Mechanical initialization failed: {e}",
                stream_name="mechanical-output-stream",
            )
            raise
        self.transaction.raise_event(message="Mechanical initialized.", stream_name="mechanical-output-stream")
        self.instance_created = True

solution/instance_management/mechanical_step.py

    @transaction()
    @instance("mechanical_instance")
    def upload_example_file_to_mechanical(self, mechanical_instance: MechanicalManager) -> None:
        """Upload the example file to Mechanical instance."""
        self.transaction.raise_event(message="Uploading file.", stream_name="mechanical-output-stream")

        try:
            mechanical = mechanical_instance.instance
            asset_handle = self.transaction.get_asset_entity_handle("example_01_geometry.agdb")
            geometry_path = self.storage_scope.get_cached(asset_handle)
            mechanical.upload(file_name=geometry_path)  # type: ignore
        except Exception as e:
            self.transaction.raise_event(message=f"File upload failed: {e}", stream_name="mechanical-output-stream")
            raise

solution/instance_management/mechanical_step.py

    @transaction()
    @instance("mechanical_instance")
    def initialize_variable_workflow(self, mechanical_instance: MechanicalManager) -> None:
        """Initialize the variable workflow in Mechanical instance."""
        self.transaction.raise_event(
            message="Starting to initialize variables.",
            stream_name="mechanical-output-stream",
        )

        mechanical = mechanical_instance.instance

        try:
            self.transaction.raise_event(
                message="Initializing variable workflow in Mechanical instance.",
                stream_name="mechanical-output-stream",
            )
            asset_handle = self.transaction.get_asset_entity_handle("example_01_geometry.agdb")
            geometry_path = mechanical_instance.storage_scope.get_cached(asset_handle)
            self.transaction.raise_event(
                message=f"Geometry path: {geometry_path}",
                stream_name="mechanical-output-stream",
            )
            project_directory = mechanical.project_directory  # type: ignore
            self.transaction.raise_event(
                message=f"Project directory: {project_directory}",
                stream_name="mechanical-output-stream",
            )

            # Build the path relative to project directory.
            combined_path = str(Path(project_directory) / geometry_path.name)  # type: ignore
            path_in_mechanical = combined_path.replace("\\", "\\\\")
            mechanical.run_python_script(f"part_file_path='{path_in_mechanical}'")
        except Exception as e:
            self.transaction.raise_event(
                message=f"Mechanical variables initialized failed: {e}",
                stream_name="mechanical-output-stream",
            )
            raise

solution/instance_management/mechanical_step.py

    @transaction(enable_termination_event=True)
    @instance("mechanical_instance")
    @long_running
    def run_script(self, mechanical_instance: MechanicalManager) -> None:
        """Run a script in the Mechanical instance."""
        mechanical = mechanical_instance.instance

        self.transaction.raise_event(
            message="Running script in Mechanical instance.",
            stream_name="mechanical-output-stream",
        )

        try:
            # Run the script
            output = mechanical.run_python_script(
                """
import json

# Section 1: Read geometry information
geometry_import_group_11 = Model.GeometryImportGroup
geometry_import_19 = geometry_import_group_11.AddGeometryImport()

geometry_import_19_format = Ansys.Mechanical.DataModel.Enums.GeometryImportPreference.\
    Format.Automatic
geometry_import_19_preferences = Ansys.ACT.Mechanical.Utilities.GeometryImportPreferences()
geometry_import_19_preferences.ProcessNamedSelections = True
geometry_import_19_preferences.ProcessCoordinateSystems = True

geometry_import_19.Import(part_file_path, geometry_import_19_format, geometry_import_19_preferences)

Model.AddStaticStructuralAnalysis()
STAT_STRUC = Model.Analyses[0]
CS_GRP = Model.CoordinateSystems
ANALYSIS_SETTINGS = STAT_STRUC.Children[0]
SOLN= STAT_STRUC.Solution

# Section 2: Set up the unit system.

ExtAPI.Application.ActiveUnitSystem = MechanicalUnitSystem.StandardMKS
ExtAPI.Application.ActiveAngleUnit = AngleUnitType.Radian

# Section 3: Define named selection and coordinate system.

NS1 = Model.NamedSelections.Children[0]
NS2 = Model.NamedSelections.Children[1]
NS3 = Model.NamedSelections.Children[2]
NS4 = Model.NamedSelections.Children[3]
GCS = CS_GRP.Children[0]
LCS1 = CS_GRP.Children[1]

# Section 4: Define remote point.

RMPT_GRP = Model.RemotePoints
RMPT_1 = RMPT_GRP.AddRemotePoint()
RMPT_1.Location = NS1
RMPT_1.XCoordinate=Quantity("7 [m]")
RMPT_1.YCoordinate=Quantity("0 [m]")
RMPT_1.ZCoordinate=Quantity("0 [m]")

#  Section 5: Define mesh settings.

MSH = Model.Mesh
MSH.ElementSize =Quantity("0.5 [m]")
MSH.GenerateMesh()

#  Section 6: Define boundary conditions.

# Insert fixed support.
FIX_SUP = STAT_STRUC.AddFixedSupport()
FIX_SUP.Location = NS2

# Insert frictionless support.
FRIC_SUP = STAT_STRUC.AddFrictionlessSupport()
FRIC_SUP.Location = NS3

#  Section 7: Define remote force.

REM_FRC1 = STAT_STRUC.AddRemoteForce()
REM_FRC1.Location = RMPT_1
REM_FRC1.DefineBy =LoadDefineBy.Components
REM_FRC1.XComponent.Output.DiscreteValues = [Quantity("1e10 [N]")]

#  Section 8: Define thermal condition.

THERM_COND = STAT_STRUC.AddThermalCondition()
THERM_COND.Location = NS4
THERM_COND.Magnitude.Output.DefinitionType=VariableDefinitionType.Formula
THERM_COND.Magnitude.Output.Formula="50*(20+z)"
THERM_COND.XYZFunctionCoordinateSystem=LCS1
THERM_COND.RangeMinimum=Quantity("-20 [m]")
THERM_COND.RangeMaximum=Quantity("1 [m]")

#  Section 9: Insert directional deformation.

DIR_DEF = STAT_STRUC.Solution.AddDirectionalDeformation()
DIR_DEF.Location = NS1
DIR_DEF.NormalOrientation =NormalOrientationType.XAxis

# Section 10: Add total deformation and force reaction probe.

TOT_DEF = STAT_STRUC.Solution.AddTotalDeformation()

# Add force reaction.
FRC_REAC_PROBE = STAT_STRUC.Solution.AddForceReaction()
FRC_REAC_PROBE.BoundaryConditionSelection = FIX_SUP
FRC_REAC_PROBE.ResultSelection =ProbeDisplayFilter.XAxis

# Section 11: Solve and get the results.

# Solve static analysis.
STAT_STRUC.Solution.Solve(True)

dir_deformation_details = {
"Minimum": str(DIR_DEF.Minimum),
"Maximum": str(DIR_DEF.Maximum),
"Average": str(DIR_DEF.Average),
}

json.dumps(dir_deformation_details)""",
            )
        except Exception as e:
            self.transaction.raise_event(message=f"Run Script failed: {e}", stream_name="mechanical-output-stream")
            raise

        self.transaction.raise_event(message=f"Run Script succeeded. {output}", stream_name="mechanical-output-stream")

solution/instance_management/mechanical_step.py

    @transaction(self=StepSpec(upload=["output_handle"]), enable_termination_event=True)
    @instance("mechanical_instance")
    @long_running
    def download_output_file(self, mechanical_instance: MechanicalManager) -> None:
        """Download the output file from Mechanical instance."""
        mechanical = mechanical_instance.instance

        self.transaction.raise_event(
            message="Downloading file to the current working directory.",
            stream_name="mechanical-output-stream",
        )

        try:
            solve_out_path = ""
            n = 0
            nmax = 10
            while not solve_out_path and n < nmax:
                for file_path in mechanical.list_files():  # type: ignore
                    if file_path.find("solve.out") != -1:  # type: ignore
                        solve_out_path = file_path  # type: ignore
                        break
                n += 1
                sleep(0.1)
            if not solve_out_path:
                raise RuntimeError("solve.out not found.")

            downloaded_files = mechanical.download(  # type: ignore
                solve_out_path, target_dir=self.storage_scope.get_storage_root()
            )
            self.output_handle = self.storage_scope.store(downloaded_files[0])  # type: ignore
        except Exception as e:
            self.transaction.raise_event(message=f"File download failed: {e}", stream_name="mechanical-output-stream")
            raise

        self.transaction.raise_event(
            message=f"File downloaded to {downloaded_files[0]}.",
            stream_name="mechanical-output-stream",
        )

solution/instance_management/mechanical_step.py

    @transaction(self=StepSpec(upload=["instance_created"]), enable_termination_event=True)
    @instance("mechanical_instance")
    @long_running
    def shutdown_mechanical(self, mechanical_instance: MechanicalManager) -> None:
        """Close the Mechanical instance."""
        self.transaction.raise_event(
            message="Starting to shutdown the instance.",
            stream_name="mechanical-output-stream",
        )

        try:
            mechanical_instance.shutdown()
        except Exception as e:
            self.transaction.raise_event(
                message=f"Mechanical shutdown failed: {e}",
                stream_name="mechanical-output-stream",
            )
            raise
        self.transaction.raise_event(
            message="Mechanical instance shutdown complete.",
            stream_name="mechanical-output-stream",
        )
        self.instance_created = False

Frontend

Expose the solution definition in the UI.

Key concept — Callback

A callback is a Dash-decorated function that fires in response to a UI event. In a SAF solution, callbacks reach the backend through project.steps.<step_name>, read or write fields, and invoke transaction methods. No manual HTTP calls are needed.

Key concept — Event listener

DashClient.create_event_listener subscribes a component to an event stream raised by the backend. Each message received by the listener triggers a callback, so the console logs, the button states, and the notifications stay in sync with the running transaction.

Compute the initial control states

• get_mechanical_page_controls_with_default is a helper, shared by every instance management page, that returns the default disabled/loading state of each button together with the name of the transaction it triggers.

• initialize_mechanical_controls starts from those defaults, then looks up the persisted state of each transaction with step.get_long_running_method_state() so the buttons show the correct enabled/disabled/loading state even after a page reload.

ui/pages/instance_management/mechanical_instance_page.py

def initialize_mechanical_controls(project: ExamplesSolution) -> dict[str, dict[str, Any]] | Any:
    """Initialize the state of controls on the Mechanical instance management page."""
    controls = get_mechanical_page_controls_with_default()

    for button in controls.keys():
        controls[button]["disabled"] = True
        controls[button]["loading"] = False

    transactions = [
        "launch_mechanical",
        "shutdown_mechanical",
        "run_script",
        "download_output_file",
    ]

    step = project.steps.mechanical_step
    transaction_states = {t: step.get_long_running_method_state(t).status.value for t in transactions}

    if any(state == "running" for state in transaction_states.values()):
        for button, config in controls.items():
            if transaction_states[config["transaction"]] == "running":
                controls[button]["disabled"] = True
                break
    elif step.instance_created:
        if transaction_states["run_script"] == "completed" or transaction_states["download_output_file"] == "completed":
            controls["run_script"]["disabled"] = False
            controls["download_output_file"]["disabled"] = False
            controls["shutdown_mechanical"]["disabled"] = False
        elif transaction_states["launch_mechanical"] == "completed":
            controls["run_script"]["disabled"] = False
            controls["shutdown_mechanical"]["disabled"] = False
    else:
        logger.info("No Mechanical instance detected, initializing page with default state")
        controls["launch_mechanical"]["disabled"] = False
    return controls

Define the step layout

The layout builds a controls card, with icon buttons to launch and shut down the instance and buttons to run the script and download the output file, and a logs card, using the initial control states computed above.

ui/pages/instance_management/mechanical_instance_page.py

def layout(project: ExamplesSolution) -> html.Div:
    """Layout for the Mechanical Instance Manager page."""
    controls = initialize_mechanical_controls(project)

    controls_card = dmc.Card(
        [
            dmc.CardSection(
                dmc.Group(
                    children=[dmc.Text("Controls", fw=500, style={"font-size": "17px"})],
                    justify="space-between",
                ),
                withBorder=True,
                inheritPadding=True,
                py="xs",
            ),
            dmc.Space(h=20),
            dmc.Group(
                [
                    dmc.Tooltip(
                        dmc.ActionIcon(
                            DashIconify(icon="streamline:startup-solid", width=30),
                            id="launch-mechanical-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["launch_mechanical"]["disabled"],
                            loading=controls["launch_mechanical"]["loading"],
                        ),
                        label="Launch Mechanical",
                        position="top",
                    ),
                    dmc.Tooltip(
                        dmc.ActionIcon(
                            DashIconify(icon="mdi:shutdown", width=30),
                            id="shutdown-mechanical-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["shutdown_mechanical"]["disabled"],
                            loading=controls["shutdown_mechanical"]["loading"],
                        ),
                        label="Shutdown Mechanical",
                        position="top",
                    ),
                ],
                gap="md",
                justify="center",
            ),
            dmc.Space(h=20),
            dmc.Divider(variant="solid"),
            dmc.Space(h=20),
            dmc.Stack(
                [
                    dmc.Button(
                        "Run Script",
                        id="run-script-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="codicon:run-all"),
                        disabled=controls["run_script"]["disabled"],
                        loading=controls["run_script"]["loading"],
                        style={"width": "70%", "font-size": "15px"},
                    ),
                    dmc.Button(
                        "Download Output File",
                        id="download-output-file-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="material-symbols:download"),
                        disabled=controls["download_output_file"]["disabled"],
                        loading=controls["download_output_file"]["loading"],
                        style={"width": "70%", "font-size": "15px"},
                    ),
                ],
                align="center",
            ),
        ],
        withBorder=True,
        shadow="sm",
        radius="md",
    )

    logs_container = dmc.Card(
        [
            dmc.CardSection(
                dmc.Group(
                    children=[
                        dmc.Text("Logs", fw=500, style={"font-size": "17px"}),
                        dmc.Tooltip(
                            dmc.ActionIcon(
                                DashIconify(icon="mdi:delete-sweep", width=24),
                                id="clear-logs-button",
                                color="gray",
                                variant="transparent",
                            ),
                            label="Clear Logs",
                            position="left",
                        ),
                    ],
                    justify="space-between",
                ),
                withBorder=True,
                inheritPadding=True,
                py="xs",
            ),
            dmc.Space(h=20),
            html.Div(
                html.Pre(
                    id="mechanical-console-logs",
                    style={
                        "whiteSpace": "pre-wrap",
                        "wordBreak": "break-all",
                        "fontSize": "14px",
                        "height": "100%",
                        "overflowY": "auto",
                        "margin": "0",
                    },
                ),
                style={
                    "height": "600px",
                    "width": "100%",
                    "overflowY": "scroll",
                },
            ),
        ],
        withBorder=True,
        shadow="sm",
        radius="md",
    )

    return html.Div(
        [
            html.H1(
                "Mechanical Instance Manager",
                className="display-3",
                style={"font-size": "40px", "font-weight": "bold"},
            ),
            dmc.Blockquote(
                "This example demonstrates how to leverage the instance management API to control Ansys Mechanical.\
                Click the Launch button to\
                start the instance. A transaction method will start Mechanical which can be used across all transaction\
                methods. Run Mechanical operations with the Run Script and\
                Download Output File buttons. Close Mechanical using the Shutdown button.",
                icon=DashIconify(icon="material-symbols:info", width=30),
                style={"font-size": "18px", "fontStyle": "italic"},
            ),
            dmc.Space(h=20),
            dmc.Alert(
                dmc.Text(
                    [
                        "⚠️ This example requires the following prerequisites:\n",
                        "- ",
                        dmc.Mark("Ansys Mechanical 2025 R2 Service Pack 4 (25R2 SP4) or later"),
                        " installed and licensed,\n",
                        "- ",
                        dmc.Mark("ansys-saf-pim-light-server package 0.3 or later"),
                        " installed in the Python environment or ",
                        dmc.Mark("optiSLang 2025 R2 or later"),
                        " installed,\n",
                    ],
                    style={"whiteSpace": "pre-line"},
                    size="md",
                ),
                title="Prerequisites",
                color="yellow",
            ),
            dmc.Space(h=20),
            dmc.Grid(
                [
                    dmc.GridCol(
                        controls_card,
                        span=3,
                    ),
                    dmc.GridCol(logs_container, span=9),
                ],
                grow=True,
                gutter="xs",
            ),
            html.Br(),
            html.Br(),
        ],
        style={"paddingLeft": "20px"},
    )

Mount the event listeners

• mechanical-output-listener subscribes to the free-form progress messages raised on the mechanical-output-stream by the step model.

• The other four listeners each subscribe to the termination event of one long-running transaction: launch_mechanical, shutdown_mechanical, run_script, and download_output_file.

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("mechanical-instance-event-listeners-container", "children"),
    Input("url", "pathname"),
)
def mount_event_listeners(project: ExamplesSolution) -> list[dict[str, Any]] | Any:
    """Mount Mechanical instance event listeners in the persistent main layout."""
    logger.info("Mounting Mechanical instance management event listeners")
    step = project.steps.mechanical_step
    return [
        DashClient.create_event_listener(step, id="mechanical-output-listener", stream_name="mechanical-output-stream"),
        DashClient.create_event_listener(step, id="launch-mechanical-listener", stream_name="launch-mechanical"),
        DashClient.create_event_listener(step, id="run-script-listener", stream_name="run-script"),
        DashClient.create_event_listener(step, id="download-output-file-listener", stream_name="download-output-file"),
        DashClient.create_event_listener(step, id="shutdown-mechanical-listener", stream_name="shutdown-mechanical"),
    ]

Trigger a transaction from a button click

• When the Launch Mechanical button is clicked, the callback starts the launch_mechanical long-running transaction and shows a loading notification.

• When the Run Script button is clicked, the _run_script helper chains upload_example_file_to_mechanical, initialize_variable_workflow, and run_script on the step, so the geometry is uploaded and the workflow variables are set before the long-running script transaction starts.

• The download_output_file and shutdown_mechanical callbacks follow the simpler pattern: each one calls the matching transaction on the step and shows a loading notification.

ui/pages/instance_management/mechanical_instance_page.py

def _run_script(project: ExamplesSolution) -> None:
    """Handle all operations related to the Mechanical step.run_script."""
    step = project.steps.mechanical_step
    step.upload_example_file_to_mechanical()
    step.initialize_variable_workflow()
    step.run_script()

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("run-script-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def run_script(n_clicks: int, project: ExamplesSolution) -> list[dict[str, Any]] | Any:
    """Run a script in the Mechanical instance."""
    notification = no_update

    if ctx.triggered_id == "run-script-button" and n_clicks:
        logger.info("Run Script button clicked, running script in Mechanical instance")

        _run_script(project)

        notification = [
            dict(
                title="Info",
                id="run-script-notification",
                action="show",
                message="Running script in Mechanical instance... Check the logs for progress.",
                autoClose=False,
                loading=True,
                color="blue",
                withCloseButton=False,
            )
        ]

    return notification

Keep the button states in sync

• sync_controls_on_clicks optimistically disables the buttons as soon as one is clicked, so the user cannot trigger two transactions at once while the backend catches up.

• sync_controls_on_backend_events reconciles the button states once a termination event is received, re-enabling the buttons that make sense for the new instance state.

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("launch-mechanical-button", "disabled", allow_duplicate=True),
    Output("launch-mechanical-button", "loading", allow_duplicate=True),
    Output("shutdown-mechanical-button", "disabled", allow_duplicate=True),
    Output("shutdown-mechanical-button", "loading", allow_duplicate=True),
    Output("run-script-button", "disabled", allow_duplicate=True),
    Output("run-script-button", "loading", allow_duplicate=True),
    Output("download-output-file-button", "disabled", allow_duplicate=True),
    Output("download-output-file-button", "loading", allow_duplicate=True),
    Input("launch-mechanical-button", "n_clicks"),
    Input("shutdown-mechanical-button", "n_clicks"),
    Input("run-script-button", "n_clicks"),
    Input("download-output-file-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def sync_controls_on_clicks(
    launch_mechanical_clicks: int,
    shutdown_mechanical_clicks: int,
    run_script_clicks: int,
    download_output_file_clicks: int,
    project: ExamplesSolution,
) -> tuple[bool, bool, bool, bool, bool, bool, bool, bool]:
    """Sync the state of the control buttons based on user interactions."""
    step = project.steps.mechanical_step
    triggered_id = ctx.triggered_id
    controls = get_mechanical_page_controls_with_default()

    if triggered_id == "launch-mechanical-button" and launch_mechanical_clicks and not step.instance_created:
        logger.info("Launch button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "launch_mechanical":
                controls[button]["loading"] = True
    elif triggered_id == "shutdown-mechanical-button" and shutdown_mechanical_clicks and step.instance_created:
        logger.info("Shutdown button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "shutdown_mechanical":
                controls[button]["loading"] = True
    elif triggered_id == "run-script-button" and run_script_clicks and step.instance_created:
        logger.info("Run Script button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "run_script":
                controls[button]["loading"] = True
    elif triggered_id == "download-output-file-button" and download_output_file_clicks and step.instance_created:
        logger.info("Download Output File button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "download_output_file":
                controls[button]["loading"] = True

    return (
        controls["launch_mechanical"]["disabled"],
        controls["launch_mechanical"]["loading"],
        controls["shutdown_mechanical"]["disabled"],
        controls["shutdown_mechanical"]["loading"],
        controls["run_script"]["disabled"],
        controls["run_script"]["loading"],
        controls["download_output_file"]["disabled"],
        controls["download_output_file"]["loading"],
    )

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("launch-mechanical-button", "disabled", allow_duplicate=True),
    Output("launch-mechanical-button", "loading", allow_duplicate=True),
    Output("shutdown-mechanical-button", "disabled", allow_duplicate=True),
    Output("shutdown-mechanical-button", "loading", allow_duplicate=True),
    Output("run-script-button", "disabled", allow_duplicate=True),
    Output("run-script-button", "loading", allow_duplicate=True),
    Output("download-output-file-button", "disabled", allow_duplicate=True),
    Output("download-output-file-button", "loading", allow_duplicate=True),
    Input("launch-mechanical-listener", "message"),
    Input("shutdown-mechanical-listener", "message"),
    Input("run-script-listener", "message"),
    Input("download-output-file-listener", "message"),
    prevent_initial_call=True,
)
def sync_controls_on_backend_events(
    launch_mechanical_message: dict[str, Any],
    shutdown_mechanical_message: dict[str, Any],
    run_script_message: dict[str, Any],
    download_output_file_message: dict[str, Any],
) -> tuple[bool, bool, bool, bool, bool, bool, bool, bool]:
    """Sync the state of the control buttons based on websocket messages from the backend indicating
    Mechanical instance state changes."""
    triggered_id = ctx.triggered_id
    controls = get_mechanical_page_controls_with_default()

    if triggered_id == "launch-mechanical-listener" and launch_mechanical_message:
        logger.info("Launch Mechanical listener triggered, updating controls")
        method_state = MethodState.model_validate_json(launch_mechanical_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button == "launch_mechanical":
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button in ["shutdown_mechanical", "run_script"]:
                    controls[button]["disabled"] = False
        else:
            controls["launch_mechanical"]["disabled"] = False
            controls["launch_mechanical"]["loading"] = False
    elif triggered_id == "shutdown-mechanical-listener" and shutdown_mechanical_message:
        logger.info("Shutdown Mechanical listener triggered, updating controls")
        method_state = MethodState.model_validate_json(shutdown_mechanical_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button in ["shutdown_mechanical", "run_script", "download_output_file"]:
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button == "launch_mechanical":
                    controls[button]["disabled"] = False
                    controls[button]["loading"] = False
        else:
            controls["shutdown_mechanical"]["disabled"] = False
            controls["shutdown_mechanical"]["loading"] = False
    elif triggered_id == "run-script-listener" and run_script_message:
        logger.info("Run Script listener triggered, updating controls")
        method_state = MethodState.model_validate_json(run_script_message["data"])
        if method_state.status.value == "completed":
            controls["run_script"]["disabled"] = False
            controls["run_script"]["loading"] = False
            controls["download_output_file"]["disabled"] = False
        else:
            controls["run_script"]["disabled"] = False
            controls["run_script"]["loading"] = False
        controls["shutdown_mechanical"]["disabled"] = False
    elif triggered_id == "download-output-file-listener" and download_output_file_message:
        logger.info("Download Output File listener triggered, updating controls")
        controls["download_output_file"]["disabled"] = False
        controls["download_output_file"]["loading"] = False
        controls["shutdown_mechanical"]["disabled"] = False
        controls["run_script"]["disabled"] = False

    return (
        controls["launch_mechanical"]["disabled"],
        controls["launch_mechanical"]["loading"],
        controls["shutdown_mechanical"]["disabled"],
        controls["shutdown_mechanical"]["loading"],
        controls["run_script"]["disabled"],
        controls["run_script"]["loading"],
        controls["download_output_file"]["disabled"],
        controls["download_output_file"]["loading"],
    )

Define the notification system.

• sync_notifications_on_backend_events reacts to the same termination events and turns each MethodState into a success or error notification through the shared handle_method_event helper.

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-mechanical-listener", "message"),
    Input("shutdown-mechanical-listener", "message"),
    Input("run-script-listener", "message"),
    Input("download-output-file-listener", "message"),
    prevent_initial_call=True,
)
def sync_notifications_on_backend_events(
    launch_mechanical_message: dict[str, Any],
    shutdown_mechanical_message: dict[str, Any],
    run_script_message: dict[str, Any],
    download_output_file_message: dict[str, Any],
) -> list[dict[str, Any]] | Any:
    """Build and return notifications from backend websocket messages about Mechanical method state changes."""
    triggered_id = ctx.triggered_id
    notification = no_update

    if triggered_id == "launch-mechanical-listener" and launch_mechanical_message:
        logger.info("Launch Mechanical listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(launch_mechanical_message["data"])
        notification = handle_method_event(
            method_state,
            "launch-mechanical-notification",
            "Mechanical instance launched successfully!",
            "Mechanical initialization failed. Please check the logs.",
        )
    elif triggered_id == "shutdown-mechanical-listener" and shutdown_mechanical_message:
        logger.info("Shutdown Mechanical listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(shutdown_mechanical_message["data"])
        notification = handle_method_event(
            method_state,
            "shutdown-mechanical-notification",
            "Mechanical instance shutdown successfully!",
            "Failed to shutdown Mechanical instance. Please check the logs.",
        )
    elif triggered_id == "run-script-listener" and run_script_message:
        logger.info("Run Script listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(run_script_message["data"])
        notification = handle_method_event(
            method_state,
            "run-script-notification",
            "Script ran successfully!",
            "Script execution failed. Please check the logs.",
        )
    elif triggered_id == "download-output-file-listener" and download_output_file_message:
        logger.info("Download output file listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(download_output_file_message["data"])
        notification = handle_method_event(
            method_state,
            "download-output-file-notification",
            "Output file downloaded successfully!",
            "Output file download failed. Please check the logs.",
        )

    return notification

ui/helpers.py

def handle_method_event(
    method_state: MethodState,
    notification_id: str,
    success_msg: str,
    error_msg: str,
    success_auto_close: int | bool = 5000,
    error_auto_close: int | bool = False,
) -> list[dict[str, Any]]:
    """Process a backend method event and update controls/notification accordingly.

    Parameters
    ----------
    method_state : MethodState
        The state of the backend method.
    notification_id : str
        The ID of the notification to update.
    success_msg : str
        The message to display if the method completed successfully.
    error_msg : str
        The message to display if the method failed.
    success_auto_close : int | bool, optional
        Time in milliseconds after which the success notification should auto-close,
        or False to disable auto-close. Default is 5000 (5 seconds).
    error_auto_close : int | bool, optional
        Time in milliseconds after which the error notification should auto-close,
        or False to disable auto-close. Default is False (no auto-close).
    """
    notification = no_update

    if method_state.status.value == "completed":
        notification = [
            dict(
                title="Success",
                id=notification_id,
                action="update",
                message=success_msg,
                color="green",
                autoClose=success_auto_close,
                withCloseButton=True,
                loading=False,
            )
        ]
    elif method_state.status.value == "failed":
        notification = [
            dict(
                title="Error",
                id=notification_id,
                action="update",
                message=error_msg,
                color="red",
                autoClose=error_auto_close,
                withCloseButton=True,
                loading=False,
            )
        ]

    return notification

Define the logs system.

• store_outputs appends every message received on the mechanical-output-listener stream to a dcc.Store, display_output mirrors that store into the console logs container, and clear_console_logs resets it.

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("mechanical-logs-store", "data", allow_duplicate=True),
    Input("mechanical-output-listener", "message"),
    State("mechanical-logs-store", "data"),
    prevent_initial_call=True,
)
def store_outputs(message: dict[str, Any], current_logs: str) -> str:
    """Store Mechanical output."""
    if message:
        new_content = message["data"].strip('"').replace("\\n", "\n")
        combined = (current_logs or "") + "\n" + new_content
        return combined
    return current_logs

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("mechanical-console-logs", "children", allow_duplicate=True),
    Input("mechanical-logs-store", "data"),
)
def display_output(current_logs: str) -> str:
    """Display Mechanical output."""
    return current_logs

ui/pages/instance_management/mechanical_instance_page.py

@callback(
    Output("mechanical-console-logs", "children"),
    Input("clear-logs-button", "n_clicks"),
    prevent_initial_call=True,
)
def clear_console_logs(n_clicks: int) -> str:
    """Clear the console logs."""
    return ""

Now that your implementation is complete, continue to the Testing section.

Testing

Finally, test your implementation to confirm it works as expected.

Run the solution and compare your results with the results shown in the Feature highlight section.