Fluent Product Instance Manager

Objective

Drive an Ansys Fluent session from a solution. Launch the solver in a long-running transaction, import a mesh, set up the physics, and run a transient simulation, while the solver transcript is streamed live to the user interface.

Source code for this example is in the example solution.

Feature highlight

Ansys Fluent is a powerful simulation tool for computational fluid dynamics (CFD), supporting a wide range of applications including fluid flow, heat transfer, and multiphase phenomena.

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

In this example, you learn how to:

• Launch a Fluent session from a long-running transaction decorated with @create_instance.

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

• Call the Fluent API to import a mesh, set up the physics, and run a transient simulation.

• Store the solver output files in entity handle fields through the storage scope.

• Stream the solver transcript to the UI with raise_event and a DashClient event listener.

• Drive button states and notifications from the event stream, then shut the instance down.

Prerequisites

Note

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

This will install the supported version of ansys-fluent-core to control the Fluent 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 Fluent 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

self.transaction.raise_event() publishes a message on a named stream. The frontend subscribes to that stream with an event listener, which lets a long-running transaction report its progress without blocking the UI.

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. This is independent from the free-form messages sent with raise_event on the fluent-output-stream, and lets the frontend react to success or failure 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 Fluent instance is currently running, and the other fields store the solver output files produced by the transactions below.

solution/instance_management/fluent_step.py

instance_created: bool = False
version: str = "252"
max_temperature_file: EntityHandle = NO_ENTITY
simulation_output: EntityHandle = NO_ENTITY

fluent_output_file: EntityHandle = NO_ENTITY

Define the step model

• The launch_fluent long-running transaction is decorated with @create_instance to create an instance of the Fluent product manager, Fluent3DDPSolverManager, while the shutdown_fluent transaction is decorated with @instance to reuse and shut down the existing instance.

• The import_mesh and run_simulation transactions are decorated with @instance to indicate they operate on the existing product instance, and call the Fluent API to import a mesh and to run a transient simulation, respectively.

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

solution/instance_management/fluent_step.py

    @transaction(self=StepSpec(download=["version"], upload=["instance_created"]), enable_termination_event=True)
    @create_instance("fluent_3ddp_solver_instance", Fluent3DDPSolverManager)
    @long_running
    def launch_fluent(self, fluent_3ddp_solver_instance: Fluent3DDPSolverManager) -> None:
        """Initialize the Fluent 3DDP Solver instance."""
        self.transaction.raise_event(
            message="Initializing Fluent 3DDP Solver instance.",
            stream_name="fluent-output-stream",
        )
        try:
            fluent_3ddp_solver_instance.initialize(version=self.version)
        except Exception as e:
            self.transaction.raise_event(
                message=f"Fluent initialization failed: {e}",
                stream_name="fluent-output-stream",
            )
            raise
        self.instance_created = True
        self.transaction.raise_event(message="Fluent initialized.", stream_name="fluent-output-stream")

