MAPDL Product Instance Manager

Objective

Drive an Ansys Mechanical APDL (MAPDL) product from a solution. Start the product instance in a long-running transaction, call the MAPDL API to build, solve, and postprocess a 2D solenoid actuator model, stream the solver output to the UI, then shut the instance down.

Source code for this example is in the example solution.

Feature highlight

PyMAPDL provides a robust Python interface to Ansys Mechanical APDL, allowing users to automate finite element analyses, integrate seamlessly with Python data processing libraries, and optimize workflows across structural, thermal, and coupled physics simulations.

This example shows how to create a product instance manager for PyMAPDL using SAF. The product instance manager enables management of the MAPDL 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 MAPDL API to set up the finite element model, solve it, and postprocess the results.

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

• Shut down the product instance from a dedicated transaction method.

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

Prerequisites

Note

This example requires the MAPDL 2025 R2 SP4 (25R2 SP4) product to be installed on your machine. To work with a MAPDL product instance, be sure to install the core-pim and instance-management-mapdl 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-mapdl"]}

This will install the supported version of ansys-mapdl-core to control the MAPDL 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.

Coding

To create a product instance manager for MAPDL 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 MAPDL instance is currently running.

solution/instance_management/mapdl_step.py

instance_created: bool = False
version: str = "252"
nodal_values: list[float] = []
solved: bool = False

Define the step model

• The launch_mapdl long-running transaction is decorated with @create_instance to create an instance of the MAPDL product manager, MapdlManager, and sets up the finite element model of the 2D solenoid actuator, while the shutdown_mapdl transaction is decorated with @instance to reuse and shut down the existing instance.

• The solve_model and postprocessing transactions are decorated with @instance to indicate they operate on the existing product instance, and call the MAPDL API to solve the model and postprocess the results, respectively.

• All four long-running transactions pass enable_termination_event=True so the frontend can track their completion status.

