AEDT Product Instance Manager#

Feature highlight#

Ansys Electronics Desktop is a unified simulation platform that integrates multiple Ansys solvers for electromagnetic, thermal, and circuit analysis, enabling seamless collaboration and coupled physics simulations.

This example shows how to create a product instance manager for AEDT products, such as Maxwell 2D, using the SAF framework. The product instance manager allows you to manage the lifecycle of an AEDT product instance, 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 AEDT API to add a rectangle to the design and to analyze it.

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

  • Wire one callback per button so the UI triggers a transaction and reports its outcome.

Prerequisites#

Note

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

This will install the supported version of PyAEDT to control the AEDT 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 AEDT 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 and termination event

self.transaction.raise_event() publishes a free-form progress message on a named stream. Passing enable_termination_event=True to @transaction additionally makes SAF 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 an AEDT instance is currently running.

solution/instance_management/aedt_step.py#
instance_created: bool = False
aedt_version: str = "252"
origin: list[float] = [0, 0, 0]
dimension: list[float] = [10, 10]
Define the step model
  • The launch_aedt long-running transaction is decorated with @create_instance to create an instance of the AEDT product manager, Maxwell2DManager, while the shutdown_aedt transaction is decorated with @instance to reuse and shut down the existing instance.

  • The add_rectangle and analyze_design transactions are decorated with @instance to indicate they operate on the existing product instance, and call the AEDT API to add a rectangle to the design and to analyze it, respectively.

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

solution/instance_management/aedt_step.py#
    @transaction(self=StepSpec(download=["aedt_version"], upload=["instance_created"]), enable_termination_event=True)
    @create_instance("maxwell_2d_instance", Maxwell2DManager)
    @long_running
    def launch_aedt(self, maxwell_2d_instance: Maxwell2DManager) -> None:
        """Initialize the Maxwell 2D instance."""
        self.transaction.raise_event(message="Initializing AEDT Maxwell 2D instance.", stream_name="aedt-output-stream")
        try:
            maxwell_2d_instance.initialize(version=self.aedt_version)
        except Exception as e:
            self.transaction.raise_event(
                message=f"Failed to initialize AEDT Maxwell 2D instance: {str(e)}", stream_name="aedt-output-stream"
            )
            raise
        self.transaction.raise_event(
            message="AEDT Maxwell 2D instance initialized successfully.",
            stream_name="aedt-output-stream",
        )
        self.instance_created = True
solution/instance_management/aedt_step.py#
    @transaction(self=StepSpec(download=["origin", "dimension"]), enable_termination_event=True)
    @instance("maxwell_2d_instance")
    @long_running
    def add_rectangle(self, maxwell_2d_instance: Maxwell2DManager) -> None:
        """Add a rectangle to the Maxwell 2D instance."""
        self.transaction.raise_event(
            message="Adding rectangle to AEDT Maxwell 2D instance.",
            stream_name="aedt-output-stream",
        )
        try:
            maxwell_2d_instance.instance.modeler.create_rectangle(self.origin, self.dimension)  # type: ignore
        except Exception as e:
            self.transaction.raise_event(
                message=f"Failed to add rectangle to AEDT Maxwell 2D instance: {e}",
                stream_name="aedt-output-stream",
            )
            raise
        self.transaction.raise_event(
            message="Rectangle added to AEDT Maxwell 2D instance successfully.",
            stream_name="aedt-output-stream",
        )
solution/instance_management/aedt_step.py#
    @transaction(enable_termination_event=True)
    @instance("maxwell_2d_instance")
    @long_running
    def analyze_design(self, maxwell_2d_instance: Maxwell2DManager) -> None:
        """Run the analysis on the Maxwell 2D instance."""
        self.transaction.raise_event(
            message="Running analysis on AEDT Maxwell 2D instance.",
            stream_name="aedt-output-stream",
        )
        try:
            result = maxwell_2d_instance.instance.analyze()  # type: ignore
            assert result, "Unsuccessful analysis on AEDT Maxwell 2D instance."
        except Exception as e:
            self.transaction.raise_event(
                message=f"Failed to run analysis on AEDT Maxwell 2D instance: {e}",
                stream_name="aedt-output-stream",
            )
            raise
        self.transaction.raise_event(
            message="Analysis on AEDT Maxwell 2D instance completed successfully.",
            stream_name="aedt-output-stream",
        )