solution/instance_management/fluent_step.py

    @transaction(self=StepSpec(upload=["fluent_output_file"]), enable_termination_event=True)
    @instance("fluent_3ddp_solver_instance")
    @long_running
    def import_mesh(self, fluent_3ddp_solver_instance: Fluent3DDPSolverManager) -> None:
        """Import the mesh into the Fluent 3DDP Solver instance."""
        self.transaction.raise_event(message="Starting to import the mesh.", stream_name="fluent-output-stream")

        filepath = self.storage_scope.get_storage_root() / "fluent_output.txt"

        def on_transcript(transcript):  # type: ignore
            try:
                filepath.write_text(str(transcript))  # type: ignore
                self.fluent_output_file = self.storage_scope.store(filepath)
                self.transaction.raise_event(
                    message=str(transcript),  # type: ignore
                    stream_name="fluent-output-stream",
                )
            except Exception as e:
                self.transaction.raise_event(message=f"Mesh import failed: {e}", stream_name="fluent-output-stream")
                raise

        session = fluent_3ddp_solver_instance.instance

        session.transcript.register_callback(on_transcript)  # type: ignore

        self.transaction.raise_event(message="Started importing Mesh", stream_name="fluent-output-stream")

        # Import mesh
        try:
            example_file = self.transaction.get_asset_entity_handle("brake.msh.h5")
            example_file_path = fluent_3ddp_solver_instance.storage_scope.get_cached(example_file)
            session.tui.file.read_case(str(example_file_path))  # type: ignore
        except Exception as e:
            self.transaction.raise_event(message=f"Mesh import failed: {e}", stream_name="fluent-output-stream")
            raise

        # Define models and material
        session.tui.define.models.energy("yes", "no", "no", "no", "yes")  # type: ignore
        session.tui.define.models.unsteady_2nd_order_bounded("yes")  # type: ignore
        session.tui.define.materials.copy("solid", "steel")  # type: ignore

        # Solve only energy equation (conduction)
        session.tui.solve.set.equations("flow", "no", "kw", "no")  # type: ignore

        # Define disc rotation
        session.tui.define.boundary_conditions.set.solid(  # type: ignore
            "disc1",
            "disc2",
            "()",
            "solid-motion?",
            "yes",
            "solid-omega",
            "no",
            -15.79,
            "solid-x-origin",
            "no",
            -0.035,
            "solid-y-origin",
            "no",
            -0.821,
            "solid-z-origin",
            "no",
            0.045,
            "solid-ai",
            "no",
            0,
            "solid-aj",
            "no",
            1,
            "solid-ak",
            "no",
            0,
            "q",
        )

        # Apply frictional heating on pad-disc surfaces
        session.tui.define.boundary_conditions.set.wall(  # type: ignore
            "wall_pad-disc1",
            "wall-pad-disc2",
            "()",
            "wall-thickness",
            "no",
            0.002,
            "q-dot",
            "no",
            2e9,
            "q",
        )

        # session.file.

        # Apply convection cooling on outer surfaces due to air flow
        session.tui.define.boundary_conditions.set.wall(  # type: ignore
            "wall-disc*",
            "wall-geom*",
            "()",
            "thermal-bc",
            "yes",
            "convection",
            "convective-heat-transfer-coefficient",
            "no",
            100,
            "q",
        )

        # Initialize flow
        session.tui.solve.initialize.initialize_flow()  # type: ignore

        session.transcript.stop()  # type: ignore

        self.transaction.raise_event(message="Mesh import success.", stream_name="fluent-output-stream")