solution/instance_management/mapdl_step.py

    @transaction(self=StepSpec(download=["version"], upload=["instance_created"]), enable_termination_event=True)
    @create_instance("mapdl_instance", MapdlManager)
    @long_running
    def launch_mapdl(self, mapdl_instance: MapdlManager) -> None:
        """Launch a MAPDL instance with the specified version."""
        self.transaction.raise_event(message="Initializing MAPDL instance.", stream_name="mapdl-output-stream")
        try:
            mapdl_instance.initialize(version=self.version)

            mapdl = mapdl_instance.instance

            mapdl.clear()  # type: ignore
            mapdl.prep7()  # type: ignore
            mapdl.title("2-D Solenoid Actuator Static Analysis")  # type: ignore

            # Set up the FE model
            mapdl.et(1, "PLANE233")  # Define PLANE233 as element type  # type: ignore
            mapdl.keyopt(1, 3, 1)  # Use axisymmetric analysis option  # type: ignore
            mapdl.keyopt(1, 7, 1)  # Condense forces at the corner nodes  # type: ignore

            # Set material properties
            mapdl.mp("MURX", 1, 1)  # Define material properties (permeability), Air  # type: ignore
            mapdl.mp("MURX", 2, 1000)  # Permeability of backiron  # type: ignore
            mapdl.mp("MURX", 3, 1)  # Permeability of coil  # type: ignore
            mapdl.mp("MURX", 4, 2000)  # Permeability of armature  # type: ignore

            # Set parameters
            n_turns = 650  # Number of coil turns
            i_current = 1.0  # Current per turn
            ta = 0.75  # Model dimensions (centimeters)
            tb = 0.75
            tc = 0.50
            td = 0.75
            wc = 1
            hc = 2
            gap = 0.25
            space = 0.25
            ws = wc + 2 * space
            hs = hc + 0.75
            w = ta + ws + tc
            hb = tb + hs
            h = hb + gap + td
            acoil = wc * hc  # Cross-section area of coil (cm**2)
            jdens = n_turns * i_current / acoil  # Current density (A/cm**2)

            smart_size = 4  # Smart Size Level for Meshing

            # Create geometry
            mapdl.rectng(0, w, 0, tb)  # Create rectangular areas  # type: ignore
            mapdl.rectng(0, w, tb, hb)  # type: ignore
            mapdl.rectng(ta, ta + ws, 0, h)  # type: ignore
            mapdl.rectng(ta + space, ta + space + wc, tb + space, tb + space + hc)  # type: ignore
            mapdl.aovlap("ALL")  # type: ignore
            mapdl.rectng(0, w, 0, hb + gap)  # type: ignore
            mapdl.rectng(0, w, 0, h)  # type: ignore
            mapdl.aovlap("ALL")  # type: ignore
            mapdl.numcmp("AREA")  # Compress out unused area numbers  # type: ignore

            # Mesh
            mapdl.asel("S", "AREA", "", 2)  # Assign attributes to coil  # type: ignore
            mapdl.aatt(3, 1, 1, 0)  # type: ignore

            mapdl.asel("S", "AREA", "", 1)  # Assign attributes to armature  # type: ignore
            mapdl.asel("A", "AREA", "", 12, 13)  # type: ignore
            mapdl.aatt(4, 1, 1)  # type: ignore

            mapdl.asel("S", "AREA", "", 3, 5)  # Assign attributes to backiron  # type: ignore
            mapdl.asel("A", "AREA", "", 7, 8)  # type: ignore
            mapdl.aatt(2, 1, 1, 0)  # type: ignore

            mapdl.pnum("MAT", 1)  # Turn material numbers on  # type: ignore
            mapdl.allsel("ALL")  # type: ignore

            mapdl.smrtsize(smart_size)  # Set smart size meshing  # type: ignore
            mapdl.amesh("ALL")  # Mesh all areas  # type: ignore

            # Scale mesh to meters
            mapdl.esel("S", "MAT", "", 4)  # Select armature elements  # type: ignore
            mapdl.cm("ARM", "ELEM")  # Define armature as a component  # type: ignore
            mapdl.allsel("ALL")  # type: ignore
            mapdl.arscale(na1="all", rx=0.01, ry=0.01, rz=1, imove=1)  # Scale model to MKS (meters)  # type: ignore
            mapdl.finish()  # type: ignore

            # Loads and boundary conditions
            mapdl.slashsolu()  # type: ignore

            # Apply current density (A/m**2)
            mapdl.esel("S", "MAT", "", 3)  # Select coil elements  # type: ignore
            mapdl.bfe("ALL", "JS", 1, "", "", jdens / 0.01**2)  # type: ignore

            mapdl.esel("ALL")  # type: ignore
            mapdl.nsel("EXT")  # Select exterior nodes  # type: ignore
            mapdl.d("ALL", "AZ", 0)  # Set potentials to zero (flux-parallel)  # type: ignore

        except Exception as e:
            self.transaction.raise_event(message=f"MAPDL initialization failed: {e}", stream_name="mapdl-output-stream")
            raise
        self.instance_created = True
        self.transaction.raise_event(message="MAPDL initialized.", stream_name="mapdl-output-stream")

solution/instance_management/mapdl_step.py

    @transaction(self=StepSpec(upload=["solved"]), enable_termination_event=True)
    @instance("mapdl_instance")
    @long_running
    def solve_model(self, mapdl_instance: MapdlManager) -> None:
        """Solve the finite element model using MAPDL."""
        self.transaction.raise_event(message="Solving Model.", stream_name="mapdl-output-stream")
        try:
            mapdl = mapdl_instance.instance

            mapdl.allsel("ALL")  # type: ignore
            mapdl.solve()  # type: ignore
            mapdl.finish()  # type: ignore

            self.solved = True
        except Exception as e:
            self.transaction.raise_event(message=f"Solve model failed: {e}", stream_name="mapdl-output-stream")
            raise
        self.transaction.raise_event(message="Solve model succeeded.", stream_name="mapdl-output-stream")

