Geometry Product Instance Manager#
Feature highlight#
PyGeometry is a versatile library for geometric modeling and spatial computation, supporting a wide range of applications including shape representation, transformation operations, and spatial analysis across simulations involving multiple physical phenomena.
This example demonstrates how to create a product instance manager for Geometry products using SAF. The product instance manager enables management of the Geometry product instance lifecycle, including starting, stopping, and accessing the product’s API.
In this example, you learn how to:
Launch a Geometry instance from a long-running transaction decorated with
@create_instance.Reuse the running instance in the other transaction methods through the
@instancedecorator.Call the Geometry API to extrude a sketch into a slot and to read the active design.
Store the resulting face count, edge count, and design name in typed step fields.
Stream progress messages 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 Geometry 2025 R2 Service Pack 4 (25R2 SP4) product to be installed on your machine.
To work with a Geometry product instance, be sure to install the core-pim and instance-management-geometry 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-geometry"]}
This will install the supported version of ansys-geometry-core to control the Geometry 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 Geometry 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, so the frontend can track its outcome without parsing log text. launch_geometry,
extrude_slot, and get_active_design are long-running and opt into this behavior;
shutdown_geometry is a regular transaction and does not raise a termination event.
Define the step fields
geometry_available tracks whether a Geometry instance is currently running, while
body_faces, body_edges, and active_design_name store the results of the Geometry
API calls.
version: str = "252"
body_faces: int = 0
body_edges: int = 0
active_design_name: str = ""
geometry_available: bool = False
Define the step model
The
launch_geometrylong-running transaction is decorated with@create_instanceto create an instance of the Geometry product manager,GeometryManager, while theshutdown_geometrytransaction is decorated with@instanceto reuse and shut down the existing instance. Unlike the other transactions,shutdown_geometryis not long-running and does not raise a termination event.The
extrude_slotandget_active_designtransactions are decorated with@instanceto indicate they operate on the existing product instance, and call the Geometry API to extrude a slot and to read the active design, respectively.launch_geometry,extrude_slot, andget_active_designpassenable_termination_event=Trueso the frontend can track their completion status.
@transaction(self=StepSpec(download=["version"]), enable_termination_event=True)
@create_instance("geometry_manager", GeometryManager)
@long_running
def launch_geometry(self, geometry_manager: GeometryManager) -> None:
"""Launch the Geometry instance and check its availability."""
self.transaction.raise_event(message="Initializing Geometry instance.", stream_name="geometry-output-stream")
try:
geometry_manager.initialize(version=self.version)
except Exception as e:
self.transaction.raise_event(
message=f"Geometry initialization failed: {e}",
stream_name="geometry-output-stream",
)
raise
self.transaction.raise_event(message="Geometry initialized.", stream_name="geometry-output-stream")
@transaction(self=StepSpec(upload=["body_faces", "body_edges"]), enable_termination_event=True)
@instance("geometry_manager")
@long_running
def extrude_slot(self, geometry_manager: GeometryManager) -> None:
"""Create a slot in the Geometry instance by extruding a sketch."""
self.transaction.raise_event(message="Starting extrude slot transaction.", stream_name="geometry-output-stream")
try:
# Create design on Geometry instance
design = geometry_manager.instance.create_design("ExtrudeSlot")
# Create a Sketch object and draw a slot
sketch = Sketch()
sketch.slot(Point2D([10, 10], UNITS.mm), Quantity(10, UNITS.mm), Quantity(5, UNITS.mm)) # type: ignore
# Extrude the sketch
body = design.extrude_sketch(name="MySlot", sketch=sketch, distance=Distance(50, UNITS.mm)) # type: ignore
if not body:
raise RuntimeError("Body was not created.")
self.body_faces = len(body.faces)
self.body_edges = len(body.edges)
except Exception as e:
self.transaction.raise_event(message=f"Extrude Slot failed: {e}", stream_name="geometry-output-stream")
raise
self.transaction.raise_event(message="Extrude Slot succeeded.", stream_name="geometry-output-stream")
@transaction(self=StepSpec(upload=["active_design_name"]), enable_termination_event=True)
@instance("geometry_manager")
@long_running
def get_active_design(self, geometry_manager: GeometryManager) -> None:
"""Get the name of the currently active design."""
self.transaction.raise_event(message="Getting active design name.", stream_name="geometry-output-stream")
try:
self.active_design_name = geometry_manager.instance.read_existing_design().name
self.transaction.raise_event(
message=f"Active design name: {self.active_design_name}.",
stream_name="geometry-output-stream",
)
except Exception as e:
self.transaction.raise_event(message=f"Get active design failed: {e}", stream_name="geometry-output-stream")
raise
self.transaction.raise_event(message="Get active design succeeded.", stream_name="geometry-output-stream")
@transaction(self=StepSpec(upload=["geometry_available"]))
@instance("geometry_manager")
def shutdown_geometry(self, geometry_manager: GeometryManager) -> None:
"""Close the Geometry instance."""
self.transaction.raise_event(message="Starting to shutdown the instance.", stream_name="geometry-output-stream")
try:
geometry_manager.shutdown()
self.geometry_available = False
except Exception as e:
self.transaction.raise_event(message=f"Geometry shutdown failed: {e}", stream_name="geometry-output-stream")
raise
self.transaction.raise_event(
message="Geometry instance shutdown complete.",
stream_name="geometry-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 console logs and the button states stay in sync with a
long-running transaction.
Define the step layout
The layout builds a controls card, with icon buttons to launch and shut down the instance and buttons to extrude a slot and get the active design, and a logs card.
Unlike the other instance management pages, the event listeners are created directly in the layout instead of through a separate
mount_event_listenerscallback:output-listenersubscribes to the free-form progress messages ongeometry-output-stream, whilelaunch-geometry-listener,extrude-slot-listener, andget-active-design-listenereach subscribe to the termination event of one long-running transaction.
def layout(project: ExamplesSolution) -> html.Div:
"""Layout for the Geometry Instance Manager page."""
step = project.steps.geometry_step
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-geometry-button",
size="xl",
color="#2790F1",
),
label="Launch Geometry",
position="top",
),
dmc.Tooltip(
dmc.ActionIcon(
DashIconify(icon="mdi:shutdown", width=30),
id="shutdown-geometry-button",
size="xl",
color="#2790F1",
disabled=True,
),
label="Shutdown Geometry",
position="top",
),
],
gap="md",
justify="center",
),
dmc.Space(h=20),
dmc.Divider(variant="solid"),
dmc.Space(h=20),
dmc.Stack(
[
dmc.Button(
"Extrude Slot",
id="extrude-slot-button",
variant="filled",
color="#2790F1",
leftSection=DashIconify(icon="mdi:design"),
disabled=True,
style={"width": "70%", "font-size": "15px"},
),
dmc.Button(
"Get Active Design",
id="get-active-design-button",
variant="filled",
color="#2790F1",
leftSection=DashIconify(icon="carbon:result"),
disabled=True,
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="geometry-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(
"Geometry 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 Geometry.\
Click the Launch action button to\
start the instance. A transaction method will start Geometry which can be used across all transaction\
methods of the solution. Run Geometry operations with the Extrude Slot and\
Get Active Design buttons. Close Geometry 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 at least Geometry 2025 R2 Service Pack 4 (25R2 SP4) to run.",
],
style={"whiteSpace": "pre-line"},
size="md",
),
title="Warning",
color="yellow",
),
dmc.Space(h=20),
dmc.Grid(
[
dmc.GridCol(
controls_card,
span=3,
),
dmc.GridCol(logs_container, span=9),
],
grow=True,
gutter="xs",
),
DashClient.create_event_listener( # pyright: ignore[reportUnknownMemberType]
step, id="output-listener", stream_name="geometry-output-stream"
),
DashClient.create_event_listener( # pyright: ignore[reportUnknownMemberType]
step, id="launch-geometry-listener", stream_name="launch-geometry"
),
DashClient.create_event_listener( # pyright: ignore[reportUnknownMemberType]
step, id="extrude-slot-listener", stream_name="extrude-slot"
),
DashClient.create_event_listener( # pyright: ignore[reportUnknownMemberType]
step, id="get-active-design-listener", stream_name="get-active-design"
),
html.Br(),
html.Br(),
],
style={"paddingLeft": "20px"},
)
Launch the instance
When the Launch Geometry button is clicked,
start_geometrystarts thelaunch_geometrylong-running transaction and shows a loading notification.When the
launch-geometry-listenerreceives the termination event,enable_geometry_extrude_slotturns it into a success or error notification and enables the shutdown and extrude slot buttons.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("launch-geometry-button", "disabled", allow_duplicate=True),
Output("launch-geometry-button", "loading", allow_duplicate=True),
Input("launch-geometry-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def start_geometry(n_clicks: int, project: ExamplesSolution) -> tuple[list[dict[str, Any]] | str, bool, bool]:
"""Initialize the Geometry instance."""
notification = no_update
disable_launch_button = no_update
loading_launch_button = no_update
if n_clicks:
step = project.steps.geometry_step
step.launch_geometry()
notification = [
dict(
title="Info",
id="start-geometry-notification",
action="show",
message="Starting Geometry instance... Please wait.",
autoClose=False,
loading=True,
color="orange",
withCloseButton=True,
)
]
disable_launch_button = True
loading_launch_button = True
return notification, disable_launch_button, loading_launch_button
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("launch-geometry-button", "disabled"),
Output("launch-geometry-button", "loading"),
Output("shutdown-geometry-button", "disabled", allow_duplicate=True),
Output("extrude-slot-button", "disabled", allow_duplicate=True),
Input("launch-geometry-listener", "message"),
prevent_initial_call=True,
)
def enable_geometry_extrude_slot(message: dict[str, Any]) -> tuple[list[dict[str, Any]] | str, bool, bool, bool, bool]:
"""Update the UI after Geometry is launched to enable extrude slot."""
notification = no_update
disable_launch_button = no_update
loading_launch_button = no_update
disable_shutdown_button = no_update
disable_extrude_slot_button = no_update
if message:
method_state = MethodState.model_validate_json(message["data"])
if method_state.status.value == "completed":
loading_launch_button = False
disable_shutdown_button = False
disable_extrude_slot_button = False
notification = [
dict(
title="Success",
id="start-geometry-notification",
action="update",
message="Geometry instance launched successfully!",
color="green",
autoClose=5000,
withCloseButton=True,
loading=False,
)
]
elif method_state.status.value == "failed":
disable_launch_button = False
loading_launch_button = False
notification = [
dict(
title="Error",
id="start-geometry-notification",
action="update",
message="Geometry initialization failed. Please check the logs.",
color="red",
autoClose=5000,
withCloseButton=True,
loading=False,
)
]
return (
notification,
disable_launch_button,
loading_launch_button,
disable_shutdown_button,
disable_extrude_slot_button,
)
Extrude a slot
When the Extrude Slot button is clicked,
extrude_slotstarts theextrude_slottransaction and shows a loading notification.When the
extrude-slot-listenerreceives the termination event,display_extrude_slot_notificationturns it into a success or error notification and enables the get active design button.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("extrude-slot-button", "disabled"),
Output("extrude-slot-button", "loading", allow_duplicate=True),
Output("shutdown-geometry-button", "disabled"),
Input("extrude-slot-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def extrude_slot(n_clicks: int, project: ExamplesSolution) -> tuple[list[dict[str, Any]] | str, bool, bool, bool]:
"""Extrude a slot in the Geometry instance."""
notification = no_update
disable_extrude_slot_button = no_update
loading_extrude_slot_button = no_update
disable_shutdown_button = no_update
if n_clicks:
step = project.steps.geometry_step
step.extrude_slot()
notification = [
dict(
title="Info",
id="extrude-slot-notification",
action="show",
message="Extruding slot... Check the logs for progress.",
autoClose=False,
loading=True,
color="orange",
withCloseButton=True,
)
]
disable_extrude_slot_button = True
loading_extrude_slot_button = True
disable_shutdown_button = True
return notification, disable_extrude_slot_button, loading_extrude_slot_button, disable_shutdown_button
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("extrude-slot-button", "disabled"),
Output("extrude-slot-button", "loading"),
Output("shutdown-geometry-button", "disabled"),
Output("get-active-design-button", "disabled", allow_duplicate=True),
Input("extrude-slot-listener", "message"),
prevent_initial_call=True,
)
def display_extrude_slot_notification(
message: dict[str, Any]
) -> tuple[list[dict[str, Any]] | str, bool, bool, bool, bool]:
"""Display extrude slot notification."""
notification = no_update
disable_extrude_slot_button = no_update
loading_extrude_slot_button = no_update
disable_shutdown_button = no_update
disable_get_active_design_button = no_update
if message:
loading_extrude_slot_button = False
disable_shutdown_button = False
method_state = MethodState.model_validate_json(message["data"])
if method_state.status.value == "completed":
disable_get_active_design_button = False
notification = [
dict(
title="Success",
id="extrude-slot-notification",
action="update",
message="Slot extruded successfully!",
autoClose=5000,
loading=False,
color="green",
withCloseButton=True,
)
]
elif method_state.status.value == "failed":
disable_extrude_slot_button = False
notification = [
dict(
title="Error",
id="extrude-slot-notification",
action="update",
message="Slot extrusion failed. Please check the logs.",
autoClose=5000,
loading=False,
color="red",
withCloseButton=True,
)
]
return (
notification,
disable_extrude_slot_button,
loading_extrude_slot_button,
disable_shutdown_button,
disable_get_active_design_button,
)
Get the active design
When the Get Active Design button is clicked,
get_active_designstarts theget_active_designtransaction and shows a loading notification.When the
get-active-design-listenerreceives the termination event,display_get_active_design_notificationturns it into a success or error notification.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("get-active-design-button", "disabled"),
Output("get-active-design-button", "loading", allow_duplicate=True),
Output("shutdown-geometry-button", "disabled"),
Input("get-active-design-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def get_active_design(n_clicks: int, project: ExamplesSolution) -> tuple[list[dict[str, Any]] | str, bool, bool, bool]:
"""Get the active design from the Geometry instance."""
notification = no_update
disable_get_active_design_button = no_update
loading_get_active_design_button = no_update
disable_shutdown_button = no_update
if n_clicks:
step = project.steps.geometry_step
step.get_active_design()
notification = [
dict(
title="Info",
id="get-active-design-notification",
action="show",
message="Getting active design... Check the logs for progress.",
autoClose=False,
loading=True,
color="orange",
withCloseButton=True,
)
]
disable_get_active_design_button = True
loading_get_active_design_button = True
disable_shutdown_button = True
return (
notification,
disable_get_active_design_button,
loading_get_active_design_button,
disable_shutdown_button,
)
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("get-active-design-button", "disabled"),
Output("get-active-design-button", "loading"),
Output("shutdown-geometry-button", "disabled"),
Input("get-active-design-listener", "message"),
prevent_initial_call=True,
)
def display_get_active_design_notification(
message: dict[str, Any]
) -> tuple[list[dict[str, Any]] | str, bool, bool, bool]:
"""Display get active design notification."""
notification = no_update
disable_get_active_design_button = no_update
loading_get_active_design_button = no_update
disable_shutdown_button = no_update
if message:
loading_get_active_design_button = False
disable_shutdown_button = False
method_state = MethodState.model_validate_json(message["data"])
if method_state.status.value == "completed":
notification = [
dict(
title="Success",
id="get-active-design-notification",
action="update",
message="Get active design completed successfully!",
autoClose=5000,
loading=False,
color="green",
withCloseButton=True,
)
]
elif method_state.status.value == "failed":
disable_get_active_design_button = False
notification = [
dict(
title="Error",
id="get-active-design-notification",
action="update",
message="Get active design failed. Please check the logs.",
autoClose=5000,
loading=False,
color="red",
withCloseButton=True,
)
]
return (
notification,
disable_get_active_design_button,
loading_get_active_design_button,
disable_shutdown_button,
)
Shut down the instance
When the Shutdown Geometry button is clicked,
shutdown_geometrycallsstep.shutdown_geometry()directly and shows the result as a notification. Becauseshutdown_geometryis not a long-running transaction, no event listener is needed: the callback updates the button states as soon as the call returns.
@callback(
Output("notification-container", "sendNotifications", allow_duplicate=True),
Output("launch-geometry-button", "disabled"),
Output("shutdown-geometry-button", "disabled"),
Output("extrude-slot-button", "disabled"),
Output("get-active-design-button", "disabled"),
Input("shutdown-geometry-button", "n_clicks"),
State("url", "pathname"),
prevent_initial_call=True,
)
def shutdown_geometry(
n_clicks: int, project: ExamplesSolution
) -> tuple[list[dict[str, Any]] | str, bool, bool, bool, bool]:
"""Shutdown the geometry instance."""
notification = no_update
disable_launch_button = no_update
disable_shutdown_button = no_update
disable_extrude_slot_button = no_update
disable_get_active_design_button = no_update
if n_clicks:
step = project.steps.geometry_step
try:
step.shutdown_geometry()
notification = [
dict(
title="Success",
id="shutdown-geometry-notification",
action="show",
message="Geometry instance shutdown successfully.",
autoClose=5000,
color="green",
withCloseButton=True,
)
]
disable_launch_button = False
disable_shutdown_button = True
disable_extrude_slot_button = True
disable_get_active_design_button = True
except Exception:
notification = [
dict(
title="Error",
id="shutdown-geometry-notification",
action="show",
message="Failed to shutdown geometry instance.",
autoClose=5000,
color="red",
withCloseButton=True,
)
]
return (
notification,
disable_launch_button,
disable_shutdown_button,
disable_extrude_slot_button,
disable_get_active_design_button,
)
Define the logs system.
display_geometry_outputappends every message received on theoutput-listenerstream to the console logs container, andclear_console_logsresets it.
@callback(
Output("geometry-console-logs", "children", allow_duplicate=True),
Input("output-listener", "message"),
State("geometry-console-logs", "children"),
prevent_initial_call=True,
)
def display_geometry_output(message: dict[str, Any], current_logs: str) -> str:
"""Display geometry 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("geometry-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.