solution/instance_management/fluent_step.py

    @transaction(
        self=StepSpec(upload=["simulation_output", "max_temperature_file", "fluent_output_file"]),
        enable_termination_event=True,
    )
    @instance("fluent_3ddp_solver_instance")
    @long_running
    def run_simulation(self, fluent_3ddp_solver_instance: Fluent3DDPSolverManager) -> None:
        """Run the simulation in the Fluent 3DDP Solver instance."""
        filepath = self.storage_scope.get_storage_root() / "fluent_output.txt"

        self.transaction.raise_event(message="Starting to run the simulation.", stream_name="fluent-output-stream")

        def on_trancript(transcript):  # type: ignore
            try:
                filepath.write_text(str(transcript))  # type: ignore
                self.fluent_output_file = self.storage_scope.store(filepath)
                self.transaction.raise_event(
                    message=str(transcript),  # type: ignore
                    stream_name="fluent-output-stream",
                )
            except Exception as e:
                self.transaction.raise_event(message=f"Run simulation failed: {e}", stream_name="fluent-output-stream")
                raise

        session = fluent_3ddp_solver_instance.instance

        session.transcript.register_callback(on_trancript)  # type: ignore

        try:
            # Post processing setup
            session.tui.solve.report_definitions.add(  # type: ignore
                "max-pad-temperature",  # type: ignore
                "volume-max",
                "field",
                "temperature",
                "zone-names",  # type: ignore
                "geom-1-innerpad",
                "geom-1-outerpad",
            )
            session.tui.solve.report_definitions.add(  # type: ignore
                "max-disc-temperature",  # type: ignore
                "volume-max",
                "field",
                "temperature",
                "zone-names",  # type: ignore
                "disc1",
                "disc2",
            )

            session.tui.solve.report_plots.add(  # type: ignore
                "max-temperature",  # type: ignore
                "report-defs",
                "max-pad-temperature",
                "max-disc-temperature",
                "()",  # type: ignore
            )

            max_temperature_file_path = (
                fluent_3ddp_solver_instance.storage_scope.get_storage_root() / "max-temperature.out"
            )
            session.tui.solve.report_files.add(  # type: ignore
                "max-temperature",  # type: ignore
                "report-defs",
                "max-pad-temperature",
                "max-disc-temperature",
                "()",  # type: ignore
                "file-name",
                str(max_temperature_file_path),
            )

            session.results.graphics.contour["contour-1"] = {  # type: ignore
                "boundary_values": True,
                "color_map": {
                    "color": "field-velocity",
                    "font_automatic": True,
                    "font_name": "Helvetica",
                    "font_size": 0.032,
                    "format": "%0.2e",
                    "length": 0.54,
                    "log_scale": False,
                    "position": 1,
                    "show_all": True,
                    "size": 100,
                    "user_skip": 9,
                    "visible": True,
                    "width": 6.0,
                },
                "coloring": {"smooth": False},
                "contour_lines": False,
                "display_state_name": "None",
                "draw_mesh": False,
                "field": "temperature",
                "filled": True,
                "mesh_object": "",
                "node_values": True,
                "range_option": {"auto_range_on": {"global_range": True}},
            }

            session.tui.display.objects.create(  # type: ignore
                "contour",  # type: ignore
                "temperature",
                "field",
                "temperature",
                "surface-list",  # type: ignore
                "wall*",
                "()",
                "color-map",
                "format",
                "%0.1f",
                "q",
                "range-option",
                "auto-range-off",
                "minimum",
                300,
                "maximum",
                400,
                "q",
                "q",
            )

            session.tui.display.views.restore_view("top")  # type: ignore
            session.tui.display.views.camera.zoom_camera(2)  # type: ignore
            session.tui.display.views.save_view("animation-view")  # type: ignore

            session.tui.solve.animate.objects.create(  # type: ignore
                "animate-temperature",  # type: ignore
                "animate-on",
                "temperature",
                "frequency-of",
                "flow-time",  # type: ignore
                "flow-time-frequency",
                0.05,
                "view",
                "animation-view",
                "q",
            )

            # Run simulation
            simulation_output_path = fluent_3ddp_solver_instance.storage_scope.get_storage_root() / "brake-final.cas.h5"
            session.tui.solve.set.transient_controls.time_step_size(0.01)  # type: ignore
            session.tui.solve.dual_time_iterate(20, 5)  # type: ignore
            session.tui.file.write_case_data(str(simulation_output_path))  # type: ignore

            self.simulation_output = fluent_3ddp_solver_instance.storage_scope.store(simulation_output_path)
            self.max_temperature_file = fluent_3ddp_solver_instance.storage_scope.store(max_temperature_file_path)

        except Exception as e:
            self.transaction.raise_event(message=f"Run simulation failed: {e}", stream_name="fluent-output-stream")
            raise

        self.transaction.raise_event(message="Run simulation success.", stream_name="fluent-output-stream")