solution/instance_management/mapdl_step.py

    @transaction(self=StepSpec(upload=["nodal_values"]), enable_termination_event=True)
    @instance("mapdl_instance")
    @long_running
    def postprocessing(self, mapdl_instance: MapdlManager) -> None:
        """Post-process the results and create a plot of the magnetic flux in the X direction."""
        self.transaction.raise_event(message="Postprocessing results.", stream_name="mapdl-output-stream")
        try:
            mapdl = mapdl_instance.instance

            mapdl.post1()  # type: ignore
            mapdl.file("file", "rmg")  # type: ignore
            mapdl.set("last")  # type: ignore

            self.nodal_values = mapdl.post_processing.nodal_values("b", "x").tolist()  # type: ignore

            # Create an MAPDL Power Graphics plot of the X-direction magnetic flux
            mapdl.graphics("power")  # type: ignore
            mapdl.rgb("INDEX", 100, 100, 100, 0)  # type: ignore
            mapdl.rgb("INDEX", 80, 80, 80, 13)  # type: ignore
            mapdl.rgb("INDEX", 60, 60, 60, 14)  # type: ignore
            mapdl.rgb("INDEX", 0, 0, 0, 15)  # type: ignore

            mapdl.edge(1, 1)  # type: ignore

            # Obtain grid and scalar data
            elem_mats = mapdl.mesh.material_type  # type: ignore
            grids = []
            scalars = []
            for mat in np.unique(elem_mats):  # type: ignore
                mapdl.esel("s", "mat", "", mat)  # type: ignore
                mapdl.nsle()  # type: ignore
                grids.append(mapdl.mesh.grid)  # type: ignore
                scalars.append(mapdl.post_processing.nodal_values("b", "x"))  # type: ignore
            mapdl.allsel()  # type: ignore

            # Color map and result plot
            plotter = pv.Plotter()  # type: ignore
            for i, grid in enumerate(grids):  # type: ignore
                plotter.add_mesh(  # type: ignore
                    grid,  # type: ignore
                    scalars=scalars[i],  # type: ignore
                    show_edges=True,
                    cmap=PyMAPDL_cmap,  # type: ignore
                    n_colors=9,
                    scalar_bar_args={
                        "color": "black",
                        "title": "B Flux X",
                        "vertical": False,
                        "n_labels": 10,
                    },
                )

            plotter.set_background(color="white")  # type: ignore
            _ = plotter.camera_position = "xy"
        except Exception as e:
            self.transaction.raise_event(
                message=f"Results postprocessing failed: {e}",
                stream_name="mapdl-output-stream",
            )
            raise
        self.transaction.raise_event(message="Results postprocessing succeeded.", stream_name="mapdl-output-stream")

solution/instance_management/mapdl_step.py

    @transaction(self=StepSpec(upload=["instance_created"]), enable_termination_event=True)
    @instance("mapdl_instance")
    @long_running
    def shutdown_mapdl(self, mapdl_instance: MapdlManager) -> None:
        """Close the MAPDL instance."""
        self.transaction.raise_event(message="Starting to shutdown the instance.", stream_name="mapdl-output-stream")
        try:
            mapdl = mapdl_instance.instance
            mapdl.graphics("FULL")  # Returning to default mode.  # type: ignore

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

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_mapdl_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_mapdl_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/mapdl_instance_page.py

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

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

    transactions = [
        "launch_mapdl",
        "shutdown_mapdl",
        "solve_model",
        "postprocessing",
    ]

    step = project.steps.mapdl_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["solve_model"] == "completed" or transaction_states["postprocessing"] == "completed":
            controls["solve_model"]["disabled"] = False
            controls["postprocess_results"]["disabled"] = False
            controls["shutdown_mapdl"]["disabled"] = False
        elif transaction_states["launch_mapdl"] == "completed":
            controls["solve_model"]["disabled"] = False
            controls["shutdown_mapdl"]["disabled"] = False
    else:
        logger.info("No MAPDL instance detected, initializing page with default state")
        controls["launch_mapdl"]["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 solve the model and postprocess the results, and a logs card, using the initial control states computed above.

ui/pages/instance_management/mapdl_instance_page.py

def layout(project: ExamplesSolution) -> html.Div:
    """Layout for the MAPDL Instance Manager page."""
    controls = initialize_mapdl_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-mapdl-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["launch_mapdl"]["disabled"],
                            loading=controls["launch_mapdl"]["loading"],
                        ),
                        label="Launch MAPDL",
                        position="top",
                    ),
                    dmc.Tooltip(
                        dmc.ActionIcon(
                            DashIconify(icon="mdi:shutdown", width=30),
                            id="shutdown-mapdl-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["shutdown_mapdl"]["disabled"],
                            loading=controls["shutdown_mapdl"]["loading"],
                        ),
                        label="Shutdown MAPDL",
                        position="top",
                    ),
                ],
                gap="md",
                justify="center",
            ),
            dmc.Space(h=20),
            dmc.Divider(variant="solid"),
            dmc.Space(h=20),
            dmc.Stack(
                [
                    dmc.Button(
                        "Solve Model",
                        id="solve-model-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="carbon:result"),
                        disabled=controls["solve_model"]["disabled"],
                        loading=controls["solve_model"]["loading"],
                        style={"width": "70%", "font-size": "15px"},
                    ),
                    dmc.Button(
                        "Postprocess Results",
                        id="postprocess-results-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="uil:process"),
                        disabled=controls["postprocess_results"]["disabled"],
                        loading=controls["postprocess_results"]["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="mapdl-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(
                "MAPDL 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 MAPDL.\
                Click the Launch button to\
                start the instance. A transaction method will start MAPDL which can be used across all transaction\
                methods. Run MAPDL operations with the Solve Model and\
                Postprocess Results buttons. Close MAPDL 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 MAPDL 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 and licensed,\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

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

• The other four listeners each subscribe to the termination event of one long-running transaction: launch_mapdl, shutdown_mapdl, solve_model, and postprocessing.

ui/pages/instance_management/mapdl_instance_page.py

@callback(
    Output("mapdl-instance-event-listeners-container", "children"),
    Input("url", "pathname"),
)
def mount_event_listeners(project: ExamplesSolution) -> list:
    """Mount MAPDL instance event listeners in the persistent main layout."""
    logger.info("Mounting MAPDL instance management event listeners")
    step = project.steps.mapdl_step
    return [
        DashClient.create_event_listener(step, id="mapdl-output-listener", stream_name="mapdl-output-stream"),
        DashClient.create_event_listener(step, id="launch-mapdl-listener", stream_name="launch-mapdl"),
        DashClient.create_event_listener(step, id="solve-model-listener", stream_name="solve-model"),
        DashClient.create_event_listener(step, id="postprocess-results-listener", stream_name="postprocessing"),
        DashClient.create_event_listener(step, id="shutdown-mapdl-listener", stream_name="shutdown-mapdl"),
    ]

Trigger a transaction from a button click

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

• The solve_model, postprocessing, and shutdown_mapdl callbacks follow the same pattern: each one calls the matching transaction on the step and shows a loading notification.

ui/pages/instance_management/mapdl_instance_page.py

@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-mapdl-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def launch_mapdl(n_clicks: int, project: ExamplesSolution) -> list[dict[str, Any]] | str:
    """Launch an instance of Ansys MAPDL."""
    notification = no_update

    if ctx.triggered_id == "launch-mapdl-button" and n_clicks:
        logger.info("Launch button clicked, starting MAPDL instance")
        step = project.steps.mapdl_step
        step.launch_mapdl()

        notification = [
            dict(
                title="Info",
                id="launch-mapdl-notification",
                action="show",
                message="Starting MAPDL instance... Please wait.",
                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/mapdl_instance_page.py

@callback(
    Output("launch-mapdl-button", "disabled", allow_duplicate=True),
    Output("launch-mapdl-button", "loading", allow_duplicate=True),
    Output("shutdown-mapdl-button", "disabled", allow_duplicate=True),
    Output("shutdown-mapdl-button", "loading", allow_duplicate=True),
    Output("solve-model-button", "disabled", allow_duplicate=True),
    Output("solve-model-button", "loading", allow_duplicate=True),
    Output("postprocess-results-button", "disabled", allow_duplicate=True),
    Output("postprocess-results-button", "loading", allow_duplicate=True),
    Input("launch-mapdl-button", "n_clicks"),
    Input("shutdown-mapdl-button", "n_clicks"),
    Input("solve-model-button", "n_clicks"),
    Input("postprocess-results-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def sync_controls_on_clicks(
    launch_mapdl_clicks: int,
    shutdown_mapdl_clicks: int,
    solve_model_clicks: int,
    postprocess_results_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.mapdl_step
    triggered_id = ctx.triggered_id
    controls = get_mapdl_page_controls_with_default()

    if triggered_id == "launch-mapdl-button" and launch_mapdl_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_mapdl":
                controls[button]["loading"] = True
    elif triggered_id == "shutdown-mapdl-button" and shutdown_mapdl_clicks and step.instance_created:
        logger.info("Shutdown button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "shutdown_mapdl":
                controls[button]["loading"] = True
    elif triggered_id == "solve-model-button" and solve_model_clicks and step.instance_created:
        logger.info("Solve Model button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "solve_model":
                controls[button]["loading"] = True
    elif triggered_id == "postprocess-results-button" and postprocess_results_clicks and step.instance_created:
        logger.info("Postprocess Results button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "postprocess_results":
                controls[button]["loading"] = True

    return (
        controls["launch_mapdl"]["disabled"],
        controls["launch_mapdl"]["loading"],
        controls["shutdown_mapdl"]["disabled"],
        controls["shutdown_mapdl"]["loading"],
        controls["solve_model"]["disabled"],
        controls["solve_model"]["loading"],
        controls["postprocess_results"]["disabled"],
        controls["postprocess_results"]["loading"],
    )

ui/pages/instance_management/mapdl_instance_page.py

@callback(
    Output("launch-mapdl-button", "disabled", allow_duplicate=True),
    Output("launch-mapdl-button", "loading", allow_duplicate=True),
    Output("shutdown-mapdl-button", "disabled", allow_duplicate=True),
    Output("shutdown-mapdl-button", "loading", allow_duplicate=True),
    Output("solve-model-button", "disabled", allow_duplicate=True),
    Output("solve-model-button", "loading", allow_duplicate=True),
    Output("postprocess-results-button", "disabled", allow_duplicate=True),
    Output("postprocess-results-button", "loading", allow_duplicate=True),
    Input("launch-mapdl-listener", "message"),
    Input("shutdown-mapdl-listener", "message"),
    Input("solve-model-listener", "message"),
    Input("postprocess-results-listener", "message"),
    prevent_initial_call=True,
)
def sync_controls_on_backend_events(
    launch_mapdl_message: dict[str, Any],
    shutdown_mapdl_message: dict[str, Any],
    solve_model_message: dict[str, Any],
    postprocess_results_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
    MAPDL instance state changes."""
    triggered_id = ctx.triggered_id
    controls = get_mapdl_page_controls_with_default()

    if triggered_id == "launch-mapdl-listener" and launch_mapdl_message:
        logger.info("Launch MAPDL listener triggered, updating controls")
        method_state = MethodState.model_validate_json(launch_mapdl_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button == "launch_mapdl":
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button in ["shutdown_mapdl", "solve_model"]:
                    controls[button]["disabled"] = False
        else:
            controls["launch_mapdl"]["disabled"] = False
            controls["launch_mapdl"]["loading"] = False
    elif triggered_id == "shutdown-mapdl-listener" and shutdown_mapdl_message:
        logger.info("Shutdown MAPDL listener triggered, updating controls")
        method_state = MethodState.model_validate_json(shutdown_mapdl_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button in ["shutdown_mapdl", "solve_model", "postprocess_results"]:
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button == "launch_mapdl":
                    controls[button]["disabled"] = False
                    controls[button]["loading"] = False
        else:
            controls["shutdown_mapdl"]["disabled"] = False
            controls["shutdown_mapdl"]["loading"] = False
    elif triggered_id == "solve-model-listener" and solve_model_message:
        logger.info("Solve Model listener triggered, updating controls")
        method_state = MethodState.model_validate_json(solve_model_message["data"])
        if method_state.status.value == "completed":
            controls["solve_model"]["disabled"] = False
            controls["solve_model"]["loading"] = False
            controls["postprocess_results"]["disabled"] = False
        else:
            controls["solve_model"]["disabled"] = False
            controls["solve_model"]["loading"] = False
        controls["shutdown_mapdl"]["disabled"] = False
    elif triggered_id == "postprocess-results-listener" and postprocess_results_message:
        logger.info("Postprocess Results listener triggered, updating controls")
        controls["postprocess_results"]["disabled"] = False
        controls["postprocess_results"]["loading"] = False
        controls["shutdown_mapdl"]["disabled"] = False
        controls["solve_model"]["disabled"] = False

    return (
        controls["launch_mapdl"]["disabled"],
        controls["launch_mapdl"]["loading"],
        controls["shutdown_mapdl"]["disabled"],
        controls["shutdown_mapdl"]["loading"],
        controls["solve_model"]["disabled"],
        controls["solve_model"]["loading"],
        controls["postprocess_results"]["disabled"],
        controls["postprocess_results"]["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/mapdl_instance_page.py

@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-mapdl-listener", "message"),
    Input("shutdown-mapdl-listener", "message"),
    Input("solve-model-listener", "message"),
    Input("postprocess-results-listener", "message"),
    prevent_initial_call=True,
)
def sync_notifications_on_backend_events(
    launch_mapdl_message: dict[str, Any],
    shutdown_mapdl_message: dict[str, Any],
    solve_model_message: dict[str, Any],
    postprocess_results_message: dict[str, Any],
) -> list[dict[str, Any]] | str:
    """Build and return notifications from backend websocket messages about MAPDL method state changes."""
    triggered_id = ctx.triggered_id
    notification = no_update

    if triggered_id == "launch-mapdl-listener" and launch_mapdl_message:
        logger.info("Launch MAPDL listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(launch_mapdl_message["data"])
        notification = handle_method_event(
            method_state,
            "launch-mapdl-notification",
            "MAPDL instance launched successfully!",
            "MAPDL initialization failed. Please check the logs.",
        )
    elif triggered_id == "shutdown-mapdl-listener" and shutdown_mapdl_message:
        logger.info("Shutdown MAPDL listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(shutdown_mapdl_message["data"])
        notification = handle_method_event(
            method_state,
            "shutdown-mapdl-notification",
            "MAPDL instance shutdown successfully!",
            "Failed to shutdown MAPDL instance. Please check the logs.",
        )
    elif triggered_id == "solve-model-listener" and solve_model_message:
        logger.info("Solve Model listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(solve_model_message["data"])
        notification = handle_method_event(
            method_state,
            "solve-model-notification",
            "Model solved successfully!",
            "Model solving failed. Please check the logs.",
        )
    elif triggered_id == "postprocess-results-listener" and postprocess_results_message:
        logger.info("Postprocess Results listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(postprocess_results_message["data"])
        notification = handle_method_event(
            method_state,
            "postprocess-results-notification",
            "Postprocessing completed successfully!",
            "Postprocessing 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 mapdl-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/mapdl_instance_page.py

@callback(
    Output("mapdl-logs-store", "data", allow_duplicate=True),
    Input("mapdl-output-listener", "message"),
    State("mapdl-logs-store", "data"),
    prevent_initial_call=True,
)
def store_outputs(message: dict[str, Any], current_logs: str) -> str:
    """Store MAPDL 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/mapdl_instance_page.py

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

ui/pages/instance_management/mapdl_instance_page.py

@callback(
    Output("mapdl-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.