solution/instance_management/aedt_step.py#
    @transaction(self=StepSpec(upload=["instance_created"]), enable_termination_event=True)
    @instance("maxwell_2d_instance")
    @long_running
    def shutdown_aedt(self, maxwell_2d_instance: Maxwell2DManager) -> None:
        """Shutdown the AEDT Maxwell 2D instance."""
        self.transaction.raise_event(
            message="Shutting down AEDT Maxwell 2D instance.",
            stream_name="aedt-output-stream",
        )
        try:
            maxwell_2d_instance.shutdown()
        except Exception as e:
            self.transaction.raise_event(
                message=f"Failed to shutdown AEDT Maxwell 2D instance: {e}",
                stream_name="aedt-output-stream",
            )
            raise
        self.transaction.raise_event(
            message="AEDT Maxwell 2D instance shutdown successfully.",
            stream_name="aedt-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 page to a backend stream. Each message fires a callback, which is how the solver output, the termination events, and the button states stay in sync with a long-running transaction.

Compute the initial control states
  • get_aedt_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_aedt_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/aedt_instance_page.py#
def initialize_aedt_controls(project: ExamplesSolution) -> dict[str, dict[str, Any]] | Any:
    """Initialize the state of controls on the AEDT instance management page."""
    controls = get_aedt_page_controls_with_default()

    step = project.steps.aedt_step

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

    transactions = [
        "launch_aedt",
        "shutdown_aedt",
        "add_rectangle",
        "analyze_design",
    ]

    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["add_rectangle"] == "completed" or transaction_states["analyze_design"] == "completed":
            controls["add_rectangle"]["disabled"] = False
            controls["analyze_design"]["disabled"] = False
            controls["shutdown_aedt"]["disabled"] = False
        elif transaction_states["launch_aedt"] == "completed":
            controls["add_rectangle"]["disabled"] = False
            controls["shutdown_aedt"]["disabled"] = False
    else:
        logger.info("No AEDT instance detected, initializing page with default state")
        controls["launch_aedt"]["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 add a rectangle and analyze the design, and a logs card, using the initial control states computed above.

ui/pages/instance_management/aedt_instance_page.py#
def layout(project: ExamplesSolution) -> html.Div:
    """Layout of the AEDT step UI."""
    controls = initialize_aedt_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-aedt-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["launch_aedt"]["disabled"],
                            loading=controls["launch_aedt"]["loading"],
                        ),
                        label="Launch AEDT",
                        position="top",
                    ),
                    dmc.Tooltip(
                        dmc.ActionIcon(
                            DashIconify(icon="mdi:shutdown", width=30),
                            id="shutdown-aedt-button",
                            size="xl",
                            color="#2790F1",
                            disabled=controls["shutdown_aedt"]["disabled"],
                            loading=controls["shutdown_aedt"]["loading"],
                        ),
                        label="Shutdown AEDT",
                        position="top",
                    ),
                ],
                gap="md",
                justify="center",
            ),
            dmc.Space(h=20),
            dmc.Divider(variant="solid"),
            dmc.Space(h=20),
            dmc.Stack(
                [
                    dmc.Button(
                        "Add Rectangle",
                        id="add-rectangle-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="material-symbols:square-outline"),
                        disabled=controls["add_rectangle"]["disabled"],
                        loading=controls["add_rectangle"]["loading"],
                        style={"width": "70%", "font-size": "15px"},
                    ),
                    dmc.Button(
                        "Analyze Design",
                        id="analyze-design-button",
                        variant="filled",
                        color="#2790F1",
                        leftSection=DashIconify(icon="tabler:analyze"),
                        disabled=controls["analyze_design"]["disabled"],
                        loading=controls["analyze_design"]["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="aedt-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(
                "AEDT 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 AEDT.\
                Click the Launch button to\
                start the instance. A transaction method will start AEDT which can be used across all transaction\
                methods. Run AEDT operations with the Add Rectangle and\
                Analyze Design buttons. Close AEDT 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("AEDT 2025 R1 (25R1)"),
                        " 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
  • aedt-output-listener subscribes to the free-form progress messages raised on the aedt-output-stream by the step model.

  • The other four listeners each subscribe to the termination event of one long-running transaction: launch_aedt, shutdown_aedt, add_rectangle, and analyze_design.

ui/pages/instance_management/aedt_instance_page.py#
@callback(
    Output("aedt-instance-event-listeners-container", "children"),
    Input("url", "pathname"),
)
def mount_event_listeners(project: ExamplesSolution) -> list[dict[str, Any]] | Any:
    """Mount AEDT instance event listeners in the persistent main layout."""
    logger.info("Mounting AEDT instance management event listeners")
    step = project.steps.aedt_step
    return [
        DashClient.create_event_listener(step, id="aedt-output-listener", stream_name="aedt-output-stream"),
        DashClient.create_event_listener(step, id="launch-aedt-listener", stream_name="launch-aedt"),
        DashClient.create_event_listener(step, id="shutdown-aedt-listener", stream_name="shutdown-aedt"),
        DashClient.create_event_listener(step, id="add-rectangle-listener", stream_name="add-rectangle"),
        DashClient.create_event_listener(step, id="analyze-design-listener", stream_name="analyze-design"),
    ]
Trigger a transaction from a button click
  • When the Launch AEDT button is clicked, the callback starts the launch_aedt long-running transaction and shows a loading notification.

  • The add_rectangle, analyze_design, and shutdown_aedt callbacks follow the same pattern: each one calls the matching transaction on the step and shows a loading notification.

ui/pages/instance_management/aedt_instance_page.py#
@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-aedt-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def launch_aedt(n_clicks: int, project: ExamplesSolution) -> list[dict[str, Any]] | Any:
    """Launch an instance of Ansys AEDT."""
    notification = no_update

    if ctx.triggered_id == "launch-aedt-button" and n_clicks:
        logger.info("Launch button clicked, starting AEDT instance")
        step = project.steps.aedt_step
        step.launch_aedt()

        notification = [
            dict(
                title="Info",
                id="launch-aedt-notification",
                action="show",
                message="Starting AEDT 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/aedt_instance_page.py#
@callback(
    Output("launch-aedt-button", "disabled", allow_duplicate=True),
    Output("launch-aedt-button", "loading", allow_duplicate=True),
    Output("shutdown-aedt-button", "disabled", allow_duplicate=True),
    Output("shutdown-aedt-button", "loading", allow_duplicate=True),
    Output("add-rectangle-button", "disabled", allow_duplicate=True),
    Output("add-rectangle-button", "loading", allow_duplicate=True),
    Output("analyze-design-button", "disabled", allow_duplicate=True),
    Output("analyze-design-button", "loading", allow_duplicate=True),
    Input("launch-aedt-button", "n_clicks"),
    Input("shutdown-aedt-button", "n_clicks"),
    Input("add-rectangle-button", "n_clicks"),
    Input("analyze-design-button", "n_clicks"),
    State("url", "pathname"),
    prevent_initial_call=True,
)
def sync_controls_on_clicks(
    launch_aedt_clicks: int,
    shutdown_aedt_clicks: int,
    add_rectangle_clicks: int,
    analyze_design_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.aedt_step
    triggered_id = ctx.triggered_id
    controls = get_aedt_page_controls_with_default()

    if triggered_id == "launch-aedt-button" and launch_aedt_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_aedt":
                controls[button]["loading"] = True
    elif triggered_id == "shutdown-aedt-button" and shutdown_aedt_clicks and step.instance_created:
        logger.info("Shutdown button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "shutdown_aedt":
                controls[button]["loading"] = True
    elif triggered_id == "add-rectangle-button" and add_rectangle_clicks and step.instance_created:
        logger.info("Add Rectangle button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "add_rectangle":
                controls[button]["loading"] = True
    elif triggered_id == "analyze-design-button" and analyze_design_clicks and step.instance_created:
        logger.info("Analyze Design button clicked, updating controls")
        for button in controls.keys():
            controls[button]["disabled"] = True
            if button == "analyze_design":
                controls[button]["loading"] = True

    return (
        controls["launch_aedt"]["disabled"],
        controls["launch_aedt"]["loading"],
        controls["shutdown_aedt"]["disabled"],
        controls["shutdown_aedt"]["loading"],
        controls["add_rectangle"]["disabled"],
        controls["add_rectangle"]["loading"],
        controls["analyze_design"]["disabled"],
        controls["analyze_design"]["loading"],
    )
ui/pages/instance_management/aedt_instance_page.py#
@callback(
    Output("launch-aedt-button", "disabled", allow_duplicate=True),
    Output("launch-aedt-button", "loading", allow_duplicate=True),
    Output("shutdown-aedt-button", "disabled", allow_duplicate=True),
    Output("shutdown-aedt-button", "loading", allow_duplicate=True),
    Output("add-rectangle-button", "disabled", allow_duplicate=True),
    Output("add-rectangle-button", "loading", allow_duplicate=True),
    Output("analyze-design-button", "disabled", allow_duplicate=True),
    Output("analyze-design-button", "loading", allow_duplicate=True),
    Input("launch-aedt-listener", "message"),
    Input("shutdown-aedt-listener", "message"),
    Input("add-rectangle-listener", "message"),
    Input("analyze-design-listener", "message"),
    prevent_initial_call=True,
)
def sync_controls_on_backend_events(
    launch_aedt_message: dict[str, Any],
    shutdown_aedt_message: dict[str, Any],
    add_rectangle_message: dict[str, Any],
    analyze_design_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
    AEDT instance state changes."""
    triggered_id = ctx.triggered_id
    controls = get_aedt_page_controls_with_default()

    if triggered_id == "launch-aedt-listener" and launch_aedt_message:
        logger.info("Launch AEDT listener triggered, updating controls")
        method_state = MethodState.model_validate_json(launch_aedt_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button == "launch_aedt":
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button in ["shutdown_aedt", "add_rectangle"]:
                    controls[button]["disabled"] = False
        else:
            controls["launch_aedt"]["disabled"] = False
            controls["launch_aedt"]["loading"] = False
    elif triggered_id == "shutdown-aedt-listener" and shutdown_aedt_message:
        logger.info("Shutdown AEDT listener triggered, updating controls")
        method_state = MethodState.model_validate_json(shutdown_aedt_message["data"])
        if method_state.status.value == "completed":
            for button in controls.keys():
                if button in ["shutdown_aedt", "add_rectangle", "analyze_design"]:
                    controls[button]["disabled"] = True
                    controls[button]["loading"] = False
                elif button == "launch_aedt":
                    controls[button]["disabled"] = False
                    controls[button]["loading"] = False
        else:
            controls["shutdown_aedt"]["disabled"] = False
            controls["shutdown_aedt"]["loading"] = False
    elif triggered_id == "add-rectangle-listener" and add_rectangle_message:
        logger.info("Add Rectangle listener triggered, updating controls")
        method_state = MethodState.model_validate_json(add_rectangle_message["data"])
        if method_state.status.value == "completed":
            controls["add_rectangle"]["disabled"] = False
            controls["add_rectangle"]["loading"] = False
            controls["analyze_design"]["disabled"] = False
        else:
            controls["add_rectangle"]["disabled"] = False
            controls["add_rectangle"]["loading"] = False
        controls["shutdown_aedt"]["disabled"] = False
    elif triggered_id == "analyze-design-listener" and analyze_design_message:
        logger.info("Analyze Design listener triggered, updating controls")
        controls["analyze_design"]["disabled"] = False
        controls["analyze_design"]["loading"] = False
        controls["shutdown_aedt"]["disabled"] = False
        controls["add_rectangle"]["disabled"] = False

    return (
        controls["launch_aedt"]["disabled"],
        controls["launch_aedt"]["loading"],
        controls["shutdown_aedt"]["disabled"],
        controls["shutdown_aedt"]["loading"],
        controls["add_rectangle"]["disabled"],
        controls["add_rectangle"]["loading"],
        controls["analyze_design"]["disabled"],
        controls["analyze_design"]["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/aedt_instance_page.py#
@callback(
    Output("notification-container", "sendNotifications", allow_duplicate=True),
    Input("launch-aedt-listener", "message"),
    Input("shutdown-aedt-listener", "message"),
    Input("add-rectangle-listener", "message"),
    Input("analyze-design-listener", "message"),
    prevent_initial_call=True,
)
def sync_notifications_on_backend_events(
    launch_aedt_message: dict[str, Any],
    shutdown_aedt_message: dict[str, Any],
    add_rectangle_message: dict[str, Any],
    analyze_design_message: dict[str, Any],
) -> list[dict[str, Any]] | Any:
    """Build and return notifications from backend websocket messages about AEDT method state changes."""
    triggered_id = ctx.triggered_id
    notification = no_update

    if triggered_id == "launch-aedt-listener" and launch_aedt_message:
        logger.info("Launch AEDT listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(launch_aedt_message["data"])
        notification = handle_method_event(
            method_state,
            "launch-aedt-notification",
            "AEDT instance launched successfully!",
            "AEDT initialization failed. Please check the logs.",
        )
    elif triggered_id == "shutdown-aedt-listener" and shutdown_aedt_message:
        logger.info("Shutdown AEDT listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(shutdown_aedt_message["data"])
        notification = handle_method_event(
            method_state,
            "shutdown-aedt-notification",
            "AEDT instance shutdown successfully!",
            "Failed to shutdown AEDT instance. Please check the logs.",
        )
    elif triggered_id == "add-rectangle-listener" and add_rectangle_message:
        logger.info("Add Rectangle listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(add_rectangle_message["data"])
        notification = handle_method_event(
            method_state,
            "add-rectangle-notification",
            "Rectangle added successfully!",
            "Failed to add rectangle. Please check the logs.",
        )
    elif triggered_id == "analyze-design-listener" and analyze_design_message:
        logger.info("Analyze Design listener triggered, updating notifications")
        method_state = MethodState.model_validate_json(analyze_design_message["data"])
        notification = handle_method_event(
            method_state,
            "analyze-design-notification",
            "Design analyzed successfully!",
            "Design analysis 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 aedt-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/aedt_instance_page.py#
@callback(
    Output("aedt-logs-store", "data", allow_duplicate=True),
    Input("aedt-output-listener", "message"),
    State("aedt-logs-store", "data"),
    prevent_initial_call=True,
)
def store_outputs(message: dict[str, Any], current_logs: str) -> str:
    """Store AEDT 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/aedt_instance_page.py#
@callback(
    Output("aedt-console-logs", "children", allow_duplicate=True),
    Input("aedt-logs-store", "data"),
)
def display_output(current_logs: str) -> str:
    """Display AEDT output."""
    return current_logs
ui/pages/instance_management/aedt_instance_page.py#
@callback(
    Output("aedt-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.