optiSLang Product Instance Manager#
Feature highlight#
Ansys optiSLang is a powerful tool for design optimization and robustness evaluation, supporting a wide range of applications including sensitivity analysis, parameter studies, and uncertainty quantification across simulations involving multiple physical phenomena.
This example demonstrates how to create a product instance manager for optiSLang products using the SAF framework. The product instance manager enables management of the optiSLang product instance lifecycle, including starting, stopping, and accessing the product’s API.
In this example, you learn how to:
Launch an optiSLang session from a long-running transaction decorated with
@create_instance.Reuse the running session in the other transaction methods through the
@instancedecorator.Call the optiSLang API to evaluate the reference design of a project and refine it by exploring nearby parameter values.
Store the project file in an entity handle field through the storage scope.
Stream the solver output to the UI with
raise_eventand aDashClientevent listener.Drive button states and notifications from the event stream, then shut the instance down.
Prerequisites#
Note
This example requires the optiSLang 2024 R2 (24R2) product to be installed on your machine.
To work with an optiSLang product instance, be sure to install the core-pim and instance-management-optislang 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-optislang"]}
This will install the supported version of ansys-optislang-core to control the optiSLang 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 optiSLang 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
optislang-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 an optiSLang instance is currently running, and
evaluate_design_example stores the downloaded project file.
instance_created: bool = False
version: str = "242"
working_dir: str = ""
objective: str = ""
evaluate_design_example: EntityHandle = NO_ENTITY
Define the step model
download_example_filedownloads the example project used to evaluate a design and stores it in anEntityHandlefield.The
launch_optislanglong-running transaction is decorated with@create_instanceto create an instance of the optiSLang product manager,OslManager, while theshutdown_optislangtransaction is decorated with@instanceto reuse and shut down the existing instance.The
evaluate_designandrefine_designtransactions are decorated with@instanceto indicate they operate on the existing product instance, and call the optiSLang API to evaluate and refine a design, respectively.All four long-running transactions pass
enable_termination_event=Trueso the frontend can track their completion status.
@transaction(self=StepSpec(upload=["evaluate_design_example"]))
def download_example_file(self) -> None:
"""Download the example file for optiSLang evaluation."""
self.transaction.raise_event(message="Downloading example file.", stream_name="optislang-output-stream")
try:
example_path: str = examples.get_files("ten_bar_truss")[1][0] # type: ignore
with Path(example_path).open(mode="rb") as f: # type: ignore
self.evaluate_design_example = self.storage_scope.store_stream(f, Path("project.opf"))
except Exception as e:
self.transaction.raise_event(
message=f"Download failed: {e}",
stream_name="optislang-output-stream",
)
raise
self.transaction.raise_event(message="Example file downloaded.", stream_name="optislang-output-stream")
@transaction(
self=StepSpec(download=["version", "evaluate_design_example"], upload=["instance_created"]),
enable_termination_event=True,
)
@create_instance("osl_manager", OslManager)
@long_running
def launch_optislang(self, osl_manager: OslManager) -> None:
"""Initialize the optiSLang instance with the project file."""
self.transaction.raise_event(message="Initializing optiSLang instance.", stream_name="optislang-output-stream")
try:
osl_manager.initialize(self.evaluate_design_example, self.version)
except Exception as e:
self.transaction.raise_event(
message=f"optiSLang initialization failed: {e}",
stream_name="optislang-output-stream",
)
raise
self.instance_created = True
self.transaction.raise_event(message="optiSLang initialized.", stream_name="optislang-output-stream")
@transaction(self=StepSpec(upload=["objective", "working_dir"]), enable_termination_event=True)
@instance("osl_manager")
@long_running
def evaluate_design(self, osl_manager: OslManager):
"""Evaluate the reference design of the optiSLang project."""
self.transaction.raise_event(message="Evaluating design.", stream_name="optislang-output-stream")
try:
osl = osl_manager.instance
project = osl.application.project
# Evaluate reference design
root_system = project.root_system # type: ignore
design = root_system.get_reference_design()
evaluated_design = root_system.evaluate_design(design)
self.objective = str(evaluated_design.objectives[0].value) # type: ignore
except Exception as e:
self.transaction.raise_event(
message=f"Design evaluation failed: {e}", stream_name="optislang-output-stream"
)
raise
self.transaction.raise_event(message="Design evaluation succeeded.", stream_name="optislang-output-stream")
self.transaction.raise_event(
message=f"Current objective: {self.objective}.",
stream_name="optislang-output-stream",
)
@transaction(self=StepSpec(upload=["objective", "working_dir"]), enable_termination_event=True)
@instance("osl_manager")
@long_running
def refine_design(self, osl_manager: OslManager):
"""Refine the design by modifying the parameters of the reference design."""
successful_designs: list[Design] = []
self.transaction.raise_event(message="Refining design.", stream_name="optislang-output-stream")
try:
root_system = osl_manager.instance.application.project.root_system # type: ignore
design = root_system.get_reference_design()
evaluated_design = root_system.evaluate_design(design)
successful_designs.append(evaluated_design)
for i in range(1):
design = successful_designs[-1].copy_unevaluated_design()
parameters = design.parameters
parameter_value = parameters[i].value # type: ignore
parameters[i].value = parameter_value - 1 # type: ignore
evaluated_design = root_system.evaluate_design(design)
successful_designs.append(evaluated_design)
self.objective = str(evaluated_design.objectives[0].value) # type: ignore
except Exception as e:
self.transaction.raise_event(message=f"Design refine failed: {e}", stream_name="optislang-output-stream")
raise
self.transaction.raise_event(message="Design refine succeeded.", stream_name="optislang-output-stream")
self.transaction.raise_event(
message=f"Current objective: {self.objective}.",
stream_name="optislang-output-stream",
)
@transaction(self=StepSpec(upload=["instance_created"]), enable_termination_event=True)
@instance("osl_manager")
@long_running
def shutdown_optislang(self, osl_manager: OslManager) -> None:
"""Close the optiSLang instance."""
self.transaction.raise_event(
message="Starting to shutdown the instance.",
stream_name="optislang-output-stream",
)
try:
osl_manager.shutdown()
except Exception as e:
self.transaction.raise_event(
message=f"optiSLang shutdown failed: {e}",
stream_name="optislang-output-stream",
)
raise
self.instance_created = False
self.transaction.raise_event(
message="optiSLang instance shutdown complete.",
stream_name="optislang-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 output, the termination events, and the button
states stay in sync with a long-running transaction.
Compute the initial control states
get_optislang_page_controls_with_defaultis a helper, shared by every instance management page, that returns the defaultdisabled/loadingstate of each button together with the name of the transaction it triggers.initialize_optislang_controlsstarts from those defaults, then looks up the persisted state of each transaction withstep.get_long_running_method_state()so the buttons show the correct enabled/disabled/loading state even after a page reload.
def initialize_optislang_controls(project: ExamplesSolution) -> dict[str, dict[str, Any]] | Any:
"""Initialize the state of controls on the optiSLang instance management page."""
controls = get_optislang_page_controls_with_default()
for button in controls.keys():
controls[button]["disabled"] = True
controls[button]["loading"] = False
transactions = [
"launch_optislang",
"shutdown_optislang",
"evaluate_design",
"refine_design",
]
step = project.steps.optislang_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["evaluate_design"] == "completed" or transaction_states["refine_design"] == "completed":
controls["evaluate_design"]["disabled"] = False
controls["refine_design"]["disabled"] = False
controls["shutdown_optislang"]["disabled"] = False
elif transaction_states["launch_optislang"] == "completed":
controls["evaluate_design"]["disabled"] = False
controls["shutdown_optislang"]["disabled"] = False
else:
logger.info("No optiSLang instance detected, initializing page with default state")
controls["launch_optislang"]["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 evaluate and refine the design, and a logs card, using the initial control states computed above.
def layout(project: ExamplesSolution) -> html.Div:
"""Layout of the optiSLang step UI."""
controls = initialize_optislang_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-optislang-button",
size="xl",
color="#2790F1",
disabled=controls["launch_optislang"]["disabled"],
loading=controls["launch_optislang"]["loading"],
),
label="Launch optiSLang",
position="top",
),
dmc.Tooltip(
dmc.ActionIcon(
DashIconify(icon="mdi:shutdown", width=30),
id="shutdown-optislang-button",
size="xl",
color="#2790F1",
disabled=controls["shutdown_optislang"]["disabled"],
loading=controls["shutdown_optislang"]["loading"],
),
label="Shutdown optiSLang",
position="top",
),
],
gap="md",
justify="center",
),
dmc.Space(h=20),
dmc.Divider(variant="solid"),
dmc.Space(h=20),
dmc.Stack(
[
dmc.Button(
"Evaluate Design",
id="evaluate-design-button",
variant="filled",
color="#2790F1",
leftSection=DashIconify(icon="hugeicons:chart-evaluation"),
disabled=controls["evaluate_design"]["disabled"],
loading=controls["evaluate_design"]["loading"],
style={"width": "70%", "font-size": "15px"},
),
dmc.Button(
"Refine Design",
id="refine-design-button",
variant="filled",
color="#2790F1",
leftSection=DashIconify(icon="material-symbols:filter-alt"),
disabled=controls["refine_design"]["disabled"],
loading=controls["refine_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="optislang-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(
"optiSLang Instance Manager",
className="display-3",
style={"font-size": "40px", "font-weight": "bold"},
),
dmc.Blockquote(
"This example demonstrates how to leverage the instance management API to control Ansys optiSLang.\
Click the Launch button to\
start the instance. A transaction method will start optiSLang which can be used across all transaction\
methods. Run optiSLang operations with the Evaluate Design and\
Refine Design buttons. Close optiSLang 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 optiSLang 2024 R2"),
" installed and licensed,\n",
"- ",
dmc.Mark("ansys-saf-pim-light-server package 0.3 or later"),
" installed in the Python environment",
],
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
optislang-output-listenersubscribes to the free-form progress messages raised on theoptislang-output-streamby the step model.The other four listeners each subscribe to the termination event of one long-running transaction:
launch_optislang,shutdown_optislang,evaluate_design, andrefine_design.
@callback(
Output("optislang-instance-event-listeners-container", "children"),
Input("url", "pathname"),
)
def mount_event_listeners(project: ExamplesSolution) -> list[dict[str, Any]] | Any:
"""Mount optiSLang instance event listeners in the persistent main layout."""
logger.info("Mounting optiSLang instance management event listeners")
step = project.steps.optislang_step
return [
DashClient.create_event_listener(step, id="optislang-output-listener", stream_name="optislang-output-stream"),
DashClient.create_event_listener(step, id="launch-optislang-listener", stream_name="launch-optislang"),
DashClient.create_event_listener(step, id="shutdown-optislang-listener", stream_name="shutdown-optislang"),
DashClient.create_event_listener(step, id="evaluate-design-listener", stream_name="evaluate-design"),
DashClient.create_event_listener(step, id="refine-design-listener", stream_name="refine-design"),
]
Trigger a transaction from a button click
When the Launch optiSLang button is clicked, the callback downloads the example project file, starts the
launch_optislanglong-running transaction, and shows a loading notification.The
evaluate_design,refine_design, andshutdown_optislangcallbacks follow the same pattern: each one calls the matching transaction on the step and shows a loading notification.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Input("launch-optislang-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def launch_optislang(n_clicks: int, project: ExamplesSolution) -> list[dict[str, Any]] | Any:
"""Launch an instance of Ansys optiSLang."""
notification = no_update
if ctx.triggered_id == "launch-optislang-button" and n_clicks:
logger.info("Launch button clicked, starting optiSLang instance")
step = project.steps.optislang_step
step.download_example_file()
step.launch_optislang()
notification = [
dict(
title="Info",
id="launch-optislang-notification",
action="show",
message="Starting optiSLang instance... Please wait.",
autoClose=False,
loading=True,
color="blue",
withCloseButton=False,
)
]
return notification
Keep the button states in sync
sync_controls_on_clicksoptimistically 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_eventsreconciles the button states once a termination event is received, re-enabling the buttons that make sense for the new instance state.
@callback(
Output("launch-optislang-button", "disabled", allow_duplicate=True),
Output("launch-optislang-button", "loading", allow_duplicate=True),
Output("shutdown-optislang-button", "disabled", allow_duplicate=True),
Output("shutdown-optislang-button", "loading", allow_duplicate=True),
Output("evaluate-design-button", "disabled", allow_duplicate=True),
Output("evaluate-design-button", "loading", allow_duplicate=True),
Output("refine-design-button", "disabled", allow_duplicate=True),
Output("refine-design-button", "loading", allow_duplicate=True),
Input("launch-optislang-button", "n_clicks"),
Input("shutdown-optislang-button", "n_clicks"),
Input("evaluate-design-button", "n_clicks"),
Input("refine-design-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def sync_controls_on_clicks(
launch_optislang_clicks: int,
shutdown_optislang_clicks: int,
evaluate_design_clicks: int,
refine_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.optislang_step
triggered_id = ctx.triggered_id
controls = get_optislang_page_controls_with_default()
if triggered_id == "launch-optislang-button" and launch_optislang_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_optislang":
controls[button]["loading"] = True
elif triggered_id == "shutdown-optislang-button" and shutdown_optislang_clicks and step.instance_created:
logger.info("Shutdown button clicked, updating controls")
for button in controls.keys():
controls[button]["disabled"] = True
if button == "shutdown_optislang":
controls[button]["loading"] = True
elif triggered_id == "evaluate-design-button" and evaluate_design_clicks and step.instance_created:
logger.info("Evaluate Design button clicked, updating controls")
for button in controls.keys():
controls[button]["disabled"] = True
if button == "evaluate_design":
controls[button]["loading"] = True
elif triggered_id == "refine-design-button" and refine_design_clicks and step.instance_created:
logger.info("Refine Design button clicked, updating controls")
for button in controls.keys():
controls[button]["disabled"] = True
if button == "refine_design":
controls[button]["loading"] = True
return (
controls["launch_optislang"]["disabled"],
controls["launch_optislang"]["loading"],
controls["shutdown_optislang"]["disabled"],
controls["shutdown_optislang"]["loading"],
controls["evaluate_design"]["disabled"],
controls["evaluate_design"]["loading"],
controls["refine_design"]["disabled"],
controls["refine_design"]["loading"],
)
@callback(
Output("launch-optislang-button", "disabled", allow_duplicate=True),
Output("launch-optislang-button", "loading", allow_duplicate=True),
Output("shutdown-optislang-button", "disabled", allow_duplicate=True),
Output("shutdown-optislang-button", "loading", allow_duplicate=True),
Output("evaluate-design-button", "disabled", allow_duplicate=True),
Output("evaluate-design-button", "loading", allow_duplicate=True),
Output("refine-design-button", "disabled", allow_duplicate=True),
Output("refine-design-button", "loading", allow_duplicate=True),
Input("launch-optislang-listener", "message"),
Input("shutdown-optislang-listener", "message"),
Input("evaluate-design-listener", "message"),
Input("refine-design-listener", "message"),
prevent_initial_call=True,
)
def sync_controls_on_backend_events(
launch_optislang_message: dict[str, Any],
shutdown_optislang_message: dict[str, Any],
evaluate_design_message: dict[str, Any],
refine_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
optiSLang instance state changes."""
triggered_id = ctx.triggered_id
controls = get_optislang_page_controls_with_default()
if triggered_id == "launch-optislang-listener" and launch_optislang_message:
logger.info("Launch optiSLang listener triggered, updating controls")
method_state = MethodState.model_validate_json(launch_optislang_message["data"])
if method_state.status.value == "completed":
for button in controls.keys():
if button == "launch_optislang":
controls[button]["disabled"] = True
controls[button]["loading"] = False
elif button in ["shutdown_optislang", "evaluate_design"]:
controls[button]["disabled"] = False
else:
controls["launch_optislang"]["disabled"] = False
controls["launch_optislang"]["loading"] = False
elif triggered_id == "shutdown-optislang-listener" and shutdown_optislang_message:
logger.info("Shutdown optiSLang listener triggered, updating controls")
method_state = MethodState.model_validate_json(shutdown_optislang_message["data"])
if method_state.status.value == "completed":
for button in controls.keys():
if button in ["shutdown_optislang", "evaluate_design", "refine_design"]:
controls[button]["disabled"] = True
controls[button]["loading"] = False
elif button == "launch_optislang":
controls[button]["disabled"] = False
controls[button]["loading"] = False
else:
controls["shutdown_optislang"]["disabled"] = False
controls["shutdown_optislang"]["loading"] = False
elif triggered_id == "evaluate-design-listener" and evaluate_design_message:
logger.info("Evaluate Design listener triggered, updating controls")
method_state = MethodState.model_validate_json(evaluate_design_message["data"])
if method_state.status.value == "completed":
controls["evaluate_design"]["disabled"] = False
controls["evaluate_design"]["loading"] = False
controls["refine_design"]["disabled"] = False
else:
controls["evaluate_design"]["disabled"] = False
controls["evaluate_design"]["loading"] = False
controls["shutdown_optislang"]["disabled"] = False
elif triggered_id == "refine-design-listener" and refine_design_message:
logger.info("Refine Design listener triggered, updating controls")
controls["refine_design"]["disabled"] = False
controls["refine_design"]["loading"] = False
controls["shutdown_optislang"]["disabled"] = False
controls["evaluate_design"]["disabled"] = False
return (
controls["launch_optislang"]["disabled"],
controls["launch_optislang"]["loading"],
controls["shutdown_optislang"]["disabled"],
controls["shutdown_optislang"]["loading"],
controls["evaluate_design"]["disabled"],
controls["evaluate_design"]["loading"],
controls["refine_design"]["disabled"],
controls["refine_design"]["loading"],
)
Define the notification system.
sync_notifications_on_backend_eventsreacts to the same termination events and turns eachMethodStateinto a success or error notification through the sharedhandle_method_eventhelper.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Input("launch-optislang-listener", "message"),
Input("shutdown-optislang-listener", "message"),
Input("evaluate-design-listener", "message"),
Input("refine-design-listener", "message"),
prevent_initial_call=True,
)
def sync_notifications_on_backend_events(
launch_optislang_message: dict[str, Any],
shutdown_optislang_message: dict[str, Any],
evaluate_design_message: dict[str, Any],
refine_design_message: dict[str, Any],
) -> list[dict[str, Any]] | Any:
"""Build and return notifications from backend websocket messages about optiSLang method state changes."""
triggered_id = ctx.triggered_id
notification = no_update
if triggered_id == "launch-optislang-listener" and launch_optislang_message:
logger.info("Launch optiSLang listener triggered, updating notifications")
method_state = MethodState.model_validate_json(launch_optislang_message["data"])
notification = handle_method_event(
method_state,
"launch-optislang-notification",
"optiSLang instance launched successfully!",
"optiSLang initialization failed. Please check the logs.",
)
elif triggered_id == "shutdown-optislang-listener" and shutdown_optislang_message:
logger.info("Shutdown optiSLang listener triggered, updating notifications")
method_state = MethodState.model_validate_json(shutdown_optislang_message["data"])
notification = handle_method_event(
method_state,
"shutdown-optislang-notification",
"optiSLang instance shutdown successfully!",
"Failed to shutdown optiSLang instance. Please check the logs.",
)
elif triggered_id == "evaluate-design-listener" and evaluate_design_message:
logger.info("Evaluate Design listener triggered, updating notifications")
method_state = MethodState.model_validate_json(evaluate_design_message["data"])
notification = handle_method_event(
method_state,
"evaluate-design-notification",
"Design evaluated successfully!",
"Design evaluation failed. Please check the logs.",
)
elif triggered_id == "refine-design-listener" and refine_design_message:
logger.info("Refine design listener triggered, updating notifications")
method_state = MethodState.model_validate_json(refine_design_message["data"])
notification = handle_method_event(
method_state,
"refine-design-notification",
"Design refined successfully!",
"Design refinement failed. Please check the logs.",
)
return notification
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_outputsappends every message received on theoptislang-output-listenerstream to adcc.Store,display_outputmirrors that store into the console logs container, andclear_console_logsresets it.
@callback(
Output("optislang-logs-store", "data", allow_duplicate=True),
Input("optislang-output-listener", "message"),
State("optislang-logs-store", "data"),
prevent_initial_call=True,
)
def store_outputs(message: dict[str, Any], current_logs: str) -> str:
"""Store optiSLang output."""
if message:
new_content = message["data"].strip('"').replace("\\n", "\n")
combined = (current_logs or "") + "\n" + new_content
return combined
return current_logs
@callback(
Output("optislang-console-logs", "children", allow_duplicate=True),
Input("optislang-logs-store", "data"),
)
def display_output(current_logs: str) -> str:
"""Display optiSLang output."""
return current_logs
@callback(
Output("optislang-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.