solution/instance_management/fluent_step.py

    @transaction(self=StepSpec(upload=["instance_created"]), enable_termination_event=True)
    @instance("fluent_3ddp_solver_instance")
    @long_running
    def shutdown_fluent(self, fluent_3ddp_solver_instance: Fluent3DDPSolverManager) -> None:
        """Shutdown the Fluent 3DDP Solver instance."""
        self.transaction.raise_event(message="Starting to shutdown the instance.", stream_name="fluent-output-stream")
        try:
            fluent_3ddp_solver_instance.shutdown()
        except Exception as e:
            self.transaction.raise_event(message=f"Fluent shutdown failed: {e}", stream_name="fluent-output-stream")
            raise
        self.instance_created = False
        self.transaction.raise_event(message="Fluent instance shutdown complete.", stream_name="fluent-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 page to a backend stream. Each message fires a callback, which is how the solver transcript, the termination events, and the button states stay in sync with a long-running transaction.

Compute the initial control states

• get_fluent_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_fluent_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/fluent_instance_page.py

def initialize_fluent_controls(project: ExamplesSolution) -> dict[str, dict[str, Any]] | Any:
    """Initialize the state of controls on the Fluent instance management page."""
    controls = get_fluent_page_controls_with_default()
    for button in controls.keys():
        controls[button]["disabled"] = True
        controls[button]["loading"] = False

    transactions = [
        "launch_fluent",
        "shutdown_fluent",
        "import_mesh",
        "run_simulation",
    ]
    step = project.steps.fluent_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["import_mesh"] == "completed" or transaction_states["run_simulation"] == "completed":
            controls["import_fluent_mesh"]["disabled"] = False
            controls["run_fluent_simulation"]["disabled"] = False
            controls["shutdown_fluent"]["disabled"] = False
        elif transaction_states["launch_fluent"] == "completed":
            controls["import_fluent_mesh"]["disabled"] = False
            controls["shutdown_fluent"]["disabled"] = False
    else:
        logger.info("No Fluent instance detected, initializing page with default state")
        controls["launch_fluent"]["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 import a mesh and run the simulation, and a logs card, using the initial control states computed above.

ui/pages/instance_management/fluent_instance_page.py

def layout(project: ExamplesSolution) -> html.Div:
    """Fluent instance management page layout."""
    controls = initialize_fluent_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-fluent-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["launch_fluent"]["disabled"],
                            loading=controls["launch_fluent"]["loading"],
                        ),
                        label="Launch Fluent",
                        position="top",
                    ),
                    dmc.Tooltip(
                        dmc.ActionIcon(
                            DashIconify(icon="mdi:shutdown", width=30),
                            id="shutdown-fluent-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["shutdown_fluent"]["disabled"],
                            loading=controls["shutdown_fluent"]["loading"],
                        ),
                        label="Shutdown Fluent",
                        position="top",
                    ),
                ],
                gap="md",
                justify="center",
            ),
            dmc.Space(h=20),
            dmc.Divider(variant="solid"),
            dmc.Space(h=20),
            dmc.Stack(
                [
                    dmc.Button(
                        "Import Mesh",
                        id="import-fluent-mesh-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="game-icons:mesh-network"),
                        disabled=controls["import_fluent_mesh"]["disabled"],
                        loading=controls["import_fluent_mesh"]["loading"],
                        style={"width": "70%", "font-size": "15px"},
                    ),
                    dmc.Button(
                        "Run Simulation",
                        id="run-fluent-simulation-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="codicon:run-all"),
                        disabled=controls["run_fluent_simulation"]["disabled"],
                        loading=controls["run_fluent_simulation"]["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="fluent-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(
                "Fluent 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 Fluent.\
                Click the Launch button to\
                start the instance. A transaction method will start Fluent which can be used across all transaction\
                methods. Run Fluent operations with the Import Mesh and\
                Run Simulation buttons. Close Fluent 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 Fluent 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",
                        "- ",
                        dmc.Mark("GLOW_PRODUCT_HOST"),
                        " environment variable configured to point to the Fluent host machine",
                        " (your local machine if running Fluent locally) in the .env file of the solution,\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

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

• The other four listeners each subscribe to the termination event of one long-running transaction: launch_fluent, shutdown_fluent, import_mesh, and run_simulation.

ui/pages/instance_management/fluent_instance_page.py

@callback(
    Output("fluent-instance-event-listeners-container", "children"),
    Input("url", "pathname"),
)
def mount_event_listeners(project: ExamplesSolution) -> list[dict[str, Any]] | Any:
    """Mount Fluent instance event listeners in the persistent main layout."""
    step = project.steps.fluent_step
    return [
        DashClient.create_event_listener(step, id="fluent-output-listener", stream_name="fluent-output-stream"),
        DashClient.create_event_listener(step, id="launch-fluent-listener", stream_name="launch-fluent"),
        DashClient.create_event_listener(step, id="import-fluent-mesh-listener", stream_name="import-mesh"),
        DashClient.create_event_listener(step, id="run-fluent-simulation-listener", stream_name="run-simulation"),
        DashClient.create_event_listener(step, id="shutdown-fluent-listener", stream_name="shutdown-fluent"),
    ]

Trigger a transaction from a button click

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

• The import_mesh, run_simulation, and shutdown_fluent callbacks follow the same pattern: each one calls the matching transaction on the step and shows a loading notification.

ui/pages/instance_management/fluent_instance_page.py

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

    if ctx.triggered_id == "launch-fluent-button" and n_clicks:
        logger.info("Launch button clicked, starting Fluent instance")
        step = project.steps.fluent_step
        step.launch_fluent()

        notification = [
            dict(
                title="Info",
                id="launch-fluent-notification",
                action="show",
                message="Starting Fluent 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/fluent_instance_page.py

@callback(
    Output("launch-fluent-button", "disabled", allow_duplicate=True),
    Output("launch-fluent-button", "loading", allow_duplicate=True),
    Output("shutdown-fluent-button", "disabled", allow_duplicate=True),
    Output("shutdown-fluent-button", "loading", allow_duplicate=True),
    Output("import-fluent-mesh-button", "disabled", allow_duplicate=True),
    Output("import-fluent-mesh-button", "loading", allow_duplicate=True),
    Output("run-fluent-simulation-button", "disabled", allow_duplicate=True),
    Output("run-fluent-simulation-button", "loading", allow_duplicate=True),
    Input("launch-fluent-button", "n_clicks"),
    Input("shutdown-fluent-button", "n_clicks"),
    Input("import-fluent-mesh-button", "n_clicks"),
    Input("run-fluent-simulation-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def sync_controls_on_clicks(
    launch_fluent_clicks: int,
    shutdown_fluent_clicks: int,
    import_fluent_mesh_clicks: int,
    run_fluent_simulation_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.fluent_step
    triggered_id = ctx.triggered_id
    controls = get_fluent_page_controls_with_default()

    if triggered_id == "launch-fluent-button" and launch_fluent_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_fluent":
                controls[button]["loading"] = True
    elif triggered_id == "shutdown-fluent-button" and shutdown_fluent_clicks and step.instance_created:
        logger.info("Shutdown button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "shutdown_fluent":
                controls[button]["loading"] = True
    elif triggered_id == "import-fluent-mesh-button" and import_fluent_mesh_clicks and step.instance_created:
        logger.info("Import Mesh button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "import_fluent_mesh":
                controls[button]["loading"] = True
    elif triggered_id == "run-fluent-simulation-button" and run_fluent_simulation_clicks and step.instance_created:
        logger.info("Run Simulation button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "run_fluent_simulation":
                controls[button]["loading"] = True

    return (
        controls["launch_fluent"]["disabled"],
        controls["launch_fluent"]["loading"],
        controls["shutdown_fluent"]["disabled"],
        controls["shutdown_fluent"]["loading"],
        controls["import_fluent_mesh"]["disabled"],
        controls["import_fluent_mesh"]["loading"],
        controls["run_fluent_simulation"]["disabled"],
        controls["run_fluent_simulation"]["loading"],
    )

ui/pages/instance_management/fluent_instance_page.py

@callback(
    Output("launch-fluent-button", "disabled", allow_duplicate=True),
    Output("launch-fluent-button", "loading", allow_duplicate=True),
    Output("shutdown-fluent-button", "disabled", allow_duplicate=True),
    Output("shutdown-fluent-button", "loading", allow_duplicate=True),
    Output("import-fluent-mesh-button", "disabled", allow_duplicate=True),
    Output("import-fluent-mesh-button", "loading", allow_duplicate=True),
    Output("run-fluent-simulation-button", "disabled", allow_duplicate=True),
    Output("run-fluent-simulation-button", "loading", allow_duplicate=True),
    Input("launch-fluent-listener", "message"),
    Input("shutdown-fluent-listener", "message"),
    Input("import-fluent-mesh-listener", "message"),
    Input("run-fluent-simulation-listener", "message"),
    prevent_initial_call=True,
)
def sync_controls_on_backend_events(
    launch_fluent_message: dict[str, Any],
    shutdown_fluent_message: dict[str, Any],
    import_fluent_mesh_message: dict[str, Any],
    run_fluent_simulation_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
    Fluent instance state changes."""
    triggered_id = ctx.triggered_id
    controls = get_fluent_page_controls_with_default()

    if triggered_id == "launch-fluent-listener" and launch_fluent_message:
        logger.info("Launch Fluent listener triggered, updating controls")
        method_state = MethodState.model_validate_json(launch_fluent_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button == "launch_fluent":
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button in ["shutdown_fluent", "import_fluent_mesh"]:
                    controls[button]["disabled"] = False
        else:
            controls["launch_fluent"]["disabled"] = False
            controls["launch_fluent"]["loading"] = False
    elif triggered_id == "shutdown-fluent-listener" and shutdown_fluent_message:
        logger.info("Shutdown Fluent listener triggered, updating controls")
        method_state = MethodState.model_validate_json(shutdown_fluent_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button in ["shutdown_fluent", "import_fluent_mesh", "run_fluent_simulation"]:
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button == "launch_fluent":
                    controls[button]["disabled"] = False
                    controls[button]["loading"] = False
        else:
            controls["shutdown_fluent"]["disabled"] = False
            controls["shutdown_fluent"]["loading"] = False
    elif triggered_id == "import-fluent-mesh-listener" and import_fluent_mesh_message:
        logger.info("Import Mesh listener triggered, updating controls")
        method_state = MethodState.model_validate_json(import_fluent_mesh_message["data"])
        if method_state.status.value == "completed":
            controls["import_fluent_mesh"]["disabled"] = False
            controls["import_fluent_mesh"]["loading"] = False
            controls["run_fluent_simulation"]["disabled"] = False
        else:
            controls["import_fluent_mesh"]["disabled"] = False
            controls["import_fluent_mesh"]["loading"] = False
        controls["shutdown_fluent"]["disabled"] = False
    elif triggered_id == "run-fluent-simulation-listener" and run_fluent_simulation_message:
        logger.info("Run Simulation listener triggered, updating controls")
        controls["run_fluent_simulation"]["disabled"] = False
        controls["run_fluent_simulation"]["loading"] = False
        controls["shutdown_fluent"]["disabled"] = False
        controls["import_fluent_mesh"]["disabled"] = False

    return (
        controls["launch_fluent"]["disabled"],
        controls["launch_fluent"]["loading"],
        controls["shutdown_fluent"]["disabled"],
        controls["shutdown_fluent"]["loading"],
        controls["import_fluent_mesh"]["disabled"],
        controls["import_fluent_mesh"]["loading"],
        controls["run_fluent_simulation"]["disabled"],
        controls["run_fluent_simulation"]["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/fluent_instance_page.py

@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-fluent-listener", "message"),
    Input("shutdown-fluent-listener", "message"),
    Input("import-fluent-mesh-listener", "message"),
    Input("run-fluent-simulation-listener", "message"),
    prevent_initial_call=True,
)
def sync_notifications_on_backend_events(
    launch_fluent_message: dict[str, Any],
    shutdown_fluent_message: dict[str, Any],
    import_fluent_mesh_message: dict[str, Any],
    run_fluent_simulation_message: dict[str, Any],
) -> list[dict[str, Any]] | Any:
    """Build and return notifications from backend websocket messages about Fluent method state changes."""
    triggered_id = ctx.triggered_id
    notification = no_update

    if triggered_id == "launch-fluent-listener" and launch_fluent_message:
        logger.info("Launch Fluent listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(launch_fluent_message["data"])
        notification = handle_method_event(
            method_state,
            "launch-fluent-notification",
            "Fluent instance launched successfully!",
            "Fluent initialization failed. Please check the logs.",
        )
    elif triggered_id == "shutdown-fluent-listener" and shutdown_fluent_message:
        logger.info("Shutdown Fluent listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(shutdown_fluent_message["data"])
        notification = handle_method_event(
            method_state,
            "shutdown-fluent-notification",
            "Fluent instance shutdown successfully!",
            "Failed to shutdown Fluent instance. Please check the logs.",
        )
    elif triggered_id == "import-fluent-mesh-listener" and import_fluent_mesh_message:
        logger.info("Import Mesh listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(import_fluent_mesh_message["data"])
        notification = handle_method_event(
            method_state,
            "import-fluent-mesh-notification",
            "Mesh imported successfully!",
            "Mesh import failed. Please check the logs.",
        )
    elif triggered_id == "run-fluent-simulation-listener" and run_fluent_simulation_message:
        logger.info("Run Simulation listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(run_fluent_simulation_message["data"])
        notification = handle_method_event(
            method_state,
            "run-fluent-simulation-notification",
            "Simulation started successfully!",
            "Simulation failed to start. 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 fluent-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/fluent_instance_page.py

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

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

ui/pages/instance_management/fluent_instance_page.